Merge pull request #4883 from loopx-project/codex/repair-archived-bound-todo-closeout-clean

fix(quota): close archived Todo-bound host turns

Author: huangruiteng

docs/heartbeat-automation-prompt.md

@@ -298,7 +298,9 @@ whose capabilities are known when the automation is installed.
   ordinary work selection. The host must repair the prior closeout, rerun the
   same current Turn, and then continue an eligible successor. Recovery is
   idempotent and no-spend; receipts created before this explicit flag are not
-  retroactively treated as unsettled;
+  retroactively treated as unsettled. Exact-ID lifecycle reads include retained
+  `Completed Work Archive` rows, while ordinary Todo lists remain active-only;
+  an archived terminal fact closes recovery without rewriting archive history;
 - use `user_gate` only for an exact authority boundary such as approval to merge
   an aggregate branch into `main`, release, launch a benchmark, or perform a
   protected action;

loopx/control_plane/coordination/local_authority.py

@@ -303,23 +303,7 @@ def canonical_todo_summary_fields(
         for item in todos
         if item.get("archive_state") == "archive"
     }
-    # Native provider records have no Markdown address. Allocate display
-    # positions from stable provider order; never read legacy Markdown here.
-    todos = [
-        {
-            **item,
-            "schema_version": TODO_ITEM_SCHEMA_VERSION,
-            "source_section": (
-                "Completed Work Archive"
-                if item["archive_state"] == "archive"
-                else TODO_SECTION_HEADINGS[item["role"]]
-            ),
-            "index": index,
-        }
-        if item.get("schema_version") == TODO_DOMAIN_ITEM_SCHEMA_VERSION
-        else item
-        for index, item in enumerate(todos, 1)
-    ]
+    todos = canonical_todo_items(todos)
     # These are native authority decisions, not persisted Todo fields. Keep the
     # records visible while every summary/selection uses the same work guard.
     if goal_acceptance_contract and goal_acceptance_contract.get("enabled") is True:
