fix(collaboration): preserve outstanding results across request corrections

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

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)} : {}),
   };
 }
 

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"] == []

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<Record<string, unknown>>)[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<Record<string, unknown>>)[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,