fix(quota): preserve scoped task workspace recovery identity

loopx/control_plane/quota/should_run_prepare.py

@@ -616,6 +616,43 @@ def _receipt_delivery_continuity(
     return anchor, todo
 
 
+def _prepare_gate_selection_candidates(
+    *, agent_identity: dict[str, Any] | None,
+    source_items: list[dict[str, Any]], planning_items: list[dict[str, Any]],
+    available_capabilities: Any, requested_todo_id: str | None,
+    receipt_bound_todo_id: str | None,
+) -> tuple[dict[str, Any] | None, str | None, dict[str, Any] | None]:
+    """Transport exact candidates through the existing selection/gate owners."""
+    # Resolve the exact choice before selecting a gate-independent route. The
+    # ranked fallback is advice, not an additional permission boundary.
+    requested_action_candidate = (
+        build_explicit_advancement_next_action(
+            agent_identity=agent_identity,
+            agent_todo_items=planning_items,
+            available_capabilities=available_capabilities,
+            todo_id=requested_todo_id,
+            selection_binding="pending_action_selection",
+        )
+        if requested_todo_id
+        else None
+    )
+    # Recovery keeps the committed work identity. Requalify that exact work
+    # against the current gate instead of borrowing an unrelated default's
+    # independence after the explicit selection has become a receipt.
+    fallback_todo_id = receipt_bound_todo_id or requested_todo_id
+    fallback_candidate = (
+        build_receipt_bound_advancement_next_action(
+            agent_identity=agent_identity,
+            agent_todo_items=source_items,
+            available_capabilities=available_capabilities,
+            receipt_bound_todo_id=receipt_bound_todo_id,
+        ) if receipt_bound_todo_id else requested_action_candidate
+    )
+    source_candidate = next((item for item in source_items
+        if item.get("todo_id") == fallback_todo_id), None) if fallback_candidate else None
+    return requested_action_candidate, fallback_todo_id, source_candidate
+
+
 def _prepare_quota_should_run_item(
     status_payload: dict[str, Any],
     *,
@@ -817,30 +854,23 @@ def _prepare_quota_should_run_item(
     if task_orchestration_contract_is_actionable(task_orchestration_contract):
         capability_monitor_contract = capability_monitor_fallback = None
     work_lane_contract = capability_monitor_contract or work_lane_contract
-    # Resolve the exact choice before selecting a gate-independent route. The
-    # ranked fallback is advice, not an additional permission boundary.
-    requested_action_candidate = (
-        build_explicit_advancement_next_action(
-            agent_identity=agent_identity,
-            agent_todo_items=agent_todo_planning_source_items,
+    requested_action_candidate, fallback_todo_id, fallback_source_candidate = (
+        _prepare_gate_selection_candidates(
+            agent_identity=agent_identity, source_items=agent_todo_source_items,
+            planning_items=agent_todo_planning_source_items,
             available_capabilities=effective_available_capabilities,
-            todo_id=requested_action_todo_id,
-            selection_binding="pending_action_selection",
+            requested_todo_id=requested_action_todo_id,
+            receipt_bound_todo_id=receipt_bound_todo_id,
         )
-        if requested_action_todo_id
-        else None
     )
     scoped_user_gate_fallback = _scoped_user_gate_fallback(
         user_todo_summary,
         agent_todo_summary,
         capability_gate=capability_gate,
         allow_unrelated_gate=bool(quota.get("safe_bypass_allowed")),
         monitor_debt_backoff_active=bool(monitor_debt_arbitration.get("active")),
-        requested_todo_id=requested_action_todo_id,
-        requested_candidate=next((
-            item for item in agent_todo_planning_source_items
-            if item.get("todo_id") == requested_action_todo_id
-        ), None) if requested_action_candidate is not None else None,
+        requested_todo_id=fallback_todo_id,
+        requested_candidate=fallback_source_candidate,
     )
     work_lane_contract = (
         scoped_user_gate_due_monitor_contract(

loopx/control_plane/todos/user_gate.py

@@ -208,6 +208,11 @@ def apply_scoped_user_gate_fallback_projection(
 
     projected = dict(payload)
     projected["scoped_user_gate_fallback"] = fallback
+    if projected.get("effective_action") == EffectiveAction.AGENT_WORKSPACE_REPAIR.value:
+        # Independence from the user gate is not workspace admission. Preserve
+        # the selected repair obligation and its explicit delivery=False so the
+        # same Todo can bind for repair and reenter after moving workspaces.
+        return projected
     projected["should_run"] = True
     if projected.get("decision") == "skip":
         projected["decision"] = "safe_bypass_user_gate_fallback"

loopx/control_plane/work_items/interaction_contract.ts

@@ -432,6 +432,10 @@ export function projectInteractionExecution(value: unknown): InteractionExecutio
     mode = "terminal_no_followup";
   } else if (action === EffectiveAction.PEER_COORDINATION_BLOCKED) {
     mode = action;
+  } else if (scopedFallback && action === EffectiveAction.AGENT_WORKSPACE_REPAIR) {
+    // The fallback proves gate independence, not permission to edit here.
+    // Keep the selected repair and its canonical delivery boundary visible.
+    mode = action;
   } else if (scopedFallback) {
     mode = "scoped_user_gate_fallback";
   } else if (notificationSuppressed) {

tests/control_plane/test_quota_settlement_cli.py

@@ -5523,11 +5523,23 @@ def test_due_successor_replan_replays_until_the_same_task_is_reopened(
     assert _spend_run_count(runtime) == 0
 
 
+@pytest.mark.parametrize("scoped_gate,pending_selection", [
+    (False, False), (False, True), (True, False),
+])
 def test_pending_selection_preserves_workspace_repair_then_reenters_same_turn(
-    tmp_path: Path,
+    tmp_path: Path, scoped_gate: bool, pending_selection: bool,
 ) -> None:
     project, runtime, registry_path = _write_fixture(tmp_path)
     _configure_selectable_alternative(project)
+    if scoped_gate:
+        state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+        state.write_text(state.read_text() + (
+            "\n## User Todo\n\n- [ ] [P0] Approve unrelated publication.\n"
+            "  <!-- loopx:todo todo_id=todo_publication_gate status=open "
+            f"task_class=user_gate blocks_agent={AGENT_ID} "
+            "decision_scope=write_scope:action:release "
+            "unblocks_todo_id=todo_publication -->\n"
+        ))
     registry = json.loads(registry_path.read_text(encoding="utf-8"))
     registry["goals"][0]["coordination"]["registered_agents"].append(
         "codex-settlement-peer"
@@ -5567,12 +5579,12 @@ def test_pending_selection_preserves_workspace_repair_then_reenters_same_turn(
     )
 
     _initialize_fixture_authority(project, runtime)
-    first_rc, first = _run_cli(
-        registry_path,
-        runtime,
-        *guard_args,
-        cwd=project,
-    )
+    if pending_selection:
+        first_rc, first = _run_cli(
+            registry_path, runtime, *guard_args, cwd=project,
+        )
+        assert first_rc == 0, first
+        assert "settlement_identity" not in first["heartbeat_receipt"]
     repair_rc, repair = _run_cli(
         registry_path,
         runtime,
@@ -5582,8 +5594,6 @@ def test_pending_selection_preserves_workspace_repair_then_reenters_same_turn(
         cwd=project,
     )
 
-    assert first_rc == 0, first
-    assert "settlement_identity" not in first["heartbeat_receipt"]
     assert repair_rc == 0, repair
     assert repair.get("error_code") != "heartbeat_receipt_identity_conflict"
     assert repair["effective_action"] == "agent_workspace_repair"
@@ -5595,7 +5605,12 @@ def test_pending_selection_preserves_workspace_repair_then_reenters_same_turn(
         "create or switch to an independent worktree/branch, then rerun quota "
         "guard before file edits"
     )
-    assert repair["heartbeat_receipt"]["status"] == "upgraded"
+    assert repair["heartbeat_receipt"]["status"] == (
+        "upgraded" if pending_selection else "committed"
+    )
+    assert repair["interaction_contract"]["agent_channel"]["delivery_allowed"] is False
+    assert repair["normal_delivery_allowed"] is False
+    assert repair["interaction_contract"]["cli_channel"]["spend_allowed_now"] is False
     assert repair["heartbeat_receipt"]["settlement_identity"]["todo_id"] == (
         ALTERNATIVE_TODO_ID
     )
@@ -5616,8 +5631,14 @@ def test_pending_selection_preserves_workspace_repair_then_reenters_same_turn(
     )
 
     assert resumed_rc == 0, resumed
-    assert resumed["effective_action"] == "normal_run"
-    assert resumed["normal_delivery_allowed"] is True
+    assert resumed["effective_action"] == (
+        "operator_gate_notify" if scoped_gate else "normal_run"
+    )
+    assert resumed["normal_delivery_allowed"] is (not scoped_gate)
+    assert resumed["interaction_contract"]["agent_channel"]["delivery_allowed"] is True
+    if scoped_gate:
+        assert resumed["scoped_user_gate_fallback"]["selected_executable"]["todo_id"] == ALTERNATIVE_TODO_ID
+        assert resumed["requires_user_action"] is True
     assert resumed["workspace_repair_allowed"] is False
     assert resumed["selected_todo"]["todo_id"] == ALTERNATIVE_TODO_ID
     assert resumed["selected_todo"]["selection_binding"] == "heartbeat_receipt"
@@ -5627,12 +5648,12 @@ def test_pending_selection_preserves_workspace_repair_then_reenters_same_turn(
     assert resumed["heartbeat_receipt"]["status"] == "upgraded"
     assert resumed["heartbeat_receipt"]["settlement_identity"] == repair["heartbeat_receipt"]["settlement_identity"]
     assert resumed["heartbeat_receipt"]["event_id"] != repair["heartbeat_receipt"]["event_id"]
-    assert _heartbeat_receipt_count(runtime, turn_instance_id) == 3
+    assert _heartbeat_receipt_count(runtime, turn_instance_id) == (3 if pending_selection else 2)
     replay_rc, replay = _run_cli(registry_path, runtime, *guard_args,
         "--todo-id", ALTERNATIVE_TODO_ID, cwd=linked_worktree)
     assert replay_rc == 0 and replay["heartbeat_receipt"]["status"] == "replayed", replay
     assert replay["heartbeat_receipt"]["event_id"] == resumed["heartbeat_receipt"]["event_id"]
-    assert _heartbeat_receipt_count(runtime, turn_instance_id) == 3
+    assert _heartbeat_receipt_count(runtime, turn_instance_id) == (3 if pending_selection else 2)
 
 
 def test_boundary_projection_repair_keeps_same_turn_alternative_selectable(

tests/control_plane/test_user_gate_lane_progress.py

@@ -59,6 +59,24 @@ def test_user_gate_fallback_preserves_repair_and_replan_precedence():
     ) is original
 
 
+def test_independent_fallback_keeps_workspace_repair_obligation_without_write_grant():
+    original = {
+        "decision": "workspace_guard", "should_run": True,
+        "effective_action": "agent_workspace_repair",
+        "normal_delivery_allowed": False, "workspace_repair_allowed": True,
+        "execution_obligation": {
+            "kind": "agent_workspace_repair", "must_attempt_work": True,
+            "delivery_allowed": False, "minimum": "move_to_independent_worktree",
+        },
+    }
+    fallback = {"selected_executable": {"todo_id": "todo_independent"}}
+    result = apply_scoped_user_gate_fallback_projection(
+        original, fallback=fallback, replan_decision_allowed=False,
+    )
+    assert result == {**original, "scoped_user_gate_fallback": fallback}
+    assert "scoped_user_gate_fallback" not in original
+
+
 def _status_payload(
     *,
     gate_action_kind: str,

tests/control_plane_ts/interaction_execution_decision.test.ts

@@ -107,6 +107,20 @@ test("should_run is not an execution-permission input", () => {
   assert.deepEqual(projected.normalization_diagnostics, []);
 });
 
+test("independent fallback retains workspace repair before delivery", () => {
+  const packet = projectInteractionExecution(executionFacts({
+    effective_action: EffectiveAction.AGENT_WORKSPACE_REPAIR,
+    state: "operator_gate", user_required: true, notify_user_on_gate: true,
+    scoped_user_gate_fallback: true,
+    canonical_execution: {kind: "agent_workspace_repair",
+      must_attempt_work: true, delivery_allowed: false},
+  }));
+  assert.equal(packet.mode, EffectiveAction.AGENT_WORKSPACE_REPAIR);
+  assert.equal(packet.must_attempt, true);
+  assert.equal(packet.delivery_allowed, false);
+  assert.equal(packet.quiet_noop_allowed, false);
+});
+
 test("canonical null and false stop compatibility fallback", () => {
   for (const canonicalDelivery of [null, false]) {
     const projected = projectInteractionExecution(executionFacts({