Use compact JSON receipts for managed Todo claims

docs/project-agent-todo-contract.md

@@ -556,6 +556,15 @@ loopx todo claim \
   --agent-id codex-main-control
 ```
 
+The command shown by quota for an ordinary managed claim uses `--thin`.
+The JSON receipt keeps the claim operation ID, next actions and projection
+status. When the display is current and has omitted fields, read the per-Todo
+details with `projection_outbox.detail_command`, a preview pinned to the
+reported provider revision. Run it before the revision advances; if it has advanced,
+read the current Todo or projection rather than claiming again. Explicit
+claims without `--thin` keep their full output. Pending or failed projections
+still return their recovery diagnostics inline.
+
 Old projects that do not yet have `coordination.registered_agents` are
 intentionally blocked when an agent tries to claim work. The CLI error includes
 the `register-agent --agent-id <agent-id> --execute` command so the agent or

loopx/cli_commands/todo.py

@@ -444,6 +444,38 @@ def _render_todo_receipt(payload: dict[str, object]) -> str:
     return "\n".join(lines)
 
 
+def _thin_claim_projection(payload: dict[str, object], route: Sequence[str]) -> None:
+    """Omit per-Todo display diagnostics from a successfully delivered claim."""
+    outbox = payload.get("projection_outbox")
+    if (
+        not isinstance(outbox, dict)
+        or outbox.get("schema_version") != "loopx_todo_projection_delivery_v0"
+        or outbox.get("status") not in ("delivered", "current")
+        or outbox.get("next_action") != "finish"
+        or outbox.get("parse_render_parity") is not True
+        or outbox.get("narrative_preserved") is not True
+        or outbox.get("legacy_fallback_used") is not False
+    ):
+        return
+    rows = outbox.get("omitted_active_records")
+    revision = outbox.get("provider_revision")
+    goal_id = payload.get("goal_id")
+    if (
+        not isinstance(rows, list) or not rows or outbox.get("active_fields_omitted") != len(rows)
+        or not isinstance(revision, str) or not revision or not isinstance(goal_id, str) or not goal_id
+    ):
+        return
+    command = [*route, "todo", "project-markdown", "--goal-id", goal_id, "--provider-revision", revision]
+    for field in ("project", "state_file"):
+        path = outbox.get(field)
+        if isinstance(path, str) and path:
+            command.extend(["--" + field.replace("_", "-"), path])
+    payload["projection_outbox"] = {
+        **{key: value for key, value in outbox.items() if key != "omitted_active_records"},
+        "detail_command": shlex.join(command),
+    }
+
+
 def handle_todo_command(
     args: argparse.Namespace,
     *,
@@ -674,6 +706,8 @@ def handle_todo_command(
                 # counters or a second inline copy of the native receipt.
                 for internal in ("lease", "lease_changed", "lease_idempotent", "original_receipt"):
                     payload.pop(internal, None)
+                if args.todo_thin:
+                    _thin_claim_projection(payload, route)
         elif args.todo_command == "update":
             validate_todo_update_options(args)
             update_operation = partial(

loopx/cli_commands/todo_argument_validation.py

@@ -153,7 +153,8 @@
     "plan": frozenset({"text", "agent_id", "project"}),
     "project-markdown": frozenset({"state_file", "execute", "provider_revision", "project"}),
     "claim": frozenset({"role", "todo_id", "claimed_by", "agent_id", "state_file",
-                        "claim_operation_id", "task_lease_idempotency_key", "task_lease_expected_version", "project", "dry_run"}),
+                        "claim_operation_id", "task_lease_idempotency_key", "task_lease_expected_version",
+                        "project", "dry_run", "todo_thin"}),
     "update": frozenset(_TODO_UPDATE_MUTABLE_FIELDS) | frozenset({
         "role", "todo_id", "agent_id", "state_file", "project", "dry_run", "authority_reason",
         "claim_operation_id", "update_operation_id", "update_expected_provider_revision",
@@ -430,7 +431,7 @@ def validate_todo_claim_options(args: argparse.Namespace) -> None:
         raise TodoClaimArgumentError(
             "todo claim only accepts --todo-id, --claimed-by, --agent-id, optional --role, "
             "--claim-operation-id, --task-lease-idempotency-key, "
-            "--task-lease-expected-version, --project, --state-file, and --dry-run; unsupported: "
+            "--task-lease-expected-version, --project, --state-file, --dry-run, and --thin; unsupported: "
             + ", ".join(unsupported)
         )
     if args.task_lease_expected_version is not None and not args.task_lease_idempotency_key:
@@ -758,8 +759,8 @@ def validate_shared_todo_options(args: argparse.Namespace) -> None:
         raise ValueError(
             "--limit is supported only by todo list"
         )
-    if args.todo_thin and args.todo_command != "list":
-        raise ValueError("--thin is supported only by todo list")
+    if args.todo_thin and args.todo_command not in {"list", "claim"}:
+        raise ValueError("--thin is supported only by todo list and todo claim")
     if getattr(args, "todo_summary", False):
         if args.todo_command != "list":
             raise ValueError("--summary is supported only by todo list")

loopx/cli_commands/todo_registration.py

@@ -463,7 +463,9 @@ def register_todo_command(
         help=(
             "For todo list, return the explicit field-only projection and omit "
             "detail lanes; returns at most two items per role, and --limit can "
-            "lower but not expand that bound."
+            "lower but not expand that bound. In a managed todo claim JSON receipt, omit "
+            "per-Todo display diagnostics from a successful projection and "
+            "return a read-only detail command."
         ),
     )
     todo_parser.add_argument(

loopx/control_plane/quota/execution_preparation.py

@@ -91,7 +91,9 @@ def project_selected_todo_execution_preparation(
         raise TypeError("typed execution preparation returned an invalid projection")
     claim = result.get("claim")
     if isinstance(claim, dict) and isinstance(claim.get("cli_args"), list):
-        claim["command_line"] = shlex.join(["loopx", *map(str, claim["cli_args"])])
+        # The displayed command selects a smaller receipt. Hosts still receive
+        # the exact mutation argv in cli_args, including on historical captures.
+        claim["command_line"] = shlex.join(["loopx", *map(str, claim["cli_args"]), "--thin"])
     return result
 
 

tests/control_plane/test_quota_execution_preparation.py

@@ -211,6 +211,75 @@ def test_projected_claim_prepares_the_same_todo_for_a_fresh_guard(
     assert after["normal_delivery_allowed"] is True
 
 
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("projection_available", [True, False])
+def test_model_claim_keeps_receipt_without_unrelated_projection_details(
+    tmp_path, monkeypatch, provider, projection_available,
+):
+    other_todos = [
+        {**_todo(), "todo_id": f"todo_other_{index:03d}", "index": index + 2,
+         "claimed_by": "agent-b", "text": f"[P1] Unrelated work {index}",
+         "note": ("Supporting details for unrelated work. " * 30).strip()}
+        for index in range(72)
+    ]
+    selected = {**_todo(), "text": "[P1] Deliver one validated synthetic change"}
+    before = _run(tmp_path, monkeypatch, provider, [], todos=[selected, *other_todos],
+                  goal_suffix="\n## Agent Todo\n")
+    preparation = before["execution_preparation"]
+    assert preparation["state"] == "claim_required"
+    action = shlex.split(before["interaction_contract"]["cli_channel"]["next_cli_actions"][0])
+    assert action[0] == "loopx"
+    assert "--thin" not in preparation["claim"]["cli_args"]  # Exact Host action.
+    operation = ["--claim-operation-id", "synthetic-thin-claim"]
+
+    def invoke(*args):
+        rc, result = run_json_cli_result(
+            *args, registry_path=tmp_path / "registry.json", runtime_root=tmp_path / "runtime",
+        )
+        assert rc == 0 and result["ok"] is True, result
+        return result
+
+    state = tmp_path / "state.md"
+    if not projection_available:
+        state.unlink()
+        state.mkdir()  # The authority is available; only the display write fails.
+    claim = invoke(*action[1:], *operation)
+    assert claim["status"] == "applied" and claim["claim_operation_id"] == operation[1]
+    assert claim["todo_id"] == TODO and claim["claimed_by"] == AGENT
+    initial = invoke("task-lease", "inspect", "--goal-id", GOAL, "--todo-id", TODO)
+    assert initial["active"] is True and initial["lease"]["owner"] == AGENT
+    outbox = claim["projection_outbox"]
+    if not projection_available:
+        assert claim["projection_delivery"] == outbox["status"] == "pending"
+        assert outbox["reason_code"] and outbox["recommended_action"]
+        assert outbox["retryable"] is True and outbox["retry_business_mutation"] is False
+        assert "detail_command" not in outbox
+        return
+
+    assert claim["projection_delivery"] == outbox["status"] == "delivered", outbox
+    assert claim["next_cli_actions"] and outbox["next_action"] == "finish"
+    if provider == "sqlite":
+        assert outbox["active_fields_omitted"] == len(other_todos)
+        assert "omitted_active_records" not in outbox
+        before_read = state.read_bytes()
+        diagnostic_args = shlex.split(outbox["detail_command"])
+        detail = invoke(*diagnostic_args[diagnostic_args.index("todo"):])
+        assert detail["dry_run"] is True and detail["provider_revision"] == outbox["provider_revision"]
+        assert state.read_bytes() == before_read
+        rows = detail["omitted_active_records"]
+        assert {row["todo_id"] for row in rows} == {row["todo_id"] for row in other_todos}
+    else:
+        rows = []
+        assert outbox["active_fields_omitted"] == 0 and outbox["omitted_active_records"] == rows
+        assert "detail_command" not in outbox
+    assert "--thin" in action
+    replay = invoke(*preparation["claim"]["cli_args"], *operation)
+    assert replay["claim_operation_id"] == claim["claim_operation_id"]
+    assert replay["projection_outbox"]["omitted_active_records"] == rows
+    current = invoke("task-lease", "inspect", "--goal-id", GOAL, "--todo-id", TODO)
+    assert current["lease"] == initial["lease"]
+
+
 @pytest.mark.parametrize("provider,scenario,location,ambient", [
     (provider, scenario, "directory", "preparation-session")
     for provider in ("file", "sqlite")
@@ -710,7 +779,7 @@ def read(*argv):
         read_model_view(before, "claim", deferred=True)
 
     claim_code, claim = run_json_cli_result(
-        *before["execution_preparation"]["claim"]["cli_args"],
+        *shlex.split(contract["cli_channel"]["next_cli_actions"][0])[1:],
         registry_path=tmp_path / "registry.json",
         runtime_root=tmp_path / "runtime",
     )

tests/test_cli_argument_diagnostics.py

@@ -557,7 +557,7 @@ def test_monitor_change_scope_requires_material_transition() -> None:
             ],
             "todo claim only accepts --todo-id, --claimed-by, --agent-id, optional --role, "
             "--claim-operation-id, --task-lease-idempotency-key, "
-            "--task-lease-expected-version, --project, --state-file, and --dry-run; unsupported: "
+            "--task-lease-expected-version, --project, --state-file, --dry-run, and --thin; unsupported: "
             "--decision-outcome, --next-agent-todo",
         ),
     ],