From f2746012447d86cacc3151bdc1dc451f69169d0a Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 27 Oct 2026 19:19:11 +0800 Subject: [PATCH] fix(collaboration): retain original conversation requests in turn agendas Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../app-conversation-and-async-inbox-v0.md | 10 +++ ...p-conversation-and-async-inbox-v0.zh-CN.md | 6 ++ .../original-project-collaboration.md | 7 ++ loopx/capabilities/manager_context/README.md | 7 ++ .../control_plane/collaboration/inbox_read.ts | 35 +++++----- .../collaboration/turn_context.ts | 2 + .../test_original_conversation_peer.py | 66 +++++++++++++++++++ .../collaboration_turn_context.test.ts | 15 +++++ 8 files changed, 132 insertions(+), 16 deletions(-) diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md index b5791c730c..17467f4946 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.md +++ b/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 diff --git a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md b/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md index fc1db79155..e3bf51ccdd 100644 --- a/docs/architecture/rfcs/app-conversation-and-async-inbox-v0.zh-CN.md +++ b/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、本人、受众、项目和 diff --git a/docs/reference/original-project-collaboration.md b/docs/reference/original-project-collaboration.md index fb03f2d6a7..16368f0525 100644 --- a/docs/reference/original-project-collaboration.md +++ b/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, diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index b073b7112d..cee86614a8 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/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 diff --git a/loopx/control_plane/collaboration/inbox_read.ts b/loopx/control_plane/collaboration/inbox_read.ts index b42faa4be0..40a3052134 100644 --- a/loopx/control_plane/collaboration/inbox_read.ts +++ b/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 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}); diff --git a/loopx/control_plane/collaboration/turn_context.ts b/loopx/control_plane/collaboration/turn_context.ts index db5ba2abf7..5f69f64114 100644 --- a/loopx/control_plane/collaboration/turn_context.ts +++ b/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, diff --git a/tests/control_plane/test_original_conversation_peer.py b/tests/control_plane/test_original_conversation_peer.py index 83c3440600..7654b4e90b 100644 --- a/tests/control_plane/test_original_conversation_peer.py +++ b/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"]) diff --git a/tests/control_plane_ts/collaboration_turn_context.test.ts b/tests/control_plane_ts/collaboration_turn_context.test.ts index dbe3df6dce..9ee3eeebab 100644 --- a/tests/control_plane_ts/collaboration_turn_context.test.ts +++ b/tests/control_plane_ts/collaboration_turn_context.test.ts @@ -33,6 +33,21 @@ test("sent work remains visible before return and until the requester consumes i results: [{...conclusion, status: "delivered"}]}), []); }); +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)}; + const peer = {direction: "outgoing", request: {request_id: request.request_id, + source_kind: "peer", source_agent_id: "requester", recipient, purpose: request.purpose}}; + const row = project(peer)[0]; + assert.deepEqual(row.recipient, recipient); + assert.equal(row.agent_id, undefined); + assert.equal(row.goal_ref, undefined); + assert.equal(row.step, "await_peer_result"); + assert.equal(project({...peer, receipt: settled, results: [conclusion]})[0].step, "consume_peer_result"); + assert.throws(() => project({...peer, request: {...peer.request, + recipient: {...recipient, goal_id: "replacement"}}}), /unsupported|unexpected|invalid/); +}); + test("current revisions, cancellation and unavailable receipts retain independent meanings", () => { const control = {revision: 2, revision_digest: "sha256:" + "b".repeat(64), brief: {purpose: "Revised requirement"}, control_state: "active", active_reservation: null};