fix(quota): preserve the active Turn after lifecycle closeout

loopx/control_plane/work_items/autonomous_replan_obligation.py

@@ -82,6 +82,7 @@ def project_todo_lifecycle_settlement_reentry(
     lifecycle_actor_args: str,
     scoped_cli_args: str,
     runtime_root: str | None = None,
+    quota_guard_action: str | None = None,
 ) -> dict[str, Any] | None:
     """Adapt a lifecycle obligation to the TS-owned reentry projection."""
 
@@ -106,6 +107,7 @@ def project_todo_lifecycle_settlement_reentry(
         triggers=triggers,
         lifecycle_actor_args=shlex.split(lifecycle_actor_args),
         quota_scoped_args=shlex.split(scoped_cli_args),
+        quota_guard_argv=shlex.split(quota_guard_action) if quota_guard_action else None,
     )
 
 
@@ -121,13 +123,15 @@ def build_autonomous_replan_cli_actions(
     lifecycle_actor_args: str = "",
     runtime_root: str | None = None,
     successor_closeout_guard: str | None = None,
+    quota_guard_action: str | None = None,
 ) -> list[str]:
     lifecycle_reentry = project_todo_lifecycle_settlement_reentry(
         payload,
         goal_id=goal_id,
         lifecycle_actor_args=lifecycle_actor_args,
         scoped_cli_args=scoped_cli_args,
         runtime_root=runtime_root,
+        quota_guard_action=quota_guard_action,
     )
     if lifecycle_reentry is not None:
         return list(lifecycle_reentry["next_cli_actions"])

loopx/control_plane/work_items/interaction_contract.py

@@ -940,6 +940,10 @@ def _admitted_interaction_next_cli_actions(
             command_prefix=command_prefix,
             lifecycle_actor_args=lifecycle_actor_args,
             runtime_root=runtime_root,
+            quota_guard_action=(
+                f"{typed_quota_guard} --turn-instance-id {shlex.quote(turn_instance_id)}"
+                if scheduler_args and turn_instance_id else None
+            ),
             successor_closeout_guard=_render_replan_successor_closeout_guard(
                 settlement_plan=settlement_plan, scheduler_args=scheduler_args,
                 typed_quota_guard=typed_quota_guard, settlement_args=settlement_args,

loopx/control_plane/work_items/replan_settlement.py

@@ -26,6 +26,7 @@ def project_todo_lifecycle_settlement_reentry(
     triggers: list[dict[str, Any]],
     lifecycle_actor_args: list[str],
     quota_scoped_args: list[str],
+    quota_guard_argv: list[str] | None = None,
 ) -> dict[str, Any]:
     """Return TS-owned exact actions for a terminal Todo succession gap."""
 
@@ -39,6 +40,8 @@ def project_todo_lifecycle_settlement_reentry(
         }
         if runtime_root is not None:
             request["runtime_root"] = runtime_root
+        if quota_guard_argv is not None:
+            request["quota_guard_argv"] = quota_guard_argv
         result = effect_runtime_result(
             "work_item.replan_settlement.reentry",
             request,

loopx/control_plane/work_items/replan_settlement.ts

@@ -141,6 +141,17 @@ export function projectTodoLifecycleSettlementReentry(value: unknown): JsonObjec
       requireNonEmptyString(item, `quota_scoped_args[${index}]`)
     )
     : [];
+  let quotaGuardArgv: string[] | null = null;
+  if (request.quota_guard_argv !== undefined) {
+    if (!Array.isArray(request.quota_guard_argv) || request.quota_guard_argv.length === 0) {
+      throw new EffectRuntimeRequestError("quota_guard_argv must be a non-empty array");
+    }
+    // The interaction owner supplies its already-scoped guard. The old Todo's
+    // completion identity must never be substituted for the active Turn.
+    quotaGuardArgv = request.quota_guard_argv.map((item, index) =>
+      requireNonEmptyString(item, `quota_guard_argv[${index}]`)
+    );
+  }
   const nextCliActions = triggers.map((trigger) => {
     const todoId = String(trigger.todo_id);
     const completionIdentityKey = String(trigger.completion_turn_key);
@@ -166,8 +177,10 @@ export function projectTodoLifecycleSettlementReentry(value: unknown): JsonObjec
       "when this completed scope has no remaining follow-up, use its original completion identity " +
       "with --no-follow-up above. This closes only that scope, not the Goal or other Todos. " +
       "Do not create lifecycle-only filler",
-    `${commandPrefix} --format json quota should-run --goal-id ${shellArgument(goalId)}` +
-      argumentSuffix(quotaScopedArgs),
+    quotaGuardArgv === null
+      ? `${commandPrefix} --format json quota should-run --goal-id ${shellArgument(goalId)}` +
+        argumentSuffix(quotaScopedArgs)
+      : quotaGuardArgv.map(shellArgument).join(" "),
   );
   return {
     schema_version: TODO_LIFECYCLE_REENTRY_SCHEMA,

tests/control_plane/test_lifecycle_guard_reentry.py

@@ -2,6 +2,8 @@
 from __future__ import annotations
 
 import json
+import shlex
+import subprocess
 from hashlib import sha256
 from pathlib import Path
 
@@ -46,15 +48,30 @@ def test_hidden_completed_source_recovers_its_original_guard_after_new_work(
         "action_kind": "monitor", "target_key": "independent-source",
         "cadence": "30m", "next_due_at": "2099-01-01T00:00:00Z",
         "expires_at": "2099-01-02T00:00:00Z"}
-    call, runtime, index = _fixture(tmp_path, monkeypatch, provider,
+    source_call, runtime, index = _fixture(tmp_path, monkeypatch, provider,
         initial_todos=[completed, *recent, watch])
+
+    def call(*args, **kwargs):
+        # This test inspects the explicit inline CLI contract, as does the
+        # affected context-off worker. Progressive reader coverage is separate.
+        return source_call(*args, *(["--context-delivery", "off"]
+            if "should-run" in args else []), **kwargs)
+
     index.write_text("")
     turn = "turn-admitted-lifecycle-recovery"
     admitted = _guard(call, turn)
     identity = admitted["heartbeat_receipt"]["settlement_identity"]
     obligation = admitted["autonomous_replan_obligation"]
     assert obligation["resolution_mode"] == "todo_lifecycle_settlement"
     assert [row["todo_id"] for row in obligation["triggers"]] == [original_id]
+    # Follow the public recovery command, rather than supplying the missing
+    # identity by hand. The completed Todo's identity belongs to an older Turn.
+    reentry = shlex.split(admitted["interaction_contract"]["cli_channel"]["next_cli_actions"][-1])
+    assert reentry[reentry.index("--turn-instance-id") + 1] == turn
+    assert reentry[reentry.index("--registry") + 1] == str(tmp_path / "registry.json")
+    assert reentry[reentry.index("--runtime-root") + 1] == str(runtime)
+    assert "--codex-app" in reentry
+    assert "--completion-identity-key" not in reentry
 
     # Independent new work cannot discharge the admitted lifecycle.
     def add_work():
@@ -98,7 +115,7 @@ def add_work():
             "--agent-id", AGENT, "--task-class", "user_gate", "--global-gate",
             "--text", "Authorize the separate newly requested source.")
 
-    recovered = _guard(call, turn)
+    recovered = call(*reentry[1:])
     assert recovered["heartbeat_receipt"]["settlement_identity"] == identity
     assert recovered.get("selected_todo") is None
     assert recovered["autonomous_replan_obligation"]["resolution_mode"] == "receipt_bound_replan_settlement"
@@ -141,3 +158,37 @@ def add_work():
         assert interaction["response_plan"]["decision"] == "ask_user"
     assert len([row for row in index.read_text().splitlines()
         if json.loads(row).get("classification") == "quota_slot_spent"]) == 1
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+@pytest.mark.parametrize("delivery", ["auto", "off"])
+def test_native_cli_context_exposes_the_same_bound_lifecycle_reentry(
+    tmp_path, monkeypatch, provider, delivery,
+):
+    completed = {"schema_version": "todo_item_v0", "archive_state": "active",
+        "source_section": "Agent Todo", "todo_id": "todo_completed", "role": "agent", "index": 1,
+        "text": "Deliver the accepted finite source audit.", "status": "done", "done": True,
+        "task_class": "advancement_task", "claimed_by": AGENT, "last_actor_agent_id": AGENT,
+        "completed_at": "2026-08-01T00:00:00Z", "updated_at": "2026-08-01T00:00:00Z",
+        "completion_continuation": "active_goal"}
+    call, _, index = _fixture(tmp_path, monkeypatch, provider, initial_todos=[completed])
+    index.write_text("")
+    turn = "turn-captured-lifecycle"
+    entry = call("quota", "should-run", "--runtime-profile", "generic_cli", "--goal-id", GOAL,
+        "--agent-id", AGENT, "--turn-instance-id", turn, "--context-delivery", delivery,
+        "--decision-output-dir", str(tmp_path / "capture"))
+    if delivery == "auto":
+        assert entry["schema_version"] == "context_capture_navigation_v1"
+        selected = call("quota", "context-read", "--reference-file", entry["recovery"]["reference_file"],
+            "--select", "/replan_action_packet/writeback", "--page-bytes", "8000")
+        assert selected["eof"] is True
+        writer_command = json.loads(selected["content"])["selections"][0]["value"]
+    else:
+        writer_command = entry["replan_action_packet"]["writeback"]
+    result = subprocess.run(shlex.split(writer_command), capture_output=True, text=True, check=True)
+    writer = json.loads(result.stdout)
+    actions = writer["writeback"]["next_cli_actions"]
+    guard = shlex.split(actions[-1])
+    assert guard[guard.index("--turn-instance-id") + 1] == turn
+    assert guard[guard.index("--runtime-profile") + 1] == "generic_cli"
+    assert guard[guard.index("--registry") + 1] == str(tmp_path / "registry.json")

tests/control_plane/test_replan_writeback_projection.py

@@ -80,6 +80,37 @@ def test_unguarded_successor_still_ends_without_inventing_turn_settlement(settle
                        "on host_action=end_current_heartbeat: stop"]
 
 
+@pytest.mark.parametrize("profile", ["generic_cli", "codex_app_heartbeat", "outer_controller"])
+@pytest.mark.parametrize("turn", [None, "turn-active-lifecycle"])
+def test_lifecycle_reentry_retains_current_route_without_reusing_completion_identity(profile, turn):
+    import shlex
+    from loopx.control_plane.scheduler.execution_context import scheduler_execution_context_for_runtime_profile
+    from loopx.control_plane.work_items.interaction_contract import interaction_next_cli_actions
+
+    actions = interaction_next_cli_actions(
+        {"goal_id": "example", "agent_identity": {"agent_id": "agent-example"},
+         "autonomous_replan_obligation": {"resolution_mode": "todo_lifecycle_settlement",
+            "triggers": [{"kind": "completed_advancement_without_successor",
+                "todo_id": "todo_completed", "completion_turn_key": "old-completion"}]}},
+        mode="autonomous_replan", available_capabilities=["shell"],
+        scheduler_execution_context=scheduler_execution_context_for_runtime_profile(profile),
+        turn_instance_id=turn, runtime_root="/tmp/lifecycle-runtime",
+        registry_path="/tmp/lifecycle-registry.json",
+    )
+    assert "--completion-identity-key old-completion" in actions[0]
+    assert "--turn-instance-id" not in actions[0]
+    guard = shlex.split(actions[-1])
+    if turn is None:
+        assert "--turn-instance-id" not in guard
+    else:
+        assert guard[guard.index("--turn-instance-id") + 1] == turn
+        assert guard[guard.index("--registry") + 1] == "/tmp/lifecycle-registry.json"
+        assert guard[guard.index("--runtime-root") + 1] == "/tmp/lifecycle-runtime"
+        assert guard[guard.index("--available-capability") + 1] == "shell"
+        assert ("--codex-app" in guard) is (profile == "codex_app_heartbeat")
+    assert "--completion-identity-key" not in guard
+
+
 @pytest.mark.parametrize("profile", ["codex_app_heartbeat", "trae_app", "outer_controller"])
 @pytest.mark.parametrize("binding_kind", ["autonomous_replan", "todo"])
 def test_successor_guard_entrypoint_retains_host_and_only_pure_replan_binding(

tests/control_plane_ts/replan_settlement.test.ts

@@ -117,6 +117,41 @@ test("Todo lifecycle reentry rejects untyped succession triggers", () => {
   );
 });
 
+test("lifecycle closeout returns to the active Turn's complete guard route", () => {
+  const guard = ["loopx", "--registry", "/tmp/registry with spaces.json",
+    "--runtime-root", "/tmp/current runtime", "--format", "json", "quota", "should-run",
+    "--goal-id", "goal-example", "--agent-id", "current-agent",
+    "--runtime-profile", "generic_cli", "--available-capability", "shell",
+    "--turn-instance-id", "current-turn"];
+  const result = projectTodoLifecycleSettlementReentry({
+    schema_version: TODO_LIFECYCLE_REENTRY_REQUEST_SCHEMA,
+    goal_id: "goal-example",
+    triggers: [{kind: "completed_advancement_without_successor", todo_id: "todo_old",
+      completion_turn_key: "old-completion-turn"}],
+    lifecycle_actor_args: ["--agent-id", "current-agent"],
+    quota_guard_argv: guard,
+  });
+  const actions = result.next_cli_actions as string[];
+  assert.match(actions[0], /--completion-identity-key old-completion-turn/);
+  assert.doesNotMatch(actions[0], /current-turn/);
+  assert.equal(actions.at(-1), "loopx --registry '/tmp/registry with spaces.json' " +
+    "--runtime-root '/tmp/current runtime' --format json quota should-run --goal-id goal-example " +
+    "--agent-id current-agent --runtime-profile generic_cli --available-capability shell " +
+    "--turn-instance-id current-turn");
+  assert.deepEqual(guard.slice(-2), ["--turn-instance-id", "current-turn"]);
+});
+
+test("lifecycle reentry rejects malformed supplied guard arguments", () => {
+  for (const guard of [null, "loopx quota should-run", [], ["loopx", 1], [""]]) {
+    assert.throws(() => projectTodoLifecycleSettlementReentry({
+      schema_version: TODO_LIFECYCLE_REENTRY_REQUEST_SCHEMA,
+      goal_id: "goal-example",
+      triggers: [{kind: "completed_advancement_without_successor", todo_id: "todo_old"}],
+      quota_guard_argv: guard,
+    }), /quota_guard_argv/);
+  }
+});
+
 test("Todo lifecycle reentry rejects a non-string runtime root", () => {
   assert.throws(
     () => projectTodoLifecycleSettlementReentry({