From ce1563fc493ea0c88f83ede6b4e09c338bf9f994 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 2 Nov 2026 12:02:14 +0530 Subject: [PATCH] fix(collaboration): retain original handoff proof failures in return readback Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../dashboard/src/data/chat-model.ts | 2 +- .../personal-workspace/collaboration-card.tsx | 6 +- .../src/features/personal-workspace/i18n.tsx | 2 + .../return-delivery-status.tsx | 10 ++- .../chat-recovery.mjs | 17 ++++ .../collaboration/return_delivery.ts | 7 +- .../manager_return_transaction.test.ts | 58 +++++++++++++ tests/test_collaboration_goal_instance.py | 82 ++++++++++++++++++- 8 files changed, 176 insertions(+), 8 deletions(-) diff --git a/apps/presentation/dashboard/src/data/chat-model.ts b/apps/presentation/dashboard/src/data/chat-model.ts index 0bafe53d7a..751887fb8f 100644 --- a/apps/presentation/dashboard/src/data/chat-model.ts +++ b/apps/presentation/dashboard/src/data/chat-model.ts @@ -101,7 +101,7 @@ export type CollaborationReadback = { }; read_status: string; decision: string; - returns: { phase: string; status: string; request_revision?: number | null }[]; + returns: { phase: string; status: string; error?: string | null; request_revision?: number | null }[]; }; export type ChatRouteCandidate = { diff --git a/apps/presentation/dashboard/src/features/personal-workspace/collaboration-card.tsx b/apps/presentation/dashboard/src/features/personal-workspace/collaboration-card.tsx index ea03a39c53..c67facf8b1 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/collaboration-card.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/collaboration-card.tsx @@ -4,6 +4,7 @@ import {useEffect, useState} from "react"; import type {RequestControlView} from "../../data/request-control"; import {RequestControl} from "./request-control"; import { useWorkspaceI18n } from "./i18n"; +import { needsOriginalDeliveryProof } from "./return-delivery-status"; import "./collaboration-card.css"; const copy = { @@ -26,7 +27,7 @@ const copy = { }; export function CollaborationCard({ request }: { request?: CollaborationReadback }) { - const { locale } = useWorkspaceI18n(); + const { locale, t } = useWorkspaceI18n(); const [observed, setObserved] = useState(null); const [unavailable, setUnavailable] = useState(false); const [editing, setEditing] = useState(false); @@ -63,7 +64,8 @@ export function CollaborationCard({ request }: { request?: CollaborationReadback {revision !== undefined ? {locale === "zh-CN" ? "请求版本" : "Request revision"} {revision} : null} {readStatus === "supplied" ? c.supplied : readStatus === "unavailable" ? c.unavailable : c.pending} {c.decision}: {decision} - {conclusion ? {conclusion.status === "delivered" ? c.delivered + {conclusion ? {needsOriginalDeliveryProof(conclusion) ? t("returnDelivery.originalHandoff") + : conclusion.status === "delivered" ? c.delivered : conclusion.status === "queued" ? c.result : c.unverified} : null}

{decisionReason && !unavailable ?

{c.reason}: {decisionReason}

: null} diff --git a/apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx b/apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx index 7cbc2dd122..1f19d0ae9d 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/i18n.tsx @@ -1624,6 +1624,7 @@ const en = { "timeline.gateHistory": "{count} historical Gates", "timeline.pending": "Working…", "returnDelivery.queued": "Return queued", + "returnDelivery.originalHandoff": "Original handoff needs verification", "returnDelivery.verifying": "Verifying delivery without resending", "returnDelivery.delivered": "Delivered to the original audience", "returnDelivery.reconciled": "Delivery verified after recovery", @@ -3285,6 +3286,7 @@ const zhCN: Record = { "timeline.gateHistory": "{count} 项历史 Gate", "timeline.pending": "正在整理…", "returnDelivery.queued": "等待回传", + "returnDelivery.originalHandoff": "原始交接待核验", "returnDelivery.verifying": "正在核验送达，不会重复发送", "returnDelivery.delivered": "已送达原受众", "returnDelivery.reconciled": "恢复后已核验送达", diff --git a/apps/presentation/dashboard/src/features/personal-workspace/return-delivery-status.tsx b/apps/presentation/dashboard/src/features/personal-workspace/return-delivery-status.tsx index 00da89919b..ca68cd99d4 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/return-delivery-status.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/return-delivery-status.tsx @@ -1,10 +1,16 @@ import { useWorkspaceI18n } from "./i18n"; import type { WorkspaceReturnDelivery } from "./personal-workspace-model"; +export function needsOriginalDeliveryProof(delivery?: Pick) { + return delivery?.error === "initial_delivery_receipt_unavailable" + && delivery.status !== "delivered" && delivery.status !== "superseded"; +} + export function ReturnDeliveryStatus({ delivery }: { delivery?: WorkspaceReturnDelivery }) { const { t, locale } = useWorkspaceI18n(); if (!delivery) return null; - const label = delivery.status === "delivered" + const needsProof = needsOriginalDeliveryProof(delivery); + const label = needsProof ? t("returnDelivery.originalHandoff") : delivery.status === "delivered" ? delivery.verification === "reconciled_after_restart" ? t("returnDelivery.reconciled") : t("returnDelivery.delivered") @@ -13,7 +19,7 @@ export function ReturnDeliveryStatus({ delivery }: { delivery?: WorkspaceReturnD : delivery.status === "explicit_unverified" ? t("returnDelivery.unverified") : t("returnDelivery.queued"); - const tone = delivery.status === "delivered" + const tone = needsProof ? "explicit_unverified" : delivery.status === "delivered" ? "delivered" : delivery.status === "verification_required" ? "verification_required" diff --git a/examples/personal-workspace-browser/chat-recovery.mjs b/examples/personal-workspace-browser/chat-recovery.mjs index 8bf00fe222..171e79b838 100644 --- a/examples/personal-workspace-browser/chat-recovery.mjs +++ b/examples/personal-workspace-browser/chat-recovery.mjs @@ -156,6 +156,16 @@ export const chatRecoveryScenario = { await page.screenshot({ path: resolve(outputDir, "collaboration-unreadable-mobile.png"), fullPage: false, animations: "disabled" }); await page.setViewportSize({ width: 1512, height: 982 }); if (api.turnRequests.length !== turnsBeforeReturn) throw new Error("Disposition readback started another model turn"); + delegatedMessage.collaboration.returns = [{ phase: "conclusion", status: "retry_pending", error: "initial_delivery_receipt_unavailable_extra" }]; + await collaboration.getByText("回复送达尚未核验", { exact: true }).waitFor({ state: "visible" }); + delegatedMessage.collaboration.returns[0].error = "initial_delivery_receipt_unavailable"; + await collaboration.getByText("原始交接待核验", { exact: true }).waitFor({ state: "visible" }); + await collaboration.evaluate((node) => node.scrollIntoView({ block: "start" })); + await page.screenshot({ path: resolve(outputDir, "collaboration-original-handoff-desktop.png"), fullPage: false, animations: "disabled" }); + await page.setViewportSize({ width: 390, height: 844 }); + if (await collaboration.evaluate((node) => node.scrollWidth > node.clientWidth + 1)) throw new Error("Original handoff explanation overflows on mobile"); + await page.screenshot({ path: resolve(outputDir, "collaboration-original-handoff-mobile.png"), fullPage: false, animations: "disabled" }); + await page.setViewportSize({ width: 1512, height: 982 }); pass("collaboration-brief", "Original conversation preserves context, constraints, inputs and receiver decision without a new turn"); const returnText = "处理结论：已核验新约束并关联现有计划，无需再次追问。"; @@ -189,13 +199,20 @@ export const chatRecoveryScenario = { await page.screenshot({ path: resolve(outputDir, "manager-automatic-conclusion.png"), fullPage: false, animations: "disabled" }); const returnedMessage = page.__loopxRuntime.messages.get(returnSessionId) .find((message) => message.message_id === "handoff.browser-fixture"); + returnedMessage.return_delivery = { schema_version: "manager_return_delivery_status_v0", phase: "conclusion", + status: "retry_pending", error: "initial_delivery_receipt_unavailable_extra" }; + await richConclusion.getByText("等待回传", { exact: true }).waitFor({ state: "visible" }); + returnedMessage.return_delivery.error = "initial_delivery_receipt_unavailable"; + await richConclusion.getByText("原始交接待核验", { exact: true }).waitFor({ state: "visible" }); returnedMessage.return_delivery = { schema_version: "manager_return_delivery_status_v0", phase: "conclusion", status: "delivered", verification: "reconciled_after_restart", + error: "initial_delivery_receipt_unavailable", }; await page.getByText("恢复后已核验送达", { exact: true }).waitFor({ state: "visible", timeout: 10_000 }); + if ((await richConclusion.innerText()).includes("原始交接待核验")) throw new Error("A stale error cannot override verified delivery"); await page.setViewportSize({ width: 390, height: 844 }); await page.getByText(returnText, { exact: true }).waitFor({ state: "visible" }); await richConclusion.scrollIntoViewIfNeeded(); diff --git a/loopx/control_plane/collaboration/return_delivery.ts b/loopx/control_plane/collaboration/return_delivery.ts index fb26ebb6b5..b7bb5a2e98 100644 --- a/loopx/control_plane/collaboration/return_delivery.ts +++ b/loopx/control_plane/collaboration/return_delivery.ts @@ -294,7 +294,9 @@ function admitReturnLifetime(command: ReturnCommand, operation: string): void { initial_delivery: facts.initial_delivery, }); if (decision.kind === "reject" || decision.kind === "omit") { - throw new EffectRuntimeRequestError("return Goal lifetime admission rejected"); + throw new EffectRuntimeRequestError("return Goal lifetime admission rejected", + decision.kind === "reject" && decision.code === "initial_delivery_unproved" + ? "initial_delivery_receipt_unavailable" : "invalid_request"); } } @@ -396,6 +398,9 @@ async function prepareReturn(command: ReturnCommand, state: JsonObject): Promise if (requireJsonObject(command.facts.route, "source route").kind === "peer") return returnResponse(); admitReturnLifetime(command, "original_return_admit"); } catch (error) { + if (error instanceof EffectRuntimeRequestError && error.code === "initial_delivery_receipt_unavailable") { + return sourceFailure(command, state, error.code); + } if (error instanceof EffectRuntimeRequestError || error instanceof SyntaxError || (error as NodeJS.ErrnoException).code) return sourceFailure(command, state); throw error; diff --git a/tests/control_plane_ts/manager_return_transaction.test.ts b/tests/control_plane_ts/manager_return_transaction.test.ts index 564cea791f..f34fc1edbd 100644 --- a/tests/control_plane_ts/manager_return_transaction.test.ts +++ b/tests/control_plane_ts/manager_return_transaction.test.ts @@ -356,3 +356,61 @@ for (const [clock, retry] of [ } finally { await f.close(); } }); } + +for (const exact of [false, true]) { + for (const accepted of [false, true]) { + test(`initial receipt conflict preserves its cause and original return (${exact}, accepted=${accepted})`, async () => { + const f = await fixture(exact, true); + try { + if (accepted) { + const plan = await f.run("prepare"); + await f.run("attempt", { ...plan, attempt }); + } + const sourcePaths = [ + f.entryPath, join(f.folder, "roundtrips", id + ".json"), + join(f.folder, "replies", id, "conclusion.json"), + ]; + const original = await Promise.all(sourcePaths.map(path => readFile(path))); + const before = accepted ? await readFile(f.statePath) : null; + const initial = f.facts.initial_delivery as JsonObject; + const conflicting = { ...f.facts, initial_delivery: { ...initial, + turn: { ...(initial.turn as JsonObject), context_handoff_receipt: { + request_id: "f".repeat(64), goal_id: "research", agent_id: "another-worker", + } }, + } }; + const result = await f.run("prepare", { facts: conflicting, + now: "2026-01-02T03:06:05.123456+00:00" }); + assert.equal(result.action, "skip", "a different receiver's receipt cannot authorize a return"); + if (exact) { + assert.equal(result.source_unavailable, true); + if (before) assert.deepEqual(await readFile(f.statePath), before); + else await assert.rejects(readFile(f.statePath)); + } else { + const held = await f.state(); + assert.equal(held.error, "initial_delivery_receipt_unavailable"); + assert.equal(held.status, accepted ? "explicit_unverified" : "retry_pending"); + if (accepted) { + assert.deepEqual(held.attempt, attempt, "an accepted write is never reset into a fresh send"); + assert.equal((await f.run("prepare", { now: "2026-01-03T03:06:05Z" })).processed, 0); + assert.deepEqual(await f.state(), held); + } else { + const bytes = await readFile(f.statePath); + assert.equal((await f.run("prepare", { now: "2026-01-02T03:06:06Z" })).processed, 0); + assert.deepEqual(await readFile(f.statePath), bytes, "unchanged source honors the original backoff"); + const restored = { ...f.facts, initial_delivery: { ...initial, + turn: { ...(initial.turn as JsonObject), context_handoff_receipt: initial.request }, + } }; + const plan = await f.run("prepare", { facts: restored, now: "2026-01-02T03:06:16Z" }); + assert.equal(plan.action, "send"); + await f.run("settle", { ...plan, facts: restored, now: "2026-01-02T03:06:16Z", + outcome: { kind: "sent", reply_verified: true, idempotency_key: attempt.provider_receipt } }); + assert.equal((await f.state()).message_id, plan.message_id); + assert.equal((await f.run("prepare", { facts: restored, now: "2026-01-03T03:06:16Z" })).processed, 0); + } + } + const retained = await Promise.all(sourcePaths.map(path => readFile(path))); + assert.deepEqual(retained, original, "diagnosis never rewrites the request, source route or result"); + } finally { await f.close(); } + }); + } +} diff --git a/tests/test_collaboration_goal_instance.py b/tests/test_collaboration_goal_instance.py index d55ad90dac..7374e471e9 100644 --- a/tests/test_collaboration_goal_instance.py +++ b/tests/test_collaboration_goal_instance.py @@ -281,7 +281,7 @@ def _external_manager_request(root: Path, registry: Path): "adopt", "Work in A.", registry=registry, - caller_goal_ref=receipt["goal_ref"], + caller_goal_ref=receipt.get("goal_ref"), ) report( root, @@ -291,11 +291,89 @@ def _external_manager_request(root: Path, registry: Path): "conclusion", "A completed its bounded review.", registry=registry, - caller_goal_ref=receipt["goal_ref"], + caller_goal_ref=receipt.get("goal_ref"), ) return store, session, receipt +@pytest.mark.parametrize("exact", [False, True]) +@pytest.mark.parametrize("external", [False, True], ids=["private", "external"]) +def test_original_receipt_conflict_readback_recovers_the_same_result(tmp_path, exact, external): + if exact: + registry = _create_source_registry(tmp_path) + else: + registry = tmp_path / "registry.json" + registry.write_text(json.dumps({"goals": [{ + "id": "delivery", "repo": str(tmp_path), "status": "active", + "coordination": {"registered_agents": ["builder", "reviewer"]}, + }]}), encoding="utf-8") + if external: + store, session, receipt = _external_manager_request(tmp_path, registry) + else: + store, session, receipt = _manager_request(tmp_path, registry) + acknowledge(tmp_path, "delivery", "builder", receipt["request_id"], "adopt", + "Checking the public input.", registry=registry, + caller_goal_ref=receipt.get("goal_ref")) + report(tmp_path, "delivery", "builder", receipt["request_id"], "conclusion", + "A completed its bounded review.", registry=registry, + caller_goal_ref=receipt.get("goal_ref")) + rid = receipt["request_id"] + original = store.turn_for_client(session["session_id"], "owner-request") + store.append_message(session["session_id"], role="user", text=original["message"], + turn_id=original["turn_id"], message_id="original-owner-request") + original_response = original["response"] + store.update_turn(session["session_id"], original["turn_id"], response={ + **original_response, "context_handoff_receipt": { + **receipt, "request_id": "f" * 64, "agent_id": "reviewer", + }, + }) + result_path = _root(tmp_path) / "replies" / rid / "conclusion.json" + result_bytes = result_path.read_bytes() + sends = [] + + def sender(route, source_session, source_turn, text): + sends.append((route, source_session, source_turn, text)) + return {"reply_verified": True, "idempotency_key": "sha256:" + "a" * 64} + + admitted_at = datetime(2026, 9, 28, tzinfo=timezone.utc) + assert drain(tmp_path, registry, store, sender, now=admitted_at) == (0 if exact else 1) + assert sends == [] + state_path = result_path.with_name("conclusion.delivery.json") + if exact: + assert not state_path.exists(), "an unproved exact request grants no receipt mutation" + else: + state = json.loads(state_path.read_text()) + assert state["status"] == "retry_pending" + assert state["error"] == "initial_delivery_receipt_unavailable" + before = state_path.read_bytes() + assert drain(tmp_path, registry, store, sender, + now=admitted_at + timedelta(seconds=1)) == 0 + assert state_path.read_bytes() == before + assert not any(m.get("origin") == "manager_followup" + for m in store.session_snapshot(session["session_id"])["messages"]) + # Only this disposable host fixture restores its original committed source. + store.update_turn(session["session_id"], original["turn_id"], response=original_response) + if not exact and not external: + snapshot = _http_snapshot(tmp_path, registry, session["session_id"]) + request = next(m["collaboration"] for m in snapshot["messages"] + if m.get("collaboration", {}).get("request_id") == rid) + assert request["returns"][0]["error"] == "initial_delivery_receipt_unavailable" + assert drain(tmp_path, registry, store, sender, + now=admitted_at + timedelta(seconds=11)) == 1 + assert len(sends) == int(external) + if external: + assert sends[0][0]["request_id"] == rid + assert sends[0][1]["session_id"] == session["session_id"] + assert sends[0][2]["turn_id"] == original["turn_id"] + assert result_path.read_bytes() == result_bytes + assert drain(tmp_path, registry, store, sender, + now=admitted_at + timedelta(days=1)) == 0 + messages = [m for m in _http_snapshot(tmp_path, registry, session["session_id"])["messages"] + if m.get("origin") == "manager_followup"] + assert len(messages) == 1 and messages[0]["turn_id"] == original["turn_id"] + assert messages[0]["return_delivery"]["status"] == "delivered" + + def test_recreated_goal_cannot_observe_or_mutate_prior_instance_requests( tmp_path: Path, ) -> None: