Refactor: split SDK example flow and related metadata helpers

sdk/python/examples/basic.py

@@ -12,37 +12,101 @@
 
 import os
 import sys
+from dataclasses import dataclass
 from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
 from todo2code import T2CClient, T2CError  # noqa: E402
 
+DEFAULT_A2A_URL = "http://localhost:8787"
+DEFAULT_EXAMPLE_ROOT = "examples/backend"
+DEFAULT_COMPARE_BASE = "origin/main"
+TODO_PATCH_PATH = ".intent-sdk/python/TODO.patch"
+TODO_AUDIT_PATH = ".intent-sdk/python/TODO.patch.json"
+TODO_RECEIPT_PATH = ".intent-sdk/python/TODO.patch.receipt.json"
 
-def main() -> int:
-    base_url = os.environ.get("T2C_A2A_URL", "http://localhost:8787")
-    token = os.environ.get("T2C_A2A_TOKEN")
-    root = os.environ.get("T2C_EXAMPLE_ROOT", "examples/backend")
 
-    client = T2CClient(base_url, token=token)
+@dataclass(frozen=True)
+class ExampleContext:
+    base_url: str
+    token: str | None
+    root: str
+    compare_base: str
+    compare_workspace: bool
+
+
+@dataclass(frozen=True)
+class ExtractionArtifacts:
+    graph: object
+    report: object
+    record_count: int
+
+
+@dataclass(frozen=True)
+class ProposalArtifacts:
+    new_proposal_ids: tuple[str, ...]
+    duplicate_proposal_ids: tuple[str, ...]
+    patch_hash: str
+
+
+def main() -> int:
+    context = read_example_context()
+    client = T2CClient(context.base_url, token=context.token)
 
     print("health:", client.health())
+    run_flow(client, context)
+    print("OK")
+    return 0
+
+
+def read_example_context() -> ExampleContext:
+    return ExampleContext(
+        base_url=os.environ.get("T2C_A2A_URL", DEFAULT_A2A_URL),
+        token=os.environ.get("T2C_A2A_TOKEN"),
+        root=os.environ.get("T2C_EXAMPLE_ROOT", DEFAULT_EXAMPLE_ROOT),
+        compare_base=os.environ.get("T2C_COMPARE_BASE", DEFAULT_COMPARE_BASE),
+        compare_workspace=os.environ.get("T2C_COMPARE_WORKSPACE", "0") == "1",
+    )
+
+
+def run_flow(client: T2CClient, context: ExampleContext) -> None:
     card = client.agent_card()
     print("agent skills:", ", ".join(skill["id"] for skill in card.get("skills", ())))
 
-    # 1. Deterministic extraction -> graph -> diagnostics.
+    extraction = run_extraction_flow(client, context.root)
+    print(f"extracted {extraction.record_count} records from {context.root}")
+
+    proposal = run_proposal_flow(
+        client,
+        context.root,
+        extraction.graph,
+        extraction.report,
+    )
+    print("proposal ids:", ",".join(proposal.new_proposal_ids) or "-")
+    print("duplicate ids:", ",".join(proposal.duplicate_proposal_ids) or "-")
+    print("patch fingerprint:", proposal.patch_hash[:16])
+
+    run_reality_and_diff(client, extraction.graph, extraction.report, context.root)
+    run_optional_workspace_comparison(
+        client,
+        context.root,
+        context.compare_base,
+        context.compare_workspace,
+    )
+
+
+def run_extraction_flow(client: T2CClient, root: str) -> ExtractionArtifacts:
     nl = client.extract_nl_result("task.md", root, nl_mode="deterministic")
-    if nl.audit is None or nl.audit.get("status") != "succeeded" or nl.audit.get("effectiveMode") != "deterministic":
-        raise RuntimeError(f"unexpected NL audit: {nl.audit}")
+    assert_audit_success(nl.audit, "NL", check_mode=True)
     print("NL audit:", nl.audit.get("status"), nl.audit.get("effectiveMode"))
