diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index 8dbb6c0f50..37df78df5b 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/loopx/capabilities/manager_context/README.md @@ -205,6 +205,13 @@ requester or missing/conflicting original return route keeps coverage incomplete Reading a result does not consume it, and observing the agenda never dispatches work or acknowledges a request. +A correction updates the current brief without acknowledging earlier results. +An older revision's undelivered return or unconsumed peer result remains in the +agenda at its original `request_revision`, even after the current result is +delivered and consumed. Resolved historical returns leave this compact view; +the full Inbox/result reader retains their history. This does not certify that +an earlier result satisfies the corrected request or authorize replaying work. + Each item names its request, receiving/source Agent where recorded, current revision, related Todo ids and the remaining receipt step. Counts are request records (`count_unit=request`), not distinct tasks: follow-up requests can link diff --git a/loopx/control_plane/collaboration/turn_context.ts b/loopx/control_plane/collaboration/turn_context.ts index 5f69f64114..c0c55a9b44 100644 --- a/loopx/control_plane/collaboration/turn_context.ts +++ b/loopx/control_plane/collaboration/turn_context.ts @@ -20,8 +20,13 @@ export function projectDelegationTurnItem(value: unknown): JsonObject | null { // Cancellation of a request does not release an in-flight publication reservation. if (cancelled && control?.active_reservation == null) return null; if (!Array.isArray(row.results)) throw new EffectRuntimeRequestError("delegation results are required"); + // A correction changes the work brief, not an earlier result's delivery or + // consumption receipt. Keep unresolved returns at their original revision; + // resolved historical results remain available through the full reader. const results = row.results.map(value => requireJsonObject(value, "delegation result")) - .filter(result => result.request_revision === (control?.revision ?? 1)); + .filter(result => result.request_revision === (control?.revision ?? 1) + || result.status !== "delivered" && result.status !== "superseded" + || request.source_kind === "peer" && result.consumed !== true); const incomplete = results.filter(result => result.status !== "delivered" && result.status !== "superseded"); // Peer returns require their requester acknowledgement. An original Chat // return has a delivery receipt, not a synthetic peer-consumption receipt. @@ -52,7 +57,8 @@ export function projectDelegationTurnItem(value: unknown): JsonObject | null { ...(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, - status: result.status, consumed: result.consumed})), additional_return_count: Math.max(0, results.length - 3)} : {}), + request_revision: result.request_revision, status: result.status, consumed: result.consumed})), + additional_return_count: Math.max(0, results.length - 3)} : {}), }; } diff --git a/tests/control_plane/test_request_amendment_recovery.py b/tests/control_plane/test_request_amendment_recovery.py index c4d0241602..4b987fc171 100644 --- a/tests/control_plane/test_request_amendment_recovery.py +++ b/tests/control_plane/test_request_amendment_recovery.py @@ -97,6 +97,49 @@ def test_actual_source_request_correction_cancel_and_same_operation_recovery(sou assert cancelled["view"]["terminal_effects"] == [] +def test_native_inbox_preserves_unconsumed_results_across_corrections(source_goal): + from loopx.capabilities.manager_context.roundtrip import report + from loopx.control_plane.collaboration.peers import consume_return, read_inbox + from loopx.control_plane.collaboration.turn_context_observation import read_delegation_turn_context + from loopx.control_plane.testing.canary_harness import run_json_cli_result + + f = source_goal + root, registry, goal = f["runtime"], f["registry"], f["ref"]["goal_id"] + rid = _request(f)["request_id"] + acknowledge(root, goal, "agent-a", rid, "adopt", "Check the original evidence", registry=registry) + report(root, goal, "agent-a", rid, "conclusion", "Original checked result", registry=registry) + view = _control(f, rid, "inspect")["view"] + revised = _control(f, rid, "revise", {"operation_id": "correct-return", **_basis(view), + "brief": {**BRIEF, "purpose": "Check the corrected contract"}, "reason": "The requirement changed"})["view"] + 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"] + 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) + 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. + for _ in range(2): + code, packet = run_json_cli_result("manager-inbox", "read", "--goal-id", goal, + "--agent-id", "agent-b", + registry_path=registry, runtime_root=root) + assert code == 0, packet + agenda = packet["unfinished_delegations"] + assert agenda["outgoing_count"] == 1, agenda + row = agenda["items"][0] + assert row["request_id"] == rid and row["request_revision"] == 2 + assert row["purpose"] == "Check the corrected contract" + 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 {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"] == [] + + def test_request_control_refuses_wrong_actor_namespace_and_untrusted_effect_facts(source_goal, tmp_path): f = source_goal rid = _request(f)["request_id"] @@ -495,7 +538,10 @@ def agenda(agent): assert current["request_id"] == rid and current["request_revision"] == 2 assert current["purpose"] == "Inspect the corrected current contract" assert current["step"] == "assess_request" - assert "returns" not in current # The old answer does not settle revised work. + assert current["receipt_state"] == "pending" # The old answer does not settle revised work. + assert current["returns"] == [{"result_key": "conclusion", "request_revision": 1, + "status": "queued", "consumed": False}] + assert current["pending_return_count"] == current["unconsumed_return_count"] == 1 assert agenda("agent-b")["items"][0]["step"] == "await_peer_result" _control(f, rid, "cancel", {"operation_id": "agenda-cancel", **_basis(revised), "reason": "Stop current request"}) assert agenda("agent-a")["items"] == [] diff --git a/tests/control_plane_ts/collaboration_turn_context.test.ts b/tests/control_plane_ts/collaboration_turn_context.test.ts index 9ee3eeebab..786f879225 100644 --- a/tests/control_plane_ts/collaboration_turn_context.test.ts +++ b/tests/control_plane_ts/collaboration_turn_context.test.ts @@ -64,6 +64,33 @@ test("current revisions, cancellation and unavailable receipts retain independen assert.equal(project({receipt: {kind: "receipt_unavailable", recorded_decision: null}})[0].step, "recover_evidence"); }); +test("a correction does not consume an earlier peer result", () => { + const receipt = {...settled, request_control_view: {revision: 2, control_state: "active", active_reservation: null}}; + const peer = {direction: "outgoing", request: {...request, source_kind: "peer"}, receipt, + results: [{...conclusion, status: "delivered"}, + {...conclusion, result_key: "conclusion-00000001", request_revision: 2, status: "delivered", consumed: true}]}; + const row = project(peer)[0]; + assert.equal(row?.step, "consume_peer_result"); + assert.equal(row.request_revision, 2); + assert.equal(row.unconsumed_return_count, 1); + assert.equal((row.returns as Array>)[0].request_revision, 1); + assert.deepEqual(project({...peer, results: peer.results.map(result => ({...result, consumed: true}))}), []); +}); + +test("a new brief retains unresolved original-route delivery without reopening resolved history", () => { + const receipt = {...settled, request_control_view: {revision: 2, control_state: "active", active_reservation: null}}; + const current = {...conclusion, result_key: "conclusion-00000001", request_revision: 2, status: "delivered"}; + for (const status of ["queued", "retry_pending", "verification_required", "explicit_unverified", "admitted"]) { + const row = project({receipt, results: [{...conclusion, status}, current]})[0]; + assert.equal(row.step, "await_return_delivery"); + assert.equal(row.pending_return_count, 1); + assert.equal((row.returns as Array>)[0].request_revision, 1); + } + for (const status of ["delivered", "superseded"]) { + assert.deepEqual(project({receipt, results: [{...conclusion, status}, current]}), []); + } +}); + 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,