fix(collaboration): retain original conversation requests in turn agendas

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md

@@ -999,6 +999,16 @@ saved-answer recovery reuse that revision without another model call; the first
 Turn and its result remain historical. The panel selects the current revision's
 state rather than presenting an older conclusion as its pending result.
 
+The requester's automatic Goal Turn agenda and explicit Inbox return read share
+the same typed outgoing-address reader. Original project requests retain their
+versioned conversation recipient, not a synthetic Goal/Agent. The agenda follows
+the current revision from waiting for a result to explicit requester consumption;
+reading a conclusion alone does not remove it. Changed requester identity or a
+missing/conflicting original return route leaves incomplete coverage, not an
+empty queue. This read-only source checkpoint neither acknowledges work nor
+dispatches another Turn; full briefs, live adoption and installed-provider
+qualification retain their separate acceptance.
+
 Keep one typed recipient identity across discovery, grants, request publication,
 receiver read/assessment and return. Extend the existing collaboration owner with
 an explicit original-conversation recipient alongside registered Goal/Agent

docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md

@@ -678,6 +678,12 @@ dynamic tools。File/SQLite 集成沿用生产来源、请求、FIFO、修订与
 队列写入和已存答案的恢复复用该修订，不再次调用模型；首个 Turn 与结果保留为历史。
 面板选择当前修订的状态，不把较早结论显示为当前待读取的结果。
 
+请求方的自动 Goal Turn 摘要与显式 Inbox 结果读取复用同一个类型化发出地址 reader。
+原项目请求保留版本化会话接收者，不伪造 Goal/Agent。摘要按当前修订区分等待结果和
+等待请求方显式消费；只读结论不移除该项。请求方身份变化或原返回路由缺失/冲突时，
+覆盖保持不完整，不报空队列。本源码只读检查点不确认工作或派发新 Turn；完整 brief、
+实际采用与已安装 provider 资格仍按各自验收保留。
+
 发现、授权、请求发布、接收方读取/评估和回传应使用同一个类型化接收者身份。
 在现有 collaboration owner 内，给已注册 Goal/Agent 接收者增加显式原会话接收者。
 后者从 fresh binding 和 canonical Session 解析，保留 provider、本人、受众、项目和

docs/reference/original-project-collaboration.md

@@ -108,6 +108,13 @@ checking it. The saved requester route is verified at dispatch, source control,
 requester read and consumption. Results do not automatically go to the
 recipient conversation's external audience.
 
+The requester's existing automatic Goal Turn agenda also includes these outgoing
+requests, identified by their exact original-conversation recipient. Waiting for
+an answer, a returned answer needing consumption, and a corrected current revision
+remain distinct. Reading the Inbox does not consume its result. A missing or
+conflicting original route remains incomplete evidence; the agenda does not
+start work, resend a result or replace the full request read.
+
 ## Qualification boundary
 
 Isolated File/SQLite integration uses the real source, binding, request, FIFO,

loopx/capabilities/manager_context/README.md

@@ -157,6 +157,13 @@ 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.
 
+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
+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.
+
 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/inbox_read.ts

@@ -241,6 +241,18 @@ function returnConsumed(root: string, key: string, row: JsonObject, reply: JsonO
   if (Object.entries(expected).some(([field, value]) => !equal(saved[field], value))) fail("peer consumption receipt scope mismatch");
   return true;
 }