+
     ast_records = client.extract_ast(root)
     markdown = client.extract_markdown_result(root, markdown_mode="deterministic")
-    if markdown.audit is None or markdown.audit.get("status") != "succeeded":
-        raise RuntimeError(f"unexpected Markdown audit: {markdown.audit}")
+    assert_audit_success(markdown.audit, "Markdown")
     print("markdown audit:", markdown.audit.get("status"), markdown.audit.get("effectiveMode"))
-    records = [*nl.records, *ast_records, *markdown.records]
-    print(f"extracted {len(records)} records from {root}")
 
+    records = [*nl.records, *ast_records, *markdown.records]
     graph = client.link(records)
     print("graph fingerprint:", graph.fingerprint[:16])
     print("records by source:", graph.stats.get("bySource"))
@@ -52,39 +116,91 @@ def main() -> int:
     for diagnostic in report.diagnostics[:3]:
         print(f"  - [{diagnostic.severity}] {diagnostic.code}: {diagnostic.title}")
 
-    # 2. Audited propose -> review -> approved no-op apply without secrets.
-    synthesis = client.propose_todo({"root": root, "graph": graph.raw, "diagnostics": report.raw, "mode": "prefer-llm"})
+    return ExtractionArtifacts(
+        graph=graph,
+        report=report,
+        record_count=len(records),
+    )
+
+
+def assert_audit_success(
+    audit: dict[str, object] | None,
+    label: str,
+    check_mode: bool = False,
+) -> None:
+    if audit is None or audit.get("status") != "succeeded":
+        raise RuntimeError(f"unexpected {label} audit: {audit}")
+    if check_mode and audit.get("effectiveMode") != "deterministic":
+        raise RuntimeError(f"unexpected {label} mode: {audit.get('effectiveMode')} for {label}")
+
+
+def run_proposal_flow(
+    client: T2CClient,
+    root: str,
+    graph: object,
+    report: object,
+) -> ProposalArtifacts:
+    synthesis = client.propose_todo({
+        "root": root,
+        "graph": graph.raw,
+        "diagnostics": report.raw,
+        "mode": "prefer-llm",
+    })
     validation = synthesis.get("validation", {})
+
     rendered = client.render_todo({
-        "root": root, "graph": graph.raw, "diagnostics": report.raw, "synthesis": synthesis, "todo": "TODO.md",
-        "patch": ".intent-sdk/python/TODO.patch", "audit": ".intent-sdk/python/TODO.patch.json",
+        "root": root,
+        "graph": graph.raw,
+        "diagnostics": report.raw,
+        "synthesis": synthesis,
+        "todo": "TODO.md",
+        "patch": TODO_PATCH_PATH,
+        "audit": TODO_AUDIT_PATH,
     })
+
     patch_hash = rendered["artifact"]["renderedPatchHash"]
     client.apply_todo({
-        "root": root, "todo": "TODO.md", "patch": ".intent-sdk/python/TODO.patch",
-        "audit": ".intent-sdk/python/TODO.patch.json", "receipt": ".intent-sdk/python/TODO.patch.receipt.json",
-        "actor": "sdk-python", "approvalHash": patch_hash,
+        "root": root,
+        "todo": "TODO.md",
+        "patch": TODO_PATCH_PATH,
+        "audit": TODO_AUDIT_PATH,
+        "receipt": TODO_RECEIPT_PATH,
+        "actor": "sdk-python",
+        "approvalHash": patch_hash,
     })
-    print("proposal ids:", ",".join(validation.get("newProposalIds", ())) or "-")
-    print("duplicate ids:", ",".join(validation.get("duplicateProposalIds", ())) or "-")
-    print("patch fingerprint:", patch_hash[:16])
 
