diff --git a/docs/reference/required-work-context.md b/docs/reference/required-work-context.md index 973970e951..63075a3ff5 100644 --- a/docs/reference/required-work-context.md +++ b/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. diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index 2d3bedadac..5cb5b40302 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/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})) : []; diff --git a/loopx/control_plane/quota/turn_envelope.ts b/loopx/control_plane/quota/turn_envelope.ts index 0977753697..0c2961b2b5 100644 --- a/loopx/control_plane/quota/turn_envelope.ts +++ b/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; diff --git a/loopx/control_plane/work_items/interaction_contract.ts b/loopx/control_plane/work_items/interaction_contract.ts index f7f4dba175..84b4e4bb96 100644 --- a/loopx/control_plane/work_items/interaction_contract.ts +++ b/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. diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py index 7ac573a4a9..1c5dcee278 100644 --- a/tests/control_plane/test_context_first_read.py +++ b/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,12 +1040,24 @@ 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) @@ -1052,7 +1065,7 @@ def test_initial_read_leaves_large_agenda_rows_to_the_matching_required_read(tmp 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] diff --git a/tests/test_delegation_turn_context.py b/tests/test_delegation_turn_context.py index 6797b1ac0c..4abd5432cb 100644 --- a/tests/test_delegation_turn_context.py +++ b/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" \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"))])