From bf24ca17ba87d69714c39d10fee789d99e1584cd Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 12 Nov 2026 17:12:55 +0800 Subject: [PATCH 1/2] fix(collaboration): surface peer returns on next turn Signed-off-by: huangruiteng --- loopx/capabilities/manager_context/README.md | 8 +++ .../collaboration/turn_context.ts | 65 ++++++++++++++++++- .../quota/progressive_context.ts | 41 +++++++++++- .../control_plane/test_context_first_read.py | 36 ++++++++++ .../test_request_amendment_recovery.py | 8 ++- .../collaboration_turn_context.test.ts | 38 ++++++++++- 6 files changed, 190 insertions(+), 6 deletions(-) diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index 09863a3b2d..4b196cd1dc 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/loopx/capabilities/manager_context/README.md @@ -241,6 +241,14 @@ awaiting explicit requester consumption. The same read model is returned by private agenda contains current-purpose excerpts and explicit links, not source transcripts, result bodies or receiver rationale. +The agenda also carries a bounded `return_attention` projection for sent peer +requests whose returned result has not been consumed. It contains only the +request/revision identity, result keys and delivery/consumption state. Native +context delivery places this projection in the next Turn's opening before long +work-context sources, while the full result text remains behind the existing +`manager-inbox read` command. This preserves asynchronous pull semantics and +does not interrupt an active Turn or treat visibility as consumption. + Outgoing requests to an original project conversation use the same typed address reader as explicit peer returns. The agenda retains that conversation recipient and its current revision without inventing a Goal/Agent receiver. A changed diff --git a/loopx/control_plane/collaboration/turn_context.ts b/loopx/control_plane/collaboration/turn_context.ts index 9890fc8a41..d512b41e8c 100644 --- a/loopx/control_plane/collaboration/turn_context.ts +++ b/loopx/control_plane/collaboration/turn_context.ts @@ -75,6 +75,63 @@ function visibleRows(rows: JsonObject[]): JsonObject[] { return visible; } +const RETURN_ATTENTION_LIMIT = 4; + +function peerReturnNeedsAttention(row: JsonObject): boolean { + return row.direction === "outgoing" && row.step === "consume_peer_result" + && typeof row.unconsumed_return_count === "number" && row.unconsumed_return_count > 0 + && Array.isArray(row.returns) && row.returns.some(value => { + const result = value !== null && typeof value === "object" && !Array.isArray(value) + ? value as JsonObject : undefined; + return result?.consumed !== true; + }); +} + +function peerReturnAttention(rows: JsonObject[]): JsonObject { + const pending = rows.filter(peerReturnNeedsAttention); + const items = pending.slice(0, RETURN_ATTENTION_LIMIT).map(row => { + const item: JsonObject = { + request_id: row.request_id, + request_revision: row.request_revision, + step: row.step, + unconsumed_return_count: row.unconsumed_return_count, + returns: row.returns, + }; + for (const field of ["source_agent_id", "agent_id"] as const) { + if (row[field] !== undefined) item[field] = row[field]; + } + return item; + }); + return {count: pending.length, items, omitted_count: pending.length - items.length}; +} + +function requirePeerReturnAttention(value: unknown): {count: number; items: JsonObject[]} { + const attention = requireJsonObject(value, "delegation return attention"); + const count = attention.count, omitted = attention.omitted_count; + if (typeof count !== "number" || !Number.isSafeInteger(count) || count < 0 + || typeof omitted !== "number" || !Number.isSafeInteger(omitted) || omitted < 0 + || !Array.isArray(attention.items) || attention.items.length > RETURN_ATTENTION_LIMIT + || count !== attention.items.length + omitted) { + throw new EffectRuntimeRequestError("invalid delegation return attention"); + } + const items = attention.items.map(value => { + const item = requireJsonObject(value, "delegation return attention item"); + if (typeof item.request_id !== "string" || !BARE_SHA256_PATTERN.test(item.request_id) + || typeof item.request_revision !== "number" || !Number.isSafeInteger(item.request_revision) + || item.request_revision < 1 || item.step !== "consume_peer_result" + || typeof item.unconsumed_return_count !== "number" + || !Number.isSafeInteger(item.unconsumed_return_count) || item.unconsumed_return_count < 1 + || !Array.isArray(item.returns) || !item.returns.some(value => { + const result = requireJsonObject(value, "delegation return attention result"); + return result.consumed !== true; + })) { + throw new EffectRuntimeRequestError("invalid delegation return attention item"); + } + return item; + }); + return {count, items}; +} + /** Input is already scope-checked by the existing inbox and result owners. */ export function projectDelegationTurnItems(params: JsonObject): JsonObject { if (!Array.isArray(params.observations) || params.observations.length > 128) { @@ -82,7 +139,8 @@ export function projectDelegationTurnItems(params: JsonObject): JsonObject { } const rows = params.observations.map(projectDelegationTurnItem).filter((value): value is JsonObject => value !== null); return {items: visibleRows(rows), incoming_count: rows.filter(row => row.direction === "incoming").length, - outgoing_count: rows.filter(row => row.direction === "outgoing").length}; + outgoing_count: rows.filter(row => row.direction === "outgoing").length, + return_attention: peerReturnAttention(rows)}; } export function projectDelegationTurnContext(params: JsonObject): JsonObject { @@ -103,9 +161,14 @@ export function projectDelegationTurnContext(params: JsonObject): JsonObject { }, 0); const incoming = total("incoming_count"), outgoing = total("outgoing_count"); const visible = visibleRows(rows); + const attention = batches.map(batch => requirePeerReturnAttention(batch.return_attention)); + const attentionCount = attention.reduce((sum, batch) => sum + batch.count, 0); + const attentionItems = attention.flatMap(batch => batch.items).slice(0, RETURN_ATTENTION_LIMIT); return {ok: true, read_complete: false, unfinished_delegations: { count_unit: "request", items: visible, incoming_count: incoming, outgoing_count: outgoing, omitted_count: incoming + outgoing - visible.length, + return_attention: {count: attentionCount, items: attentionItems, + omitted_count: attentionCount - attentionItems.length}, scan_complete: params.scan_complete === true && params.unavailable_count === 0, unavailable_count: params.unavailable_count, observed_at: params.observed_at, read_command: params.read_command, diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index 2d3bedadac..21fddb6933 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/loopx/control_plane/quota/progressive_context.ts @@ -867,8 +867,7 @@ export function scopedReplanYardstickReversal(decision: JsonObject) { return {cue, fields, path: [...REPLAN_CONTEXT_PATH, "explore_basis", "work_yardsticks", index]}; } -/** A pending Manager Inbox read already delivers these scoped request rows. */ -function managerAgendaOpeningFields(value: unknown, requiredReads: unknown[]): string[] | undefined { +function requiredManagerAgenda(value: unknown, requiredReads: unknown[]): JsonObject | undefined { const {content, ...read} = jsonObject(value) ?? {}; const source = jsonObject(content), agenda = jsonObject(source?.unfinished_delegations); if (read.source !== "turn_start_capability_hook" || read.kind !== "operator_inbox" @@ -879,12 +878,20 @@ function managerAgendaOpeningFields(value: unknown, requiredReads: unknown[]): s || agenda.scan_complete !== true || agenda.unavailable_count !== 0 || !Array.isArray(agenda.items) || !known(agenda, ["count_unit", "items", "incoming_count", "outgoing_count", "omitted_count", - "scan_complete", "unavailable_count", "observed_at", "read_command", "instruction"]) + "return_attention", "scan_complete", "unavailable_count", "observed_at", "read_command", "instruction"]) || !requiredReads.some(required => isDeepStrictEqual(required, read))) return undefined; + return agenda; +} + +/** A pending Manager Inbox read already delivers these scoped request rows. */ +function managerAgendaOpeningFields(value: unknown, requiredReads: unknown[]): string[] | undefined { + const agenda = requiredManagerAgenda(value, requiredReads); + if (!agenda) return undefined; const {incoming_count: incoming, outgoing_count: outgoing, omitted_count: omitted, items} = agenda; if (typeof incoming !== "number" || !Number.isSafeInteger(incoming) || incoming < 0 || typeof outgoing !== "number" || !Number.isSafeInteger(outgoing) || outgoing < 0 || typeof omitted !== "number" || !Number.isSafeInteger(omitted) || omitted < 0 + || !Array.isArray(items) || !Number.isSafeInteger(incoming + outgoing) || incoming + outgoing !== items.length + omitted || Buffer.byteLength(JSON.stringify(items), "utf8") < 2048) return undefined; const fields = ["incoming_count", "outgoing_count", "scan_complete"]; @@ -897,6 +904,26 @@ function managerAgendaOpeningFields(value: unknown, requiredReads: unknown[]): s return undefined; } +/** Keep only returned-result identities in the first page; full Inbox text stays on its owner read. */ +function managerReturnAttention(value: unknown, requiredReads: unknown[]): JsonObject | undefined { + const agenda = requiredManagerAgenda(value, requiredReads); + const attention = jsonObject(agenda?.return_attention); + if (!attention || !known(attention, ["count", "items", "omitted_count"]) + || typeof attention.count !== "number" || !Number.isSafeInteger(attention.count) || attention.count <= 0 + || typeof attention.omitted_count !== "number" || !Number.isSafeInteger(attention.omitted_count) + || attention.omitted_count < 0 || !Array.isArray(attention.items) || attention.items.length > 4 + || attention.count !== attention.items.length + attention.omitted_count) return undefined; + for (const value of attention.items) { + const item = jsonObject(value); + if (!item || item.step !== "consume_peer_result" || typeof item.request_id !== "string" + || typeof item.unconsumed_return_count !== "number" || item.unconsumed_return_count <= 0 + || !Array.isArray(item.returns) || !item.returns.some(value => jsonObject(value)?.consumed !== true)) { + return undefined; + } + } + return attention; +} + function initialContextReadGroups(context: JsonObject, decisionFields: readonly string[]): { opening: string[]; remaining: string[]; } { @@ -982,6 +1009,14 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly ? channel.required_reads : []; const sources = object(channel?.work_context)?.sources; const sourceRows = Array.isArray(sources) ? sources.map((source, index) => ({source, index})) : []; + // A long Manager Inbox agenda stays behind its required read. Surface only + // the bounded returned-result identities before work so paging or context + // recovery cannot hide that the requester has a result to consume. + for (const {source, index} of sourceRows) { + if (managerReturnAttention(source, requiredInboxReads)) { + include(...work, "sources", index, "content", "unfinished_delegations", "return_attention"); + } + } const selectedTodo = object(context.selected_todo); const identityAgent = object(context.agent_identity)?.agent_id; const agentId = context.agent_id ?? identityAgent; diff --git a/tests/control_plane/test_context_first_read.py b/tests/control_plane/test_context_first_read.py index 7ac573a4a9..e950f9614c 100644 --- a/tests/control_plane/test_context_first_read.py +++ b/tests/control_plane/test_context_first_read.py @@ -1071,6 +1071,42 @@ def test_initial_read_leaves_large_agenda_rows_to_the_matching_required_read(tmp assert json.loads(detail["content"])["selections"][0]["value"] == agenda +def test_initial_read_surfaces_bounded_peer_return_attention_before_long_sources(tmp_path, decision): + command = "loopx manager-inbox read --goal-id fixture --agent-id author" + read = {"source": "turn_start_capability_hook", "kind": "operator_inbox", + "capability_id": "manager-context", "hook_id": "manager.context_inbox", + "ordering": "before_work", "command": command} + attention = {"count": 1, "omitted_count": 0, "items": [{ + "request_id": "a" * 64, "request_revision": 1, "step": "consume_peer_result", + "unconsumed_return_count": 1, + "returns": [{"result_key": "conclusion", "request_revision": 1, + "status": "delivered", "consumed": False}], + }]} + items = [{"request_id": f"{index:064x}", "direction": "incoming" if index < 10 else "outgoing", + "step": "review_request_work", "purpose": "Keep the original evidence. " * 16} + for index in range(20)] + agenda = {"count_unit": "request", "incoming_count": 10, "outgoing_count": 11, + "items": items, "omitted_count": 1, "return_attention": attention, + "scan_complete": True, "unavailable_count": 0, "read_command": command, + "instruction": "Read the full current request before acting."} + decision.update(should_run=True, normal_delivery_allowed=True, effective_action="normal_run") + channel = decision["interaction_contract"]["agent_channel"] + channel.update(delivery_allowed=True, must_attempt=True) + channel["required_reads"].append(read) + sources = channel["work_context"]["sources"] + sources.append({**read, "content": {"ok": True, "read_complete": False, + "unfinished_delegations": agenda}}) + + view = native_context_capture_view(decision, tmp_path) + _, values = read_selection(view["first_read_command"]) + prefix = f"/interaction_contract/agent_channel/work_context/sources/{len(sources) - 1}/content/unfinished_delegations" + attention_path = prefix + "/return_attention" + assert values[attention_path] == attention + assert prefix not in values and prefix + "/items" not in values + goal_text = "/interaction_contract/agent_channel/work_context/sources/0/content/text" + assert list(values).index(attention_path) < list(values).index(goal_text) + + def test_unknown_settlement_plan_is_not_compacted(tmp_path, decision): plan = {"schema_version": "future", "new_stop": "Stop before every new effect."} decision["interaction_contract"]["cli_channel"]["settlement_plan"] = plan diff --git a/tests/control_plane/test_request_amendment_recovery.py b/tests/control_plane/test_request_amendment_recovery.py index 4913241e66..b54444fe7a 100644 --- a/tests/control_plane/test_request_amendment_recovery.py +++ b/tests/control_plane/test_request_amendment_recovery.py @@ -128,6 +128,10 @@ def test_native_inbox_preserves_unconsumed_results_across_corrections(source_goa assert code == 0, packet agenda = packet["unfinished_delegations"] assert agenda["outgoing_count"] == 1, agenda + attention = agenda["return_attention"] + assert attention["count"] == 1 and attention["omitted_count"] == 0 + assert attention["items"][0]["request_id"] == rid + assert attention["items"][0]["step"] == "consume_peer_result" row = agenda["items"][0] assert row["request_id"] == rid and row["request_revision"] == 2 assert row["purpose"] == "Check the corrected contract" @@ -137,7 +141,9 @@ def test_native_inbox_preserves_unconsumed_results_across_corrections(source_goa assert {p.relative_to(_root(root)): p.read_bytes() for p in _root(root).rglob("*.json")} == before consume_return(root, goal, "agent-b", rid, registry=registry) for actor in ("agent-a", "agent-b"): - assert read_delegation_turn_context(root, registry, goal, actor)["unfinished_delegations"]["items"] == [] + agenda = read_delegation_turn_context(root, registry, goal, actor)["unfinished_delegations"] + assert agenda["items"] == [] + assert agenda["return_attention"] == {"count": 0, "items": [], "omitted_count": 0} def test_request_control_refuses_wrong_actor_namespace_and_untrusted_effect_facts(source_goal, tmp_path): diff --git a/tests/control_plane_ts/collaboration_turn_context.test.ts b/tests/control_plane_ts/collaboration_turn_context.test.ts index 786f879225..424585a620 100644 --- a/tests/control_plane_ts/collaboration_turn_context.test.ts +++ b/tests/control_plane_ts/collaboration_turn_context.test.ts @@ -33,6 +33,31 @@ test("sent work remains visible before return and until the requester consumes i results: [{...conclusion, status: "delivered"}]}), []); }); +test("a returned peer result stays in bounded turn attention even beyond the visible agenda", () => { + const observations = Array.from({length: 21}, (_, index) => ({ + direction: "outgoing", + request: {...request, request_id: index.toString(16).padStart(64, "0"), source_kind: "peer"}, + receipt: index === 20 ? settled : {kind: "pending", recorded_decision: null}, + todo_ids: [], evidence_unavailable: false, + results: index === 20 ? [{...conclusion, status: "delivered"}] : [], + })); + const batch = projectDelegationTurnItems({observations}) as Record; + assert.equal(batch.items.length, 20); + assert.equal(batch.items.some((row: Record) => row.request_id === "14".padStart(64, "0")), false); + assert.equal(batch.return_attention.count, 1); + assert.equal(batch.return_attention.items[0].request_id, "14".padStart(64, "0")); + assert.deepEqual(Object.keys(batch.return_attention.items[0]).sort(), [ + "agent_id", "request_id", "request_revision", "returns", "step", "unconsumed_return_count", + ]); + + const result = projectDelegationTurnContext({batches: [batch], scan_complete: true, unavailable_count: 0, + read_command: "loopx manager-inbox read", observed_at: "2026-01-01T00:00:00Z"}); + const attention = (result.unfinished_delegations as Record).return_attention; + assert.equal(attention.count, 1); + assert.equal(attention.omitted_count, 0); + assert.equal(attention.items[0].step, "consume_peer_result"); +}); + test("an original conversation retains its typed address instead of inventing a Goal receiver", () => { const recipient = {schema_version: "collaboration_original_conversation_recipient_v0", binding_id: "b".repeat(24), session_id: "original-session", identity_sha256: "c".repeat(64)}; @@ -93,7 +118,8 @@ test("a new brief retains unresolved original-route delivery without reopening r test("a bounded agenda never mistakes omitted or unavailable work for an empty queue", () => { const items = Array.from({length: 26}, (_, i) => ({...project()[0], request_id: String(i), direction: i < 23 ? "incoming" : "outgoing"})); - const result = projectDelegationTurnContext({batches: [{items, incoming_count: 23, outgoing_count: 3}], scan_complete: true, unavailable_count: 0, + const result = projectDelegationTurnContext({batches: [{items, incoming_count: 23, outgoing_count: 3, + return_attention: {count: 0, items: [], omitted_count: 0}}], scan_complete: true, unavailable_count: 0, read_command: "loopx manager-inbox read", observed_at: "2026-01-01T00:00:00Z"}); assert.equal(result.read_complete, false); const agenda = result.unfinished_delegations as Record; @@ -102,7 +128,17 @@ test("a bounded agenda never mistakes omitted or unavailable work for an empty q assert.equal(agenda.items.length, 20); assert.equal(agenda.items[1].direction, "outgoing"); assert.equal(agenda.omitted_count, 6); + assert.deepEqual(agenda.return_attention, {count: 0, items: [], omitted_count: 0}); assert.equal(agenda.scan_complete, true); const unknown = projectDelegationTurnContext({batches: [], scan_complete: true, unavailable_count: 1}).unfinished_delegations as Record; assert.equal(unknown.scan_complete, false); }); + +test("return attention aggregation rejects inconsistent bounded counts", () => { + const item = {request_id: "b".repeat(64), request_revision: 1, step: "consume_peer_result", + unconsumed_return_count: 1, returns: [{result_key: "conclusion", consumed: false}]}; + const params = {batches: [{items: [], incoming_count: 0, outgoing_count: 1, + return_attention: {count: 2, items: [item], omitted_count: 0}}], scan_complete: true, + unavailable_count: 0}; + assert.throws(() => projectDelegationTurnContext(params), /invalid delegation return attention/); +}); From b34af9fff7da9dc7f4d73e41646b6e7fc8c7f201 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Fri, 13 Nov 2026 02:19:11 +0800 Subject: [PATCH 2/2] fix(context): surface unconfirmed peer returns before wide work scopes Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../collaboration/turn_context.ts | 5 +- .../quota/progressive_context.ts | 23 ++++--- .../test_request_amendment_recovery.py | 66 ++++++++++++++++++- 3 files changed, 79 insertions(+), 15 deletions(-) diff --git a/loopx/control_plane/collaboration/turn_context.ts b/loopx/control_plane/collaboration/turn_context.ts index d512b41e8c..95a74eb15d 100644 --- a/loopx/control_plane/collaboration/turn_context.ts +++ b/loopx/control_plane/collaboration/turn_context.ts @@ -44,6 +44,7 @@ export function projectDelegationTurnItem(value: unknown): JsonObject | null { const purpose = String(control?.brief && (control.brief as JsonObject).purpose || request.purpose || "").replace(/\s+/g, " ").trim(); const todos = Array.isArray(row.todo_ids) ? row.todo_ids : []; + const visibleReturns = (step === "consume_peer_result" ? unconsumed : results).slice(-3); return { request_id: request.request_id, direction, ...(request.goal_ref ? {goal_ref: request.goal_ref} : {}), @@ -57,9 +58,9 @@ export function projectDelegationTurnItem(value: unknown): JsonObject | null { ...(receipt.remaining_work ? {remaining_work: receipt.remaining_work} : {}), ...(todos.length ? {todo_ids: todos.slice(0, 4), additional_todo_count: Math.max(0, todos.length - 4)} : {}), ...(results.length ? {pending_return_count: incomplete.length, unconsumed_return_count: unconsumed.length, - returns: results.slice(-3).map(result => ({result_key: result.result_key, + returns: visibleReturns.map(result => ({result_key: result.result_key, request_revision: result.request_revision, status: result.status, consumed: result.consumed})), - additional_return_count: Math.max(0, results.length - 3)} : {}), + additional_return_count: results.length - visibleReturns.length} : {}), }; } diff --git a/loopx/control_plane/quota/progressive_context.ts b/loopx/control_plane/quota/progressive_context.ts index 21fddb6933..f0214fdd57 100644 --- a/loopx/control_plane/quota/progressive_context.ts +++ b/loopx/control_plane/quota/progressive_context.ts @@ -996,24 +996,28 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly if (turnBindingRequired) for (const field of ["turn_binding_required", ...admissionFields]) include("interaction_contract", "cli_channel", field); - include("interaction_contract", "agent_channel", "required_reads"); - const work = ["interaction_contract", "agent_channel", "work_context"]; - include(...work, "instruction"); - if (deferredReentry) { - // Surface the owner's reopen and same-Turn guard before the long read. - for (const field of admissionFields) include("interaction_contract", "cli_channel", field); - } const channel = object(interaction.agent_channel); 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.required_reads : []; const sources = object(channel?.work_context)?.sources; const sourceRows = Array.isArray(sources) ? sources.map((source, index) => ({source, index})) : []; + const attentionSources = sourceRows.filter(({source}) => managerReturnAttention(source, requiredInboxReads)); + const requiredReadsPath = ["interaction_contract", "agent_channel", "required_reads"]; + if (!attentionSources.length) include(...requiredReadsPath); + const work = ["interaction_contract", "agent_channel", "work_context"]; + include(...work, "instruction"); + if (deferredReentry) { + // Surface the owner's reopen and same-Turn guard before the long read. + for (const field of admissionFields) include("interaction_contract", "cli_channel", field); + } // A long Manager Inbox agenda stays behind its required read. Surface only // the bounded returned-result identities before work so paging or context // recovery cannot hide that the requester has a result to consume. - for (const {source, index} of sourceRows) { - if (managerReturnAttention(source, requiredInboxReads)) { + const cueStart = pointers.length; + if (attentionSources.length) { + include(...requiredReadsPath); + for (const {index} of attentionSources) { include(...work, "sources", index, "content", "unfinished_delegations", "return_attention"); } } @@ -1022,7 +1026,6 @@ function initialContextReadGroups(context: JsonObject, decisionFields: readonly const agentId = context.agent_id ?? identityAgent; // Deliver the opted-in decision cue before work, while the candidate or // test can still change. The template and evidence stay on their owner read. - const cueStart = pointers.length; if (replanReversal) { for (const field of replanReversal.fields) include(...replanReversal.path, field); } diff --git a/tests/control_plane/test_request_amendment_recovery.py b/tests/control_plane/test_request_amendment_recovery.py index b54444fe7a..dad6fe2da9 100644 --- a/tests/control_plane/test_request_amendment_recovery.py +++ b/tests/control_plane/test_request_amendment_recovery.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import shlex import subprocess import sys @@ -114,10 +115,14 @@ def test_native_inbox_preserves_unconsumed_results_across_corrections(source_goa acknowledge(root, goal, "agent-a", rid, "adopt", "Check the corrected evidence", registry=registry, request_revision=2, request_revision_digest=revised["revision_digest"]) result = report(root, goal, "agent-a", rid, "conclusion", "Corrected checked result", registry=registry, request_revision=2) - key = result["result_key"] + newer = [result["result_key"]] + for update in ("independent-check", "review-complete"): + newer.append(report(root, goal, "agent-a", rid, "conclusion", f"Corrected result: {update}", + registry=registry, request_revision=2, update_id=update)["result_key"]) returned = read_inbox(root, registry, goal, "agent-b")["peer_returns"]["items"] - assert {item.get("result_key", "conclusion") for item in returned} == {"conclusion", key} - consume_return(root, goal, "agent-b", rid, registry=registry, result_key=key) + assert {item.get("result_key", "conclusion") for item in returned} == {"conclusion", *newer} + for key in newer: + consume_return(root, goal, "agent-b", rid, registry=registry, result_key=key) before = {p.relative_to(_root(root)): p.read_bytes() for p in _root(root).rglob("*.json")} # Consecutive native Inbox reads must retain the same original result # obligation without executing, re-delivering or consuming anything. @@ -138,6 +143,7 @@ def test_native_inbox_preserves_unconsumed_results_across_corrections(source_goa assert row["step"] == "consume_peer_result" and row["unconsumed_return_count"] == 1 old = next(item for item in row["returns"] if item["result_key"] == "conclusion") assert old["request_revision"] == 1 and old["consumed"] is False + assert old in attention["items"][0]["returns"] assert {p.relative_to(_root(root)): p.read_bytes() for p in _root(root).rglob("*.json")} == before consume_return(root, goal, "agent-b", rid, registry=registry) for actor in ("agent-a", "agent-b"): @@ -162,6 +168,60 @@ def test_request_control_refuses_wrong_actor_namespace_and_untrusted_effect_fact _control(f, "../not-a-request", "inspect") +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +def test_stateless_quota_opening_prioritizes_peer_return_over_large_write_scope(tmp_path, provider): + from loopx.capabilities.manager_context.roundtrip import report + from tests.control_plane.test_quota_settlement_cli import ( + AGENT_ID, GOAL_ID, TODO_ID, _initialize_fixture_authority, _run_cli, _write_fixture, + ) + + project, runtime, registry = _write_fixture(tmp_path) + config = json.loads(registry.read_text()) + goal = config["goals"][0] + goal["status"] = "active" + state = project / goal["state_file"] + state.write_text(state.read_text().replace("status: active-read-only", "status: active", 1)) + peer = "result-provider" + goal["coordination"]["registered_agents"].append(peer) + scope = [f"modules/component-{index:03d}/experiments/artifact-encodings/public-validation/**" + for index in range(290)] + goal["coordination"]["write_scope"] = scope + registry.write_text(json.dumps(config)) + _initialize_fixture_authority(project, runtime, provider=provider) + + rid = peer_request(runtime, registry, GOAL_ID, AGENT_ID, peer, "request-check", BRIEF)["request_id"] + acknowledge(runtime, GOAL_ID, peer, rid, "adopt", "Check the evidence", registry=registry) + report(runtime, GOAL_ID, peer, rid, "conclusion", "Evidence checked", registry=registry) + code, view = _run_cli(registry, runtime, "quota", "should-run", "--codex-app", + "--goal-id", GOAL_ID, "--agent-id", AGENT_ID, "--todo-id", TODO_ID, + "--turn-instance-id", "stateless-peer-open", "--context-delivery", "on", + "--max-page-output-bytes", "12000", cwd=project, raw_guard=False) + assert code == 0 and "first_read_command" in view, view + + def read(command): + words = shlex.split(command) + code, page = _run_cli(registry, runtime, *words[words.index("quota"):], cwd=project) + assert code == 0 and page["ok"], page + return page + + page = read(view["first_read_command"]) + first = page["content"] + assert not page["eof"] and len(json.dumps(page, ensure_ascii=False, separators=(",", ":")).encode()) + 1 <= 12000 + assert rid in first and 'return_attention"' in first + assert "required_reads" in first and "manager-inbox read" in first + chunks = [first] + while not page["eof"]: + page = read(page["next_command"]) + chunks.append(page["content"]) + selections = {item["pointer"]: item["value"] for item in json.loads("".join(chunks))["selections"]} + agenda = next(value for path, value in selections.items() if path.endswith("/unfinished_delegations") + or path.endswith("/unfinished_delegations/return_attention")) + attention = agenda.get("return_attention", agenda) + assert attention["count"] == 1 and attention["items"][0]["request_id"] == rid + assert attention["items"][0]["returns"][0]["result_key"] == "conclusion" + assert selections["/goal_boundary"]["write_scope"] == scope + + def test_real_request_reservation_goal_commit_and_terminal_linkage(source_goal, tmp_path, capsys, monkeypatch): f = source_goal rid, commit = _proposal(f, tmp_path, capsys)