@@ -366,3 +350,36 @@ def canonical_todo_summary_fields(
             if standing_authority:
                 fields["standing_decision_authority"] = standing_authority
     return fields
+
+
+def canonical_todo_items(todos: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Adapt every canonical Todo, including retained archive history."""
+
+    from ..todos.active_state_editing import TODO_SECTION_HEADINGS
+
+    # Native provider records have no Markdown address. Allocate display
+    # positions from stable provider order; never read legacy Markdown here.
+    return [
+        {
+            **item,
+            **(
+                {"schema_version": TODO_ITEM_SCHEMA_VERSION}
+                if item.get("schema_version") == TODO_DOMAIN_ITEM_SCHEMA_VERSION
+                else {}
+            ),
+            "source_section": "Completed Work Archive",
+            "index": index,
+        }
+        if item.get("archive_state") == "archive"
+        else (
+            {
+                **item,
+                "schema_version": TODO_ITEM_SCHEMA_VERSION,
+                "source_section": TODO_SECTION_HEADINGS[item["role"]],
+                "index": index,
+            }
+            if item.get("schema_version") == TODO_DOMAIN_ITEM_SCHEMA_VERSION
+            else item
+        )
+        for index, item in enumerate(todos, 1)
+    ]

loopx/control_plane/todos/goal_todo_projection.py

@@ -316,6 +316,60 @@ def todo_summaries_from_fields(
         uncapped_todo_count=uncapped_todo_count,
     )
 
+
+def exact_archived_todo_summaries(
+    *,
+    archived_items: list[dict[str, Any]],
+    source: str,
+    projection_fields: dict[str, Any] | None,
+    projection_overlay: dict[str, Any] | None,
+    rollout_events: list[dict[str, Any]],
+    roles: list[str],
+    status: str | None,
+    todo_id: str,
+    agent_id: str | None,
+    limit: int | None,
+) -> GoalTodoSummaries | None:
+    """Project one exact retained Todo without widening normal active lists."""
+
+    item = next(
+        (
+            dict(candidate)
+            for candidate in archived_items
+            if normalize_todo_id(candidate.get("todo_id")) == todo_id
+            and candidate.get("archive_state") == "archive"
+        ),
+        None,
+    )
+    if item is None:
+        return None
+    item_role = item.get("role")
+    if item_role not in {"user", "agent"} or item_role not in roles:
+        return None
+    summary = compact_todo_group(
+        [item],
+        source_section=str(item.get("source_section") or "Completed Work Archive"),
+        role=item_role,
+        include_empty_source=True,
+        resume_source_items=archived_items,
+        rollout_events=rollout_events,
+        item_limit=None,
+    )
+    if summary is None:
+        return None
+    return todo_summaries_from_fields(
+        fields={f"{item_role}_todos": summary},
+        source=source,
+        projection_fields=projection_fields,
+        projection_overlay=projection_overlay,
+        rollout_events=rollout_events,
+        roles=roles,
+        status=status,
+        todo_id=todo_id,
+        agent_id=agent_id,
+        limit=limit,
+    )
+
 def project_goal_todo_items(
     goal: dict[str, Any] | None,
     *,
@@ -345,6 +399,7 @@ def project_goal_todo_items(
 __all__ = [
     "GoalTodoSummaries",
     "empty_todo_summary",
+    "exact_archived_todo_summaries",
     "filtered_todo_summary",
     "goal_todo_summaries",
     "merge_todo_projection_fields",

loopx/todos.py

@@ -89,9 +89,11 @@
     todo_list_projection_contract,
 )
 from .control_plane.todos.goal_todo_projection import (
+    exact_archived_todo_summaries,
     goal_todo_summaries,
     todo_summaries_from_fields,
 )
+from .control_plane.todos.active_state_todo_parser import parse_todo_source
 from .control_plane.todos import monitor_metadata as todo_monitor_metadata
 from .control_plane.todos.mutation_authority import authorize_todo_lifecycle_mutation, todo_update_authority_action
 from .control_plane.todos.succession_warning import build_open_parent_successor_advisory
@@ -117,6 +119,7 @@
 )
 from .control_plane.coordination.legacy_writer_fence import legacy_todo_write_transaction
 from .control_plane.coordination.local_authority import (
+    canonical_todo_items,
     canonical_todo_summary_fields,
     claim_canonical_todo_if_promoted,
     local_authority_is_promoted,
@@ -234,9 +237,10 @@ def list_goal_todos(
     else:
         if not resolved_state_file.exists():
             raise ValueError(f"active state file does not exist: {resolved_state_file}")
+        state_text = resolved_state_file.read_text(encoding="utf-8")
         projected = goal_todo_summaries(
             goal,
-            state_text=resolved_state_file.read_text(encoding="utf-8"),
+            state_text=state_text,
             state_path=resolved_state_file,
             rollout_events=rollout_events,
             roles=roles,
@@ -245,6 +249,33 @@ def list_goal_todos(
             agent_id=normalized_agent_id,
             limit=limit,
         )
+    if normalized_todo_id and not projected.todos:
+        if canonical_read is not None:
+            archived_items = [
+                item
+                for item in canonical_todo_items(canonical_read["todos"])
+                if item.get("archive_state") == "archive"
+            ]
+        else:
+            _active_items, archived_items, _source_sections = parse_todo_source(
+                state_text,
+                goal=goal,
+                state_path=resolved_state_file,
+            )
+        archived_projection = exact_archived_todo_summaries(
+            archived_items=archived_items,
+            source=projected.source,
+            projection_fields=projected.projection_fields,
+            projection_overlay=projected.projection_overlay,
+            rollout_events=rollout_events,
+            roles=roles,
+            status=status,
+            todo_id=normalized_todo_id,
+            agent_id=normalized_agent_id,
+            limit=limit,
+        )
+        if archived_projection is not None:
+            projected = archived_projection
     source = projected.source
     projection_fields = projected.projection_fields
     projection_overlay = projected.projection_overlay

tests/control_plane/test_quota_settlement_cli.py

@@ -1601,6 +1601,132 @@ def test_prior_host_closeout_survives_hidden_todo_lifecycle(
     assert resumed["quota"]["spent_slots"] == prior["quota"]["spent_slots"]
 
 
+@pytest.mark.parametrize("provider", ["legacy", "file", "sqlite"])
+def test_prior_host_closeout_reads_archived_exact_todo(
+    tmp_path: Path,
+    provider: str,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from canonical_authority_fixture import (
+        initialize_canonical_authority,
+        isolate_sqlite_runtime,
+    )
+    from loopx.control_plane.coordination.runtime_shadow import (
+        build_todo_runtime_shadow_projection,
+    )
+
+    if provider == "sqlite":
+        isolate_sqlite_runtime(tmp_path, monkeypatch)
+    project, runtime, registry_path = _write_fixture(tmp_path)
+    guard = (
+        "quota",
+        "should-run",
+        "--codex-app",
+        "--goal-id",
+        GOAL_ID,
+        "--agent-id",
+        AGENT_ID,
+        "--scan-path",
+        str(project),
+    )
+    prior_turn_id = "turn-archived-closeout-prior"
+    rc, prior = _run_cli(
+        registry_path,
+        runtime,
+        *guard,
+        "--turn-instance-id",
+        prior_turn_id,
+        "--todo-id",
+        TODO_ID,
+    )
+    assert rc == 0, prior
+    assert prior["heartbeat_receipt"]["closeout_required"] is True
+
+    state = project / ".codex" / "goals" / GOAL_ID / "ACTIVE_GOAL_STATE.md"
+    if provider != "legacy":
+        rc, listed = _run_cli(
+            registry_path, runtime, "todo", "list", "--goal-id", GOAL_ID
+        )
+        assert rc == 0, listed
+        projection = build_todo_runtime_shadow_projection(
+            goal_id=GOAL_ID,
+            handoff_mode="soft_claim",
+            todos=listed["todos"],
+        )
+        initialize_canonical_authority(
+            runtime, GOAL_ID, projection, state_path=state, provider=provider
+        )
+
+    rc, complete = _run_cli(
+        registry_path,
+        runtime,
+        "todo",
+        "complete",
+        "--goal-id",
+        GOAL_ID,
+        "--agent-id",
+        AGENT_ID,
+        "--todo-id",
+        TODO_ID,
+        "--turn-instance-id",
+        prior_turn_id,
+        "--claimed-by",
+        AGENT_ID,
+        "--evidence",
+        "archived closeout lifecycle validated",
+        "--next-agent-todo",
+        "Continue after archived closeout validation.",
+        "--next-claimed-by",
+        AGENT_ID,
+        "--next-action-kind",
+        "implement",
+    )
+    assert rc == 0, complete
+    rc, archived = _run_cli(
+        registry_path,
+        runtime,
+        "todo",
+        "archive-completed",
+        "--goal-id",
+        GOAL_ID,
+        "--max-active-done",
+        "0",
+        "--execute",
+    )
+    assert rc == 0, archived
+    assert archived["moved_count"] == 1
+
+    rc, active = _run_cli(
+        registry_path, runtime, "todo", "list", "--goal-id", GOAL_ID
+    )
+    assert rc == 0, active
+    assert TODO_ID not in {item["todo_id"] for item in active["todos"]}
+    rc, exact = _run_cli(
+        registry_path,
+        runtime,
+        "todo",
+        "list",
+        "--goal-id",
+        GOAL_ID,
+        "--role",
+        "agent",
+        "--todo-id",
+        TODO_ID,
+    )
+    assert rc == 0, exact
+    assert exact["todo"]["todo_id"] == TODO_ID
+    assert exact["todo"]["status"] == "done"
+    assert exact["todo"]["archive_state"] == "archive"
+    assert exact["todo"]["source_section"] == "Completed Work Archive"
+
+    rc, observed = _run_cli(registry_path, runtime, *guard, "--begin-turn")
+    assert rc == 0, observed
+    assert observed["effective_action"] != "unsettled_host_turn_recovery", observed.get(
+        "unsettled_host_turn_recovery"
+    )
+    assert observed["quota"]["spent_slots"] == prior["quota"]["spent_slots"]
+
+
 def test_standard_codex_app_settlement_is_receipted_and_idempotent(
     tmp_path: Path,
 ) -> None: