Avoid rereading the Manager Inbox before Todo selection

docs/reference/required-work-context.md

@@ -214,13 +214,16 @@ revision and advisory authority. This captured pre-work view does not replace th
 preference owner's fresh exact-scope read before a later preference-dependent
 external action, and neither delivery nor a preference grants authority.
 The initial read includes the bounded `unfinished_delegations` projection and
-`read_complete` flag. For an admitted normal work Turn with a large
-agenda, a complete scan and a matching required Manager Inbox read, the initial
-view keeps the incoming and outgoing request counts, scan status and required
-command. A complete scan can still hit the summary's display limit. The initial
-view then also keeps the exact omitted count and reconciliation guidance. The
-required read and its existing cursor deliver the scoped request rows. The
-captured summary remains available at its original address. An incomplete scan, unavailable scan records or an unmatched read keeps the whole summary inline.
+`read_complete` flag. For normal work with a large agenda, a complete scan
+and a matching required Manager Inbox read, the initial view keeps the incoming
+and outgoing request counts, scan status and required command. This also
+applies during explicit Todo selection when work and quota spending are
+still disabled. A complete scan can still hit the summary's display limit.
+The initial view then also keeps the exact omitted count and reconciliation
+guidance. The required read and its existing cursor deliver the scoped request
+rows. The captured summary remains available at its original address. An
+incomplete scan, unavailable scan records or an unmatched read keeps the whole
+summary inline.
 Adoption and queued returns are intermediate states. Read the full current request before
 assessing, executing or returning it. Use the existing scoped request and
 consumption routes for those actions.

loopx/control_plane/quota/progressive_context.ts

@@ -8,7 +8,7 @@ import {completeGoalNarrative} from "../work_items/goal_context_source.ts";
 import type {JsonObject} from "../effect_program.ts";
 import {jsonObject, requireBoolean, requireJsonObject} from "../runtime_decode.ts";
 import {GOAL_OUTCOMES_SCHEMA, OUTCOME_CONTEXT_NAVIGATION_SCHEMA} from "../goals/outcome_requirements.ts";
-import {FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE, selectedWorkTodoId}
+import {FULL_OUTCOME_READ_GUIDANCE, PROGRESSIVE_OUTCOME_READ_GUIDANCE, hasUnboundActionSelection, selectedWorkTodoId}
   from "../work_items/interaction_contract.ts";
 import {quotaSummaryReferences, contextSourceIdentity, contextProjectionChanges,
   type ProjectionSourceValues} from "./diagnostic_projection.ts";
@@ -977,8 +977,13 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly
     for (const field of admissionFields) include("interaction_contract", "cli_channel", field);
   }
   const channel = object(interaction.agent_channel);
+  // The matching inbox read is already required during explicit Todo selection.
+  const selectingBeforeWork = context.ok === true && context.should_run === true &&
+    channel?.delivery_allowed === false && cli?.spend_allowed_now === false &&
+    cli.spend_after_validation === false && hasUnboundActionSelection(cli);
   const requiredInboxReads = context.effective_action === EffectiveAction.NORMAL_RUN && context.normal_delivery_allowed === true
-    && channel?.delivery_allowed === true && channel.must_attempt === true && Array.isArray(channel.required_reads)
+    && channel?.must_attempt === true && (channel.delivery_allowed === true || selectingBeforeWork)
+    && Array.isArray(channel.required_reads)
     ? channel.required_reads : [];
   const sources = object(channel?.work_context)?.sources;
   const sourceRows = Array.isArray(sources) ? sources.map((source, index) => ({source, index})) : [];

loopx/control_plane/quota/turn_envelope.ts

@@ -31,7 +31,9 @@ import { projectTurnStartUnavailableContext } from "../capability_hooks.ts";
 import { jsonObject, requireBoolean, requireJsonObject, requireNonEmptyString, requireStringLiteral,
   sourceNumberTokens } from "../runtime_decode.ts";
 import { projectPendingCapabilityIntent } from "../work_items/pending_capability_intent.ts";
-import { INTERACTION_CONTRACT_SCHEMA_VERSION, assertGoalContextDelivered, assertInteractionRequiredReads, capturedTodoTextSha256, decodeInteractionContract, projectInteractionRequiredReads, selectedWorkTodo, selectedWorkTodoId, USER_CHANNEL_NOTIFICATION_POLICIES } from "../work_items/interaction_contract.ts";
+import { INTERACTION_CONTRACT_SCHEMA_VERSION, assertGoalContextDelivered, assertInteractionRequiredReads,
+  capturedTodoTextSha256, decodeInteractionContract, hasUnboundActionSelection, projectInteractionRequiredReads,
+  selectedWorkTodo, selectedWorkTodoId, USER_CHANNEL_NOTIFICATION_POLICIES } from "../work_items/interaction_contract.ts";
 import { measureTurnEnvelope, TURN_ENVELOPE_SECTION_TARGETS } from "./turn_envelope_budget.ts";
 export { TURN_ENVELOPE_BUDGET_BYTES } from "./turn_envelope_budget.ts";
 
@@ -2359,8 +2361,7 @@ function progressivePreparationScopes(payload: JsonObject): PreparationScopeDeta
     obligation?.kind === EffectiveAction.AGENT_WORKSPACE_REPAIR && obligation.contract === "agent_workspace_guard" &&
     obligation.must_attempt_work === true && obligation.delivery_allowed === false;
   if (!repair && payload.effective_action !== EffectiveAction.NORMAL_RUN) return [];
-  const selecting = cli.selection_required === true && !Object.hasOwn(cli, "execution_preparation") &&
-    jsonObject(cli.selection_command)?.schema_version === "action_selection_cli_command_v1";
+  const selecting = hasUnboundActionSelection(cli);
   const claiming = cli.selection_required !== true && payload.normal_delivery_allowed === false &&
     preparation?.schema_version === EXECUTION_PREPARATION_SCHEMA_VERSION && preparation.state === "claim_required" &&
     preparation.delivery_allowed === false && jsonObject(preparation.claim)?.host_execution_allowed === true;

loopx/control_plane/work_items/interaction_contract.ts

@@ -765,6 +765,12 @@ function shellQuote(value: string): string {
   return /^[A-Za-z0-9_./:-]+$/.test(value) ? value : `'${value.replaceAll("'", `'"'"'`)}'`;
 }
 
+/** Identify the explicit selection route before any work has been bound. */
+export function hasUnboundActionSelection(cli: JsonObject | null | undefined): boolean {
+  return cli?.selection_required === true && !Object.hasOwn(cli, "execution_preparation") &&
+    jsonObject(cli.selection_command)?.schema_version === "action_selection_cli_command_v1";
+}
+
 /** The adaptive primary is the work identity for both reads and settlement.
  * Historical callers may supply their already-decoded text: reapplying JS
  * trim would change Python whitespace/BOM identities in persisted journals.

tests/control_plane/test_context_first_read.py

@@ -1018,7 +1018,8 @@ def test_initial_read_preserves_unfinished_requests_and_incomplete_observation(t
 
 @pytest.mark.parametrize("route", [
     "matching", "bounded", "different", "missing", "incomplete", "omitted", "unknown",
-    "bad_counts", "small", "held", "missing_instruction",
+    "bad_counts", "small", "held", "missing_instruction", "selection",
+    "selection_unavailable_route", "selection_spend_allowed", "selection_different_read",
 ])
 def test_initial_read_leaves_large_agenda_rows_to_the_matching_required_read(tmp_path, decision, route):
     command = "loopx manager-inbox read --goal-id fixture --agent-id author"
@@ -1039,20 +1040,32 @@ def test_initial_read_leaves_large_agenda_rows_to_the_matching_required_read(tmp
         del agenda["instruction"]
     decision.update(should_run=True, normal_delivery_allowed=True, effective_action="normal_run")
     channel = decision["interaction_contract"]["agent_channel"]
-    channel.update(delivery_allowed=route != "held", must_attempt=True)
+    selecting = route.startswith("selection")
+    channel.update(delivery_allowed=not selecting and route != "held", must_attempt=True)
+    if selecting:
+        decision["interaction_contract"]["cli_channel"] = {
+            "selection_required": True, "next_cli_actions": [],
+            "spend_allowed_now": route == "selection_spend_allowed", "spend_after_validation": False,
+            **({"selection_command": {"schema_version": "action_selection_cli_command_v1",
+                "route_prefix": "loopx --format json",
+                "command_args_template": "quota should-run --todo-id '{todo_id}' --turn-instance-id fixture-turn",
+                "candidate_discovery_args": "todo list --goal-id fixture --status open"}}
+                if route != "selection_unavailable_route" else {}),
+        }
     channel["work_context"]["sources"].append({**read, "content": {
         "ok": True, "read_complete": False, "unfinished_delegations": agenda,
     }})
     if route != "missing":
-        required = {**read, "command": command + " --agent-id other"} if route == "different" else read
+        required = ({**read, "command": command + " --agent-id other"}
+                    if route in {"different", "selection_different_read"} else read)
         channel["required_reads"].append(required)
 
     view = native_context_capture_view(decision, tmp_path)
     _, values = read_selection(view["first_read_command"])
     prefix = "/interaction_contract/agent_channel/work_context/sources/3/content"
     assert values[prefix + "/read_complete"] is False
     agenda_path = prefix + "/unfinished_delegations"
-    if route in {"matching", "bounded"}:
+    if route in {"matching", "bounded", "selection"}:
         assert agenda_path not in values and agenda_path + "/items" not in values
         for field in ("incoming_count", "outgoing_count", "scan_complete"):
             assert values[f"{agenda_path}/{field}"] == agenda[field]

tests/test_delegation_turn_context.py

@@ -97,6 +97,81 @@ def test_separate_agent_has_no_access_to_another_receivers_agenda(flow):
     assert agenda(root, registry, "unrelated")["items"] == []
 
 
+@pytest.mark.skipif(os.name != "posix", reason="native context delivery requires POSIX")
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_unbound_cli_first_read_leaves_inbox_rows_to_its_required_read(tmp_path, monkeypatch, provider):
+    from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime
+    from tests.control_plane.test_context_first_read import read_selection
+    from tests.control_plane.test_quota_settlement_cli import (
+        AGENT_ID, GOAL_ID, TURN_ID, _heartbeat_receipt_count, _initialize_fixture_authority,
+        _run_cli, _spend_run_count, _write_fixture,
+    )
+    from loopx.control_plane.turn_driver.decision_capture import read_context_read_reference
+
+    isolate_sqlite_runtime(tmp_path, monkeypatch)
+    monkeypatch.delenv("CODEX_THREAD_ID", raising=False)
+    project, runtime, registry = _write_fixture(tmp_path)
+    state = project / f".codex/goals/{GOAL_ID}/ACTIVE_GOAL_STATE.md"
+    with state.open("a", encoding="utf-8") as stream:
+        stream.write("\n- [ ] [P2] Verify a different source.\n"
+            f"  <!-- loopx:todo todo_id=todo_other_source status=open claimed_by={AGENT_ID} "
+            "task_class=advancement_task action_kind=validate -->\n")
+    _initialize_fixture_authority(project, runtime, provider=provider)
+    request_ids = set()
+    for index in range(23):
+        result = deliver(runtime, registry, session={"session_id": "source", "channel_id": "manager"},
+            turn={"client_turn_id": f"synthetic-review-{index}", "origin": "web", "message":
+                "Compare the original requirements with the retained report before claiming a result."},
+            request={"goal_id": GOAL_ID, "agent_id": AGENT_ID})
+        request_ids.add(result["request_id"])
+
+    captures = tmp_path / "captures"
+    captures.mkdir(mode=0o700)
+    console = Path(sysconfig.get_path("scripts")) / "loopx"
+    assert console.is_file(), "Install this checkout's console with uv sync --extra test."
+    code, view = _run_cli(registry, runtime, "--cli-bin", str(console), "quota", "should-run", "--codex-app",
+        "--context-delivery", "on", "--goal-id", GOAL_ID, "--agent-id", AGENT_ID,
+        "--turn-instance-id", TURN_ID, "--decision-output-root", str(captures), cwd=project)
+    assert code == 0 and view["mode"] == "full", view
+    _, reference = read_context_read_reference(view["recovery"]["reference_file"])
+    path = Path(reference["path"])
+    original = path.read_bytes()
+    channel = json.loads(original)["interaction_contract"]["agent_channel"]
+    index, = (i for i, row in enumerate(channel["work_context"]["sources"]) if row.get("kind") == "operator_inbox")
+    address = f"/interaction_contract/agent_channel/work_context/sources/{index}/content/unfinished_delegations"
+    _, first = read_selection(view["first_read_command"])
+    if "remaining_read_command" in view:
+        _, remaining = read_selection(view["remaining_read_command"])
+        first.update(remaining)
+    cli = "/interaction_contract/cli_channel"
+    assert first[cli + "/selection_required"] is True
+    assert first[cli + "/spend_allowed_now"] is False
+    assert first["/interaction_contract/agent_channel/delivery_allowed"] is False
+    assert address not in first and address + "/items" not in first
+    assert first[address + "/incoming_count"] == len(request_ids) == 23
+    assert first[address + "/omitted_count"] > 0
+    read, = (row for row in first["/interaction_contract/agent_channel/required_reads"] if row.get("kind") == "operator_inbox")
+    command = shlex.split(read["command"])
+    assert command[0] == str(console)
+    cursor, seen = None, set()
+    while True:
+        result = subprocess.run([*command, *(["--cursor", cursor] if cursor else [])],
+            cwd=project, capture_output=True, text=True, encoding="utf-8", timeout=40)
+        assert result.returncode == 0, result.stderr
+        page = json.loads(result.stdout.split("\n```json\n", 1)[1].removesuffix("\n```\n"))
+        page_ids = {item["request_id"] for item in page["items"]}
+        assert page["ok"] and page_ids and page_ids <= request_ids - seen
+        seen.update(page_ids)
+        cursor = page["next_cursor"]
+        if cursor is None:
+            break
+    assert seen == request_ids
+    _, saved = read_selection(view["recovery"]["command"] + " --select " + shlex.quote(address))
+    assert saved[address] == channel["work_context"]["sources"][index]["content"]["unfinished_delegations"]
+    assert path.read_bytes() == original
+    assert _heartbeat_receipt_count(runtime, TURN_ID) == 1 and _spend_run_count(runtime) == 0
+
+
 @pytest.mark.parametrize("provider", ["file", "sqlite"])
 @pytest.mark.parametrize("delivery", ["host_envelope", pytest.param("native_files",
     marks=pytest.mark.skipif(os.name != "posix", reason="native context delivery requires POSIX"))])