-    # 3. Intent-vs-reality view.
+    return ProposalArtifacts(
+        new_proposal_ids=tuple(validation.get("newProposalIds", ())),
+        duplicate_proposal_ids=tuple(validation.get("duplicateProposalIds", ())),
+        patch_hash=patch_hash,
+    )
+
+
+def run_reality_and_diff(
+    client: T2CClient,
+    graph: object,
+    report: object,
+    root: str,
+) -> None:
     reality = client.reality(graph, report, gapsOnly=True, includeSvg=True)
     print("reality svg bytes:", len(reality.get("svg", "")))
     print(reality["markdown"].split("\n")[4])
 
-    # 4. Git diff rendered as SVG.
     git_diff = client.diff_git(root=root, revision="HEAD", includeSvg=True)
     print("git diff files:", len(git_diff.get("diffs", ())))
 
-    # 5. Optional origin/main -> local filesystem Intent comparison.
-    if os.environ.get("T2C_COMPARE_WORKSPACE") == "1":
-        comparison = client.compare_workspace(root=root, base=os.environ.get("T2C_COMPARE_BASE", "origin/main"))
-        print("workspace trend:", comparison.get("trend", {}).get("direction"))
 
-    print("OK")
-    return 0
+def run_optional_workspace_comparison(
+    client: T2CClient,
+    root: str,
+    compare_base: str,
+    compare_workspace: bool,
+) -> None:
+    if not compare_workspace:
+        return
+
+    comparison = client.compare_workspace(root=root, base=compare_base)
+    print("workspace trend:", comparison.get("trend", {}).get("direction"))
 
 
 if __name__ == "__main__":

src/interfaces/intake_cli.py

@@ -23,6 +23,7 @@
     5: "idempotencyKey", 6: "authenticatedPrincipal", 7: "expectedVersion",
     8: "timestamp", 9: "payloadHash", 10: "payload",
 }
+SKIP_UNKNOWN_FIELD = object()
 
 
 def _varint(value: int) -> bytes:
@@ -81,32 +82,55 @@ def decode_envelope(data: bytes) -> dict[str, Any]:
     offset = 0
     while offset < len(data):
         start = offset
-        tag, offset = _read_varint(data, offset)
-        number, wire = tag >> 3, tag & 7
-        if wire == 0:
-            value, offset = _read_varint(data, offset)
-            if number == 7:
-                output["expectedVersion"] = value
-            else:
-                unknown.append(base64.b64encode(data[start:offset]).decode())
-        elif wire == 2:
-            length, offset = _read_varint(data, offset)
-            end = offset + length
-            if end > len(data):
-                raise ValueError("truncated length-delimited field")
-            raw = data[offset:end]
-            offset = end
-            if number in FIELDS and number != 7:
-                output[FIELDS[number]] = json.loads(raw) if number == 10 else raw.decode()
-            else:
-                unknown.append(base64.b64encode(data[start:offset]).decode())
-        else:
-            raise ValueError(f"unsupported wire type {wire}")
+        number, wire, value_start, offset = read_field_metadata(data, offset)
+        value = read_field_value(data, number, wire, value_start, offset)
+        if value is SKIP_UNKNOWN_FIELD:
+            unknown.append(base64.b64encode(data[start:offset]).decode())
+            continue
+        if value is not None:
+            output[value[0]] = value[1]
     if unknown:
         output["unknownFields"] = unknown
     return output
 
 
+def read_field_metadata(data: bytes, offset: int) -> tuple[int, int, int, int]:
+    tag, offset = _read_varint(data, offset)
+    number, wire = tag >> 3, tag & 7
+    if wire == 0:
+        value_start = offset
+        _, offset = _read_varint(data, offset)
+        return number, wire, value_start, offset
+    if wire == 2:
+        length, offset = _read_varint(data, offset)
+        end = offset + length
+        if end > len(data):
+            raise ValueError("truncated length-delimited field")
+        return number, wire, offset, end
+    raise ValueError(f"unsupported wire type {wire}")
+
+
+def read_field_value(
+    data: bytes,
+    number: int,
+    wire: int,
+    value_start: int,
+    value_end: int,
+) -> tuple[str, Any] | object | None:
+    if value_start > len(data) or value_end > len(data) or value_end < value_start:
+        raise ValueError("invalid field payload")
+    if wire == 0 and number == 7:
+        value, _ = _read_varint(data, value_start)
+        return ("expectedVersion", value)
+    if wire == 0:
+        return SKIP_UNKNOWN_FIELD
+    if wire == 2 and number in FIELDS and number != 7:
+        payload = data[value_start:value_end]
+        field_value = json.loads(payload) if number == 10 else payload.decode()
+        return (FIELDS[number], field_value)
+    return SKIP_UNKNOWN_FIELD
+
+
 def execute(args: argparse.Namespace) -> int:
     repository = pathlib.Path(args.repository).resolve()
     cli = repository / "dist" / "src" / "cli.js"
@@ -119,35 +143,58 @@ def execute(args: argparse.Namespace) -> int:
     return subprocess.run(command, cwd=repository, check=False).returncode
 
 
-def main() -> int:
+def run_command(args: argparse.Namespace) -> int:
+    return execute(args)
+
+
+def run_encode(args: argparse.Namespace) -> int:
+    envelope = json.loads(pathlib.Path(args.input).read_text(encoding="utf-8"))
+    pathlib.Path(args.output).write_bytes(encode_envelope(envelope))
+    return 0
+
+
+def run_decode(args: argparse.Namespace) -> int:
+    envelope = decode_envelope(pathlib.Path(args.input).read_bytes())
+    pathlib.Path(args.output).write_text(json.dumps(envelope, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
+    return 0
+
+
+def build_parser() -> argparse.ArgumentParser:
     parser = argparse.ArgumentParser(description="Dependency-free todo2code governed-intake CLI")
     sub = parser.add_subparsers(dest="mode", required=True)
     for operation in ("command", "query"):
-        run = sub.add_parser(operation)
-        run.add_argument("input")
-        run.add_argument("--repository", default=".")
-        run.add_argument("--root", default=".")
-        run.add_argument("--project-dir", default="project")
-        run.add_argument("--protobuf", action="store_true")
+        add_command_parser(sub, operation)
+
     encode = sub.add_parser("encode")
     encode.add_argument("input")
     encode.add_argument("output")
+    encode.set_defaults(handler=run_encode)
+
     decode = sub.add_parser("decode")
     decode.add_argument("input")
     decode.add_argument("output")
+    decode.set_defaults(handler=run_decode)
+
+    return parser
+
+
+def add_command_parser(sub: argparse._SubParsersAction, operation: str) -> None:
+    run = sub.add_parser(operation)
+    run.add_argument("input")
+    run.add_argument("--repository", default=".")
+    run.add_argument("--root", default=".")
+    run.add_argument("--project-dir", default="project")
+    run.add_argument("--protobuf", action="store_true")
+    run.set_defaults(handler=run_command, operation=operation)
+
+
+def main() -> int:
+    parser = build_parser()
     args = parser.parse_args()
     try:
-        if args.mode in ("command", "query"):
-            args.operation = args.mode
-            return execute(args)
-        if args.mode == "encode":
-            envelope = json.loads(pathlib.Path(args.input).read_text(encoding="utf-8"))
-            pathlib.Path(args.output).write_bytes(encode_envelope(envelope))
-        else:
-            envelope = decode_envelope(pathlib.Path(args.input).read_bytes())
-            pathlib.Path(args.output).write_text(json.dumps(envelope, indent=2, ensure_ascii=False) + "\n", encoding="utf-8")
-        return 0
-    except (OSError, ValueError, json.JSONDecodeError) as error:
+        result = args.handler(args)
+        return int(result)
+    except (OSError, ValueError, json.JSONDecodeError, UnicodeError) as error:
         print(f"T2C-INTAKE-INVALID-WIRE: {error}", file=sys.stderr)
         return 2