+function outgoingEntry(source: InboxSource, operation: JsonObject): JsonObject {
+  const conversation = operation.schema_version === CONVERSATION_ENTRY_SCHEMA;
+  const requester = conversation ? requireJsonObject(operation.requester, "original requester") : null;
+  if (conversation ? !source.exact || requester?.agent_id !== source.agent || !equal(requester?.goal_ref, source.scope.caller_goal_ref)
+    : operation.goal_id !== source.goal || operation.source_agent_id !== source.agent
+      || source.exact && !equal(operation.goal_ref, source.scope.caller_goal_ref)) fail("peer return scope mismatch");
+  const row = conversation ? entry({...source, recipient: normalizeOriginalConversationRecipient(operation.recipient)}, requestId(operation.request_id))
+    : entry(source, requestId(operation.request_id), requireNonEmptyString(operation.agent_id, "peer Agent"));
+  if (row.source_agent_id !== source.agent || row.source_id !== operation.source_id
+      || conversation && !equal(row.requester, requester)) fail("peer return scope mismatch");
+  return row;
+}
 async function collectReturns(source: InboxSource): Promise<{view: JsonObject; deliveries: ReturnDelivery[]}> {
   const items: JsonObject[] = [], deliveries: ReturnDelivery[] = [];
   if (source.recipient) return {view: {items: [], has_more: false}, deliveries};
@@ -250,15 +262,9 @@ async function collectReturns(source: InboxSource): Promise<{view: JsonObject; d
     if (!name.endsWith(".json") || !BARE_SHA256_PATTERN.test(name.slice(0, -5))) continue;
     const operation = readRecord(join(folder, name));
     const conversation = operation.schema_version === CONVERSATION_ENTRY_SCHEMA;
-    const requester = conversation ? requireJsonObject(operation.requester, "original requester") : null;
-    if (conversation ? !source.exact || requester?.agent_id !== source.agent || !equal(requester?.goal_ref, source.scope.caller_goal_ref)
-      : operation.goal_id !== source.goal || operation.source_agent_id !== source.agent
-        || source.exact && !equal(operation.goal_ref, source.scope.caller_goal_ref)) fail("peer return scope mismatch");
     let row: JsonObject;
-    try { row = conversation ? entry({...source, recipient: normalizeOriginalConversationRecipient(operation.recipient)}, requestId(operation.request_id))
-      : entry(source, requestId(operation.request_id), requireNonEmptyString(operation.agent_id, "peer Agent")); }
+    try { row = outgoingEntry(source, operation); }
     catch (error) { if (missing(error)) continue; throw error; }
-    if (row.source_agent_id !== source.agent || row.source_id !== operation.source_id) fail("peer return scope mismatch");
     const replies = join(source.root, "replies", String(row.request_id));
     for (const [resultIndex, key] of resultKeys(replies).entries()) {
       if (resultIndex > 0 && resultIndex % 128 === 0) await yieldToRuntime();
@@ -346,14 +352,11 @@ async function turnContext(source: InboxSource, params: JsonObject): Promise<Jso
       let row: JsonObject;
       try {
         if (direction === "incoming") row = pendingEntry(source, join(folder, name));
-        else {
-          const operation = readRecord(join(folder, name));
-          if (operation.goal_id !== source.goal || operation.source_agent_id !== source.agent
-              || source.exact && !equal(operation.goal_ref, source.scope.caller_goal_ref)) fail("delegation operation scope mismatch");
-          row = entry(source, requestId(operation.request_id), requireNonEmptyString(operation.agent_id, "peer Agent"));
-          if (row.source_agent_id !== source.agent || row.source_id !== operation.source_id) fail("delegation request scope mismatch");
-        }
-        if (!lifecycle(source, "inbox_observe", row)) continue;
+        else row = outgoingEntry(source, readRecord(join(folder, name)));
+        if (row.schema_version === CONVERSATION_ENTRY_SCHEMA) {
+          verifyOriginalConversationReturnRoute({entry: row,
+            route: readRecord(join(source.root, "roundtrips", String(row.request_id) + ".json"))});
+        } else if (!lifecycle(source, "inbox_observe", row)) continue;
       } catch (error) { if (!recordFailure(error)) throw error; unavailable++; continue; }
       const state = receiptState(source.root, row), links = linksReceipt(source.root, row);
       let results: JsonObject[] = [], evidenceUnavailable = links.unavailable;
@@ -363,7 +366,7 @@ async function turnContext(source: InboxSource, params: JsonObject): Promise<Jso
       const receipt: JsonObject = {kind: state.kind, recorded_decision: state.recorded_decision};
       if (view) receipt.request_control_view = {control_state: view.control_state, active_reservation: view.active_reservation,
         revision: view.revision, revision_digest: view.revision_digest, brief: {purpose: (view.brief as JsonObject).purpose}};
-      observations.push({direction, request: {...Object.fromEntries(["request_id", "agent_id", "source_kind", "source_agent_id", "goal_ref"]
+      observations.push({direction, request: {...Object.fromEntries(["request_id", "agent_id", "source_kind", "source_agent_id", "goal_ref", "recipient"]
         .filter(key => Object.hasOwn(row, key)).map(key => [key, row[key]])),
       purpose: (jsonObject(row.brief)?.purpose || [...(row.message as string ?? "")].slice(0, 2000).join(""))},
       receipt, todo_ids: links.value.todo_ids ?? [], evidence_unavailable: evidenceUnavailable, results});

loopx/control_plane/collaboration/turn_context.ts

@@ -3,6 +3,7 @@ import type {JsonObject} from "../effect_program.ts";
 import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts";
 import {requireJsonObject, requireStringLiteral} from "../runtime_decode.ts";
 import {BARE_SHA256_PATTERN} from "../content_digest.ts";
+import {normalizeOriginalConversationRecipient} from "./semantic_request.ts";
 
 export function projectDelegationTurnItem(value: unknown): JsonObject | null {
   const row = requireJsonObject(value, "delegation observation");
@@ -44,6 +45,7 @@ export function projectDelegationTurnItem(value: unknown): JsonObject | null {
     request_revision: control?.revision ?? 1,
     ...(control ? {request_digest: control.revision_digest} : {}),
     ...(request.source_agent_id ? {source_agent_id: request.source_agent_id} : {}),
+    ...(request.recipient ? {recipient: normalizeOriginalConversationRecipient(request.recipient)} : {}),
     agent_id: request.agent_id,
     purpose: [...purpose].slice(0, 120).join(""), purpose_truncated: [...purpose].length > 120,
     receipt_state: kind, decision: receipt.recorded_decision ?? null, step,

tests/control_plane/test_original_conversation_peer.py

@@ -71,6 +71,15 @@ def allow(f):
     return f["service"].configure(f["session"]["session_id"], f["grant"])
 
 
+def requester_agenda(f, agent="builder"):
+    from loopx.capabilities.manager_context import turn_start_hook
+    from loopx.control_plane.capability_hooks import dispatch_turn_start_hooks
+
+    observed = dispatch_turn_start_hooks([turn_start_hook(f["root"], f["registry"], "delivery", agent)])
+    assert observed["failures"] == []
+    return observed["contexts"][0]["content"]["unfinished_delegations"]
+
+
 def test_original_project_has_explicit_grant_and_returns_through_requester_inbox(project_peer):
     f = project_peer
     sid = f["session"]["session_id"]
@@ -88,11 +97,23 @@ def test_original_project_has_explicit_grant_and_returns_through_requester_inbox
     assert saved["goal_id"] is None and saved["upstream_thread_id"] == f["session"]["upstream_thread_id"]
     peer_messages = [row for row in f["store"].messages(sid) if row.get("origin") == "peer"]
     assert [row["text"] for row in peer_messages] == [f["body"]["brief"]["purpose"]]
+    from loopx.control_plane.collaboration.inbox import _root
+    before = {path: path.read_bytes() for path in _root(f["root"]).rglob("*.json")}
+    agenda = requester_agenda(f)
+    assert {path: path.read_bytes() for path in _root(f["root"]).rglob("*.json")} == before
+    assert agenda["scan_complete"] and agenda["unavailable_count"] == 0
+    outgoing = next(row for row in agenda["items"] if row["request_id"] == first["request_id"])
+    assert outgoing["direction"] == "outgoing" and outgoing["step"] == "consume_peer_result"
+    assert outgoing["recipient"] == f["body"]["recipient"] and "agent_id" not in outgoing
+    assert first["request_id"] not in {row["request_id"] for row in requester_agenda(f, "reviewer")["items"]}
+    assert "Verified original project result." not in json.dumps(agenda)
     result = read_inbox(f["root"], f["registry"], "delivery", "builder", caller_goal_ref=f["ref"])
     assert result["peer_returns"]["items"][0]["text"] == "Verified original project result."
     assert result["peer_returns"]["items"][0]["decision"] == "adopt"
+    assert next(row for row in requester_agenda(f)["items"] if row["request_id"] == first["request_id"])["step"] == "consume_peer_result"
     assert consume_return(f["root"], "delivery", "builder", first["request_id"], registry=f["registry"], caller_goal_ref=f["ref"])["consumed"]
     assert "peer_returns" not in read_inbox(f["root"], f["registry"], "delivery", "builder", caller_goal_ref=f["ref"])
+    assert requester_agenda(f)["outgoing_count"] == 0
     again = f["service"].publish(f["body"])
     assert again["turn_id"] == first["turn_id"] and again["replayed"]
     packets = [json.loads(line) for line in f["capture"].read_text().splitlines()]
@@ -104,6 +125,33 @@ def test_original_project_has_explicit_grant_and_returns_through_requester_inbox
     assert not any("dynamicTools" in p.get("params", {}) for p in packets)
 
 
+@pytest.mark.parametrize("changed_record", ["operation", "entry"])
+def test_outgoing_agenda_refuses_a_changed_original_requester(project_peer, monkeypatch, changed_record):
+    from loopx.control_plane.collaboration.inbox import _root
+    from loopx.control_plane.collaboration.turn_context_observation import read_delegation_turn_context
+
+    f = project_peer
+    allow(f)
+    monkeypatch.setattr(f["runtime"], "resume_session_queue", lambda **_: None)
+    published = f["service"].publish(f["body"])
+    folder = _root(f["root"]) / ("peer-operations" if changed_record == "operation" else "entries")
+    paths = [path for path in folder.glob("*/*.json")
+        if json.loads(path.read_text()).get("request_id") == published["request_id"]]
+    assert len(paths) == 1
+    path = paths[0]
+    saved = json.loads(path.read_text())
+    saved["requester"] = {**saved["requester"], "goal_ref": {**f["ref"], "goal_instance_id": "ginst_" + "b" * 32}}
+    path.write_text(json.dumps(saved))
+    before = {path: path.read_bytes() for path in _root(f["root"]).rglob("*.json")}
+    agenda = read_delegation_turn_context(f["root"], f["registry"], "delivery", "builder")["unfinished_delegations"]
+    assert not agenda["scan_complete"] and agenda["unavailable_count"] == 1
+    assert published["request_id"] not in {row["request_id"] for row in agenda["items"]}
+    with pytest.raises(ValueError, match="scope mismatch"):
+        read_inbox(f["root"], f["registry"], "delivery", "builder", caller_goal_ref=f["ref"])
+    assert {path: path.read_bytes() for path in _root(f["root"]).rglob("*.json")} == before
+    assert not any(json.loads(line).get("method") == "turn/start" for line in f["capture"].read_text().splitlines())
+
+
 def test_queue_recovery_and_concurrent_retry_preserve_fifo_and_original_session(project_peer, monkeypatch):
     f = project_peer
     sid = f["session"]["session_id"]
@@ -123,6 +171,10 @@ def test_queue_recovery_and_concurrent_retry_preserve_fifo_and_original_session(
     queued = f["store"].queued_turns(sid)
     assert [row["turn_id"] for row in queued] == [human["turn_id"], results[0]["turn_id"]]
     assert [row["origin"] for row in queued] == ["web", "peer"]
+    agenda = requester_agenda(f)
+    assert agenda["scan_complete"] and agenda["unavailable_count"] == 0
+    outgoing = next(row for row in agenda["items"] if row["request_id"] == results[0]["request_id"])
+    assert outgoing["step"] == "await_peer_result" and outgoing["recipient"] == f["body"]["recipient"]
     for change in [{"brief": _brief("Changed payload")}, {"operation_id": "project-consultation", "parent_request_id": "b" * 64}]:
         with pytest.raises((ValueError, FileNotFoundError)):
             service.publish({**f["body"], **change})
@@ -217,6 +269,7 @@ def test_completed_request_correction_survives_queue_loss_restart_and_retry(proj
     assert original["status"] == "completed"
     assert read_inbox(f["root"], f["registry"], "delivery", "builder", caller_goal_ref=f["ref"])["peer_returns"]["items"][0]["request_revision"] == 1
     assert consume_return(f["root"], "delivery", "builder", first["request_id"], registry=f["registry"], caller_goal_ref=f["ref"])["consumed"]
+    assert requester_agenda(f)["outgoing_count"] == 0
     retained = {path: path.read_bytes() for path in result_paths(_root(f["root"]) / "replies" / first["request_id"])}
     view = f["service"]._view(f["service"]._entry(first["request_id"], f["body"]["recipient"]))
     control = {"requester": f["body"]["requester"], "recipient": f["body"]["recipient"],
@@ -239,6 +292,11 @@ def test_completed_request_correction_survives_queue_loss_restart_and_retry(proj
     assert current["queue_status"] == "queued"
     assert current["request_control_view"]["assessment"] is None
     assert [reply["request_revision"] for reply in current["returns"]] == [1]
+    agenda = requester_agenda(f)
+    assert agenda["scan_complete"] and agenda["unavailable_count"] == 0
+    corrected = next(row for row in agenda["items"] if row["request_id"] == first["request_id"])
+    assert corrected["request_revision"] == 2 and corrected["step"] == "await_peer_result"
+    assert corrected["purpose"] == "Verify corrected project evidence"
     f["runtime"].close()
     restarted = ChatRuntimeController(store=f["store"], codex_bin=str(f["fake"]), project_contexts=f["runtime"].project_contexts,
         registry_path=f["registry"])
@@ -261,6 +319,10 @@ def lose_corrected_result(path, value):
         returned = read_inbox(f["root"], f["registry"], "delivery", "builder", caller_goal_ref=f["ref"])["peer_returns"]["items"]
         assert [reply["request_revision"] for reply in returned] == [2]
         assert returned[0]["decision"] == "adopt"
+        assert next(row for row in requester_agenda(f)["items"] if row["request_id"] == first["request_id"])["step"] == "consume_peer_result"
+        assert consume_return(f["root"], "delivery", "builder", first["request_id"], registry=f["registry"],
+            caller_goal_ref=f["ref"], result_key=returned[0].get("result_key", "conclusion"))["consumed"]
+        assert requester_agenda(f)["outgoing_count"] == 0
         service.control(control)
         service.publish(f["body"])
         packets = [json.loads(line) for line in f["capture"].read_text().splitlines()]
@@ -396,10 +458,14 @@ def test_original_return_route_is_checked_before_model_input_and_requester_read(
     path = _root(f["root"]) / "roundtrips" / (first["request_id"] + ".json")
     original = _read(path)
     _write(path, {**original, "source_id": "peer:another-source"})
+    agenda = requester_agenda(f)
+    assert not agenda["scan_complete"] and agenda["unavailable_count"] == 1
+    assert first["request_id"] not in {row["request_id"] for row in agenda["items"]}
     with pytest.raises(ValueError, match="return route identity"):
         f["service"].validate_turn(sid, turn)
     assert not any(json.loads(line).get("method") == "turn/start" for line in f["capture"].read_text().splitlines())
     _write(path, original)
+    assert requester_agenda(f)["scan_complete"]
     f["service"].validate_turn(sid, turn)
     monkeypatch.setattr(f["runtime"], "resume_session_queue", original_resume)
     f["service"].publish(f["body"])