+ {row.agenda?.remaining_work ?
+ {zh ? "仍需完成" : "Remaining work"}
{row.agenda.remaining_work.summary}
+ {zh ? "继续条件：" : "Continue when: "}{row.agenda.remaining_work.resume_when}
+ : null}
{row.decision_reason ?
{row.decision_reason}
: null}
{row.brief ? <>
{row.brief.context}
{row.brief.constraints.length ? <>
{zh ? "约束" : "Constraints"}
{row.brief.constraints.map((value, index) => - {value}
)}
: null}
diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py
index a5bb1a07a5..2f9cbb505d 100644
--- a/loopx/capabilities/manager_context/roundtrip.py
+++ b/loopx/capabilities/manager_context/roundtrip.py
@@ -7,6 +7,7 @@
from __future__ import annotations
import logging
+import json
import threading
from itertools import islice
from contextlib import contextmanager, ExitStack
@@ -179,6 +180,7 @@ def report(
attachment_refs=None,
workspace=None,
request_revision=None,
+ remaining_work=None,
):
"""Chat audience adapter; the shared Inbox owns result validation/persistence."""
if scope is None and registry is not None:
@@ -204,13 +206,15 @@ def report(
attachment_refs=attachment_refs,
workspace=workspace,
request_revision=request_revision,
+ remaining_work=remaining_work,
)
row = _entry(root, goal_id, agent_id, request_id, scope=scope)
route = _route(root, row, scope=scope)
if route.get("kind") == "peer" and phase != "conclusion":
raise ValueError("peer replies require a conclusion; report a concrete result or blocker")
+ audience_text = text + ("\n" + json.dumps(remaining_work, ensure_ascii=False) if remaining_work is not None else "")
if (route["channel_id"] != "peer" and not conversation_scope(route)["private_conversation"]
- and not scan_public_boundary_text(text)["ok"]):
+ and not scan_public_boundary_text(audience_text)["ok"]):
raise ValueError("reply contains private boundary material; write an audience-safe conclusion")
from ...control_plane.collaboration.inbox import record_result
attachments = None
@@ -234,6 +238,7 @@ def report(
update_id=update_id,
attachments=attachments,
request_revision=request_revision,
+ remaining_work=remaining_work,
),
"status": (
"queued_for_requester"
diff --git a/loopx/cli_commands/manager_inbox.py b/loopx/cli_commands/manager_inbox.py
index 2652620293..cbd6564818 100644
--- a/loopx/cli_commands/manager_inbox.py
+++ b/loopx/cli_commands/manager_inbox.py
@@ -79,6 +79,8 @@ def register_manager_inbox(subparsers, add_format):
"--phase", choices=("decision", "conclusion"), default="conclusion"
)
parser.add_argument("--reply-text")
+ parser.add_argument("--remaining-work", help="For report: accepted work still owed after this partial conclusion; pair with --resume-when. No Todo or execution is created.")
+ parser.add_argument("--resume-when", help="For report: concrete continuation condition for --remaining-work; descriptive only, not a scheduler rule.")
parser.add_argument("--attachment-ref", action="append", default=[],
help="For report: explicit relative file in the registered Goal workspace; bound-owner App only, at most four files.")
parser.add_argument("--update-id", help="For report: stable identity for a later conclusion; retry the same id and text.")
@@ -95,6 +97,11 @@ def register_manager_inbox(subparsers, add_format):
def handle_manager_inbox(args, registry_path, runtime_root):
try:
+ remainder = getattr(args, "remaining_work", None)
+ resume = getattr(args, "resume_when", None)
+ if remainder is not None or resume is not None:
+ if args.manager_inbox_action != "report" or remainder is None or resume is None:
+ raise ValueError("--remaining-work and --resume-when require report and each other")
if args.manager_inbox_action == "original-conversation":
from ..control_plane.collaboration.peers import original_conversation_request
if not args.conversation_operation or not args.conversation_file:
@@ -295,6 +302,7 @@ def handle_manager_inbox(args, registry_path, runtime_root):
update_id=getattr(args, "update_id", None),
attachment_refs=getattr(args, "attachment_ref", None),
request_revision=getattr(args, "request_revision", None),
+ remaining_work={"summary": remainder, "resume_when": resume} if remainder is not None else None,
workspace=Path.cwd(),
registry=registry_path,
)
diff --git a/loopx/collaboration_mcp.py b/loopx/collaboration_mcp.py
index 7005b93eba..cd38bfc701 100644
--- a/loopx/collaboration_mcp.py
+++ b/loopx/collaboration_mcp.py
@@ -322,13 +322,18 @@ def request_peer(
"For a bound-owner App return, attachment_refs explicitly attaches up to four "
"relative Goal-workspace files; a filename in text does not attach a file. "
"Neither a blocker nor a returned result certifies completion of the work. "
+ "For a partial result, supply remaining_work with summary and resume_when; "
+ "it stays visible for the receiver without creating a Todo or wakeup. "
+ conversation_answer_instruction()
))
def return_result(request_id: str, text: str, update_id: str | None = None,
- attachment_refs: list[str] | None = None, request_revision: int | None = None) -> dict:
+ attachment_refs: list[str] | None = None, request_revision: int | None = None,
+ remaining_work: dict[str, str] | None = None) -> dict:
"""Return a conclusion to the original requester. For a later changed fact,
append an update with a stable update_id; retry with the same id and text.
Neither a blocker nor a returned result certifies completion of the work.
+ For a partial result, remaining_work contains summary and resume_when;
+ the receiving Agent still owns this remainder. It creates no Todo or wakeup.
"""
check_scope()
# The host adapter selects Chat/Lark transport; the shared collaboration
@@ -345,6 +350,7 @@ def return_result(request_id: str, text: str, update_id: str | None = None,
update_id=update_id,
attachment_refs=attachment_refs,
request_revision=request_revision,
+ remaining_work=remaining_work,
workspace=workspace,
registry=registry,
caller_goal_ref=caller_goal_ref,
diff --git a/loopx/control_plane/collaboration/inbox.py b/loopx/control_plane/collaboration/inbox.py
index 55d0912344..6a061ca544 100644
--- a/loopx/control_plane/collaboration/inbox.py
+++ b/loopx/control_plane/collaboration/inbox.py
@@ -435,6 +435,7 @@ def record_result(
update_id: str | None = None,
attachments: list[dict[str, Any]] | None = None,
request_revision: int | None = None,
+ remaining_work: dict[str, str] | None = None,
):
"""Persist a receiver conclusion; the transport adapter validates its audience."""
request_id = row["request_id"]
@@ -462,6 +463,7 @@ def record_result(
"decision": decision.get("decision"), "update_id": update_id,
"request_revision": request_revision, "legacy_assessment": decision or None,
"attachments": attachments,
+ "remaining_work": remaining_work,
"results": observations,
})
except EffectRuntimeRejected as exc:
@@ -499,6 +501,7 @@ def inspect_result_identity(reply, row, path):
"request": row, "phase": reply.get("phase"), "text": reply.get("text"),
"decision": reply.get("decision"), "update_id": reply.get("update_id"),
"attachments": reply.get("attachments"),
+ "remaining_work": reply.get("remaining_work"),
"request_revision": reply.get("request_revision"), "legacy_assessment": legacy.get("value"),
"results": observations,
})
@@ -521,7 +524,7 @@ def _result_observation(key, value):
"value": {field: value[field] for field in (
"request_id", "goal_id", "agent_id", "source_id", "goal_ref", "phase",
"decision", "result_key", "previous_result_key", "update_id", "attachments",
- "request_revision", "request_revision_digest", "recipient", "requester",
+ "request_revision", "request_revision_digest", "recipient", "requester", "remaining_work",
) if field in value},
"text_chars": len(text),
"text_sha256": hashlib.sha256(text.encode()).hexdigest(),
diff --git a/loopx/control_plane/collaboration/inbox_read.ts b/loopx/control_plane/collaboration/inbox_read.ts
index 40a3052134..411efe7b24 100644
--- a/loopx/control_plane/collaboration/inbox_read.ts
+++ b/loopx/control_plane/collaboration/inbox_read.ts
@@ -148,7 +148,7 @@ function resultObservation(key: string, reply: JsonObject): JsonObject {
if (/[\uD800-\uDFFF]/u.test(text)) fail("invalid reply text encoding");
return {key, value: Object.fromEntries(["request_id", "goal_id", "agent_id", "source_id", "goal_ref", "phase",
"decision", "result_key", "previous_result_key", "update_id", "attachments", "request_revision",
- "request_revision_digest", "recipient", "requester"].filter(field => Object.hasOwn(reply, field)).map(field => [field, reply[field]])),
+ "request_revision_digest", "recipient", "requester", "remaining_work"].filter(field => Object.hasOwn(reply, field)).map(field => [field, reply[field]])),
text_chars: [...text].length, text_sha256: createHash("sha256").update(text).digest("hex")};
}
function receiptState(root: string, row: JsonObject): JsonObject {
@@ -160,11 +160,17 @@ function receiptState(root: string, row: JsonObject): JsonObject {
if (Object.hasOwn(row, "request_control")) {
observation.request = row;
observation.read = observe(join(root, "reads", id + ".json"));
- observation.results = resultKeys(folder).map(key => {
- try { return resultObservation(key, readRecord(join(folder, key + ".json"))); }
- catch (error) { if (!recordFailure(error)) throw error; return {unavailable: true}; }
- });
}
+ // Later conclusions may retain an explicit remainder, also on legacy requests.
+ // Read their existing immutable chain; never infer scope from reply prose or Todos.
+ observation.results = resultKeys(folder).map(key => {
+ try {
+ const initial = jsonObject(observation.conclusion);
+ return resultObservation(key, key === "conclusion" && initial?.state === "read"
+ ? requireJsonObject(initial.value, "initial conclusion") : readRecord(join(folder, key + ".json")));
+ }
+ catch (error) { if (!recordFailure(error)) throw error; return {unavailable: true}; }
+ });
return (inspectCollaborationInboxReceipts({observations: [observation]}).items as JsonObject[])[0];
}
function inboxItem(row: JsonObject, state: JsonObject): JsonObject {
@@ -175,6 +181,7 @@ function inboxItem(row: JsonObject, state: JsonObject): JsonObject {
request_revision: view.revision, request_revision_digest: view.revision_digest,
source_request: {request_id: row.request_id, request_revision: view.revision, request_digest: view.revision_digest}});
item.recorded_decision = state.recorded_decision;
+ if (state.remaining_work) item.remaining_work = state.remaining_work;
if (state.recorded_decision !== null) Object.assign(item, {receiver_decision_recorded: true,
next_action: "Review this request's recorded decision and linked work before returning its actual result. A recorded adoption is not a completed task; preserve the current owner and do not reprioritize unrelated work."});
if ((state.warnings as unknown[]).length) Object.assign(item, {warnings: state.warnings,
@@ -227,6 +234,7 @@ function inspectResult(root: string, row: JsonObject, key: string, reply: JsonOb
? observe(join(root, "decisions", String(row.request_id) + ".json")) : {};
const inspected = planCollaborationResult({request: row, phase: reply.phase, text: reply.text,
decision: reply.decision, update_id: reply.update_id, attachments: reply.attachments,
+ remaining_work: reply.remaining_work,
request_revision: reply.request_revision, legacy_assessment: legacy.value, results});
return (reply.result_key ?? reply.phase) === key ? inspected : null;
} catch (error) { if (!recordFailure(error)) throw error; return null; }
@@ -281,7 +289,8 @@ async function collectReturns(source: InboxSource): Promise<{view: JsonObject; d
brief: inspected.request_brief ?? row.brief,
...(Object.hasOwn(inspected, "request_revision") ? {request_revision: inspected.request_revision,
current_request_revision: inspected.current_request_revision} : {}),
- text: reply.text, decision: reply.decision, created_at: reply.created_at});
+ text: reply.text, ...(reply.remaining_work ? {remaining_work: reply.remaining_work} : {}),
+ decision: reply.decision, created_at: reply.created_at});
if (items.length > 20) break;
deliveries.push({key, request: Object.fromEntries(["request_id", "goal_id", "agent_id", "goal_ref", "recipient", "requester"]
.filter(field => Object.hasOwn(row, field)).map(field => [field, row[field]])),
@@ -364,6 +373,7 @@ async function turnContext(source: InboxSource, params: JsonObject): Promise
@@ -33,10 +33,10 @@ function receiverFollowthrough(value: unknown): JsonObject {
const evidenceUnavailable = requireBoolean(params.evidence_unavailable, "receiver evidence availability");
// Only explicit request links can establish its work. A busy Agent, a
// matching title, and an unrelated rolling review task are not such links.
- const unresolved = work.some((todo) => todo.status !== "done");
const step = kind === "receipt_unavailable" || evidenceUnavailable ? "recover_evidence"
+ : kind === "settled" ? "inspect_recorded_result"
: decision === null ? "assess_request"
- : decision === "adopt" && (unresolved || work.length === 0) ? "review_request_work"
+ : decision === "adopt" ? "review_request_work"
: "return_answer";
return {
step,
@@ -86,14 +86,18 @@ function inspect(value: unknown): JsonObject {
card_read_status: readUnavailable ? "unavailable" : readAt !== null ? "supplied" : "pending",
card_decision: recorded ?? (view.assessment_readback === "unavailable" ? "unavailable" : "pending")};
const results = collaborationResultObservations(request, row.results, recorded, legacy);
- const answered = results.some(result => result.value.phase === "conclusion"
- && (result.value.request_revision ?? 1) === view!.revision);
+ const latest = results.filter(result => result.value.phase === "conclusion"
+ && (result.value.request_revision ?? 1) === view!.revision)
+ .sort((a, b) => a.key.localeCompare(b.key)).at(-1);
+ const remaining = latest?.remainingWork;
+ const answered = latest !== undefined && !remaining;
const warnings: Warning[] = view.assessment_readback === "unavailable" ? ["decision_unreadable_or_conflicting"] : [];
const kind: InboxState = warnings.length ? "receipt_unavailable"
: view.control_state === "cancelled" ? "settled"
: view.active_reservation !== null ? "awaiting_conclusion"
: !recorded ? "pending" : answered ? "settled" : "awaiting_conclusion";
return {kind, recorded_decision: recorded, warnings, request_control_view: view, readback,
+ ...(remaining ? {remaining_work: remaining} : {}),
result_revisions: results.map(result => ({result_key: result.key, request_revision: result.value.request_revision ?? 1}))};
} catch (error) {
if (!(error instanceof EffectRuntimeRequestError)) throw error;
@@ -105,6 +109,7 @@ function inspect(value: unknown): JsonObject {
const recorded = decision.state === "read" && sameRequest(decision.value, request, false)
? receiverDecision(decision.value.decision) : null;
const warnings: Warning[] = [];
+ let remaining: JsonObject | null = null;
if (decision.state !== "absent" && !recorded) warnings.push("decision_unreadable_or_conflicting");
if (conclusion.state !== "absent") {
const result = conclusion.state === "read" ? conclusion.value : null;
@@ -116,12 +121,26 @@ function inspect(value: unknown): JsonObject {
}
if (decision.state === "absent") warnings.push("decision_missing_for_conclusion");
}
+ try {
+ // An explicit partial result stays owed even after its reply is delivered.
+ // Later append-only results supersede its scope, never its stored receipt.
+ if (row.results !== undefined) {
+ const latest = collaborationResultObservations(request, row.results, recorded)
+ .filter(result => result.value.phase === "conclusion")
+ .sort((a, b) => a.key.localeCompare(b.key)).at(-1);
+ remaining = latest?.remainingWork ?? null;
+ } else if (conclusion.state === "read" && conclusion.value) remaining = resultRemainingWork(conclusion.value.remaining_work);
+ } catch (error) {
+ if (!(error instanceof EffectRuntimeRequestError)) throw error;
+ if (!warnings.includes("conclusion_unreadable_or_conflicting")) warnings.push("conclusion_unreadable_or_conflicting");
+ }
// This is receipt/readback state only. Settled means a receiver reply exists
// (or a legacy ACK owes no return), never accepted work or verified delivery.
const kind: InboxState = warnings.length ? "receipt_unavailable"
: !recorded ? "pending"
+ : remaining ? "awaiting_conclusion"
: conclusion.state === "read" || !routePresent ? "settled" : "awaiting_conclusion";
- return { kind, recorded_decision: recorded, warnings };
+ return { kind, recorded_decision: recorded, warnings, ...(remaining ? {remaining_work: remaining} : {}) };
}
/** One bounded read-model batch; no store writes, execution or authority. */
diff --git a/loopx/control_plane/collaboration/peers.py b/loopx/control_plane/collaboration/peers.py
index 7d31fdca5e..7c15054291 100644
--- a/loopx/control_plane/collaboration/peers.py
+++ b/loopx/control_plane/collaboration/peers.py
@@ -654,6 +654,7 @@ def return_result(
scope=None,
update_id=None,
request_revision=None,
+ remaining_work=None,
):
"""Route by the saved recipient, never by an Agent's coordinator role."""
if scope is None and registry is not None:
@@ -672,6 +673,7 @@ def return_result(
scope=goal_scope,
update_id=update_id,
request_revision=request_revision,
+ remaining_work=remaining_work,
)
row = _entry(root, goal_id, agent_id, request_id, scope=scope)
if row.get("source_kind") != "peer":
@@ -702,6 +704,7 @@ def return_result(
route=route,
update_id=update_id,
request_revision=request_revision,
+ remaining_work=remaining_work,
),
"status": "queued_for_requester",
}
diff --git a/loopx/control_plane/collaboration/result_publication.ts b/loopx/control_plane/collaboration/result_publication.ts
index 1305b550ca..2a4175b2cf 100644
--- a/loopx/control_plane/collaboration/result_publication.ts
+++ b/loopx/control_plane/collaboration/result_publication.ts
@@ -17,6 +17,31 @@ function decisionKey(revision: number): string {
}
const UPDATE_ID = /^[A-Za-z0-9][A-Za-z0-9_.:-]{0,119}$/;
+/** Receiver-authored scope on an immutable result, not a completion verdict or scheduler. */
+export function resultRemainingWork(value: unknown): JsonObject | null {
+ if (value === undefined || value === null) return null;
+ const work = requireJsonObject(value, "remaining work");
+ if (Object.keys(work).sort().join(",") !== "resume_when,summary") {
+ throw new EffectRuntimeRequestError("remaining work needs a summary and resume_when only");
+ }
+ const result: JsonObject = {};
+ for (const key of ["summary", "resume_when"]) {
+ const text = work[key];
+ if (typeof text !== "string" || !text.trim() || [...text].length > 600
+ || /[\uD800-\uDFFF]/u.test(text)) {
+ throw new EffectRuntimeRequestError("remaining work fields must contain 1-600 Unicode characters");
+ }
+ result[key] = text.trim();
+ }
+ return result;
+}
+
+/** The original conversation and peer return read the same bounded remainder. */
+export function collaborationResultText(reply: JsonObject): string {
+ const work = resultRemainingWork(reply.remaining_work);
+ return String(reply.text) + (work ? `\n\n仍需完成 / Remaining work: ${work.summary}\n继续条件 / Continue when: ${work.resume_when}` : "");
+}
+
export function resultAttachmentRefs(params: JsonObject): JsonObject {
if (params.workspace_current === false) {
throw new EffectRuntimeRequestError("result file workspace Goal instance changed");
@@ -63,6 +88,10 @@ export function collaborationResultObservations(request: JsonObject, observation
const results = observations.map((item) => {
const record = requireJsonObject(item, "result observation");
const value = requireJsonObject(record.value, "result receipt");
+ const remainingWork = resultRemainingWork(value.remaining_work);
+ if (remainingWork && value.phase !== "conclusion") {
+ throw new EffectRuntimeRequestError("remaining work requires a conclusion");
+ }
const historical = (request.request_control !== undefined || value.request_revision !== undefined) ? requestRevisionReceiptContext(request, value.request_revision, legacyAssessment) : null;
if (typeof record.key !== "string" || !sameRequest(value, request, true)
|| value.decision !== (historical === null ? decision : (historical.assessment as JsonObject | null)?.decision)
@@ -73,13 +102,16 @@ export function collaborationResultObservations(request: JsonObject, observation
throw new EffectRuntimeRequestError("result receipt identity or content conflict");
}
return { key: record.key, value, textSha256: record.text_sha256,
- attachments: resultAttachments(value.attachments) };
+ attachments: resultAttachments(value.attachments), remainingWork };
});
if (new Set(results.map((r) => r.key)).size !== results.length) {
throw new EffectRuntimeRequestError("duplicate result observation");
}
let previous = "conclusion";
const updates = results.filter((r) => UPDATE_KEY.test(r.key)).sort((a, b) => a.key.localeCompare(b.key));
+ if (updates.length && !results.some(result => result.key === "conclusion")) {
+ throw new EffectRuntimeRequestError("initial conclusion is required");
+ }
const ids = new Set();
for (const [index, receipt] of updates.entries()) {
if (receipt.key !== `conclusion-${String(index + 1).padStart(8, "0")}`
@@ -112,6 +144,10 @@ export function collaborationResultObservations(request: JsonObject, observation
export function planCollaborationResult(params: JsonObject): JsonObject {
const request = requireJsonObject(params.request, "result request");
const phase = requireStringLiteral(params.phase, ["decision", "conclusion"], "result phase");
+ const remainingWork = resultRemainingWork(params.remaining_work);
+ if (remainingWork && phase !== "conclusion") {
+ throw new EffectRuntimeRequestError("remaining work requires a conclusion");
+ }
const legacyOriginal = request.schema_version === "loopx_manager_context_entry_v1"
&& request.request_control === undefined;
if (legacyOriginal && params.request_revision != null && params.request_revision !== 1) {
@@ -127,11 +163,13 @@ export function planCollaborationResult(params: JsonObject): JsonObject {
if (typeof params.text !== "string" || !params.text.trim() || Array.from(params.text).length > 20000) {
throw new EffectRuntimeRequestError("bounded reply text is required");
}
+ if ([...collaborationResultText({text: params.text.trim(), remaining_work: remainingWork})].length > 20000) {
+ throw new EffectRuntimeRequestError("reply and remaining work must fit the existing 20000-character result budget");
+ }
const results = collaborationResultObservations(request, params.results, decision, params.legacy_assessment ?? null);
const updates = results.filter((r) => UPDATE_KEY.test(r.key)).sort((a, b) => a.key.localeCompare(b.key));
const previous = updates.at(-1)?.key ?? "conclusion";
const conclusion = results.find((r) => r.key === "conclusion");
- if (updates.length && !conclusion) throw new EffectRuntimeRequestError("initial conclusion is required");
const revision = context?.revision ?? 1;
const primaryUpdate = `request-revision-${revision}`;
const initialForRevision = conclusion && (conclusion.value.request_revision ?? 1) === revision;
@@ -150,12 +188,14 @@ export function planCollaborationResult(params: JsonObject): JsonObject {
Object.assign(value, { phase, decision, text: params.text.trim() });
if (context !== null) Object.assign(value, {request_revision: context.revision, request_revision_digest: context.revision_digest});
if (attachments.length) value.attachments = attachments;
+ if (remainingWork) value.remaining_work = remainingWork;
const key = old?.key ?? (isUpdate ? `conclusion-${String(updates.length + 1).padStart(8, "0")}` : initialKey);
if (phase === "decision" && key !== phase) value.result_key = key;
if (isUpdate) Object.assign(value, { result_key: key, update_id: updateId, previous_result_key: old?.value.previous_result_key ?? previous });
if (old && (old.textSha256 !== createHash("sha256").update(String(value.text)).digest("hex") || old.value.phase !== phase
|| (old.value.request_revision ?? 1) !== revision
- || JSON.stringify(old.attachments) !== JSON.stringify(attachments))) {
+ || JSON.stringify(old.attachments) !== JSON.stringify(attachments)
+ || JSON.stringify(old.remainingWork) !== JSON.stringify(remainingWork))) {
throw new EffectRuntimeRequestError("reply already committed; conflicting replacement rejected; use a stable update_id for a later conclusion");
}
const interimSuperseded = phase === "decision" && results.some(r =>
diff --git a/loopx/control_plane/collaboration/return_delivery.ts b/loopx/control_plane/collaboration/return_delivery.ts
index b7bb5a2e98..91fa6bdd9e 100644
--- a/loopx/control_plane/collaboration/return_delivery.ts
+++ b/loopx/control_plane/collaboration/return_delivery.ts
@@ -16,7 +16,7 @@ import { atomicWriteJson, withFileMutationLock } from "../effect_runtime_io.ts";
import { parseTodoTimestampMicros } from "../runtime_timestamp.ts";
import { decideCollaborationLifecycle } from "./goal_instance_lifecycle.ts";
import { resolveConversationScope } from "./conversation_scope.ts";
-import { planCollaborationResult, collaborationResultDeliveryReady } from "./result_publication.ts";
+import { planCollaborationResult, collaborationResultDeliveryReady, collaborationResultText } from "./result_publication.ts";
export const MANAGER_RETURN_DELIVERY_ATTEMPT_SCHEMA =
"manager_return_delivery_attempt_v0";
@@ -340,6 +340,7 @@ async function inspectReturnPublication(command: ReturnCommand): Promise<{ reply
catch { /* The shared historical-revision decoder rejects unavailable assessment evidence. */ }
const inspected = planCollaborationResult({ request: entry, phase: reply.phase, text: reply.text,
decision: reply.decision, update_id: reply.update_id, attachments: reply.attachments,
+ remaining_work: reply.remaining_work,
request_revision: reply.request_revision, legacy_assessment: legacy, results });
if ((reply.result_key ?? reply.phase) !== command.key || entry.request_id !== command.request.request_id) {
throw new EffectRuntimeRequestError("return reply identity mismatch");
@@ -435,7 +436,7 @@ async function prepareReturn(command: ReturnCommand, state: JsonObject): Promise
const verify = priorStatus === "verification_required" || command.exact && state.attempt != null;
const prefix = command.exact ? (reply.phase === "conclusion" ? "处理结论" : "处理进展")
: (reply.phase === "conclusion" ? "协作回复" : "协作进展");
- const text = `${prefix} · ${reply.agent_id} · 委托 ${String(reply.request_id).slice(0, 8)}\n\n${reply.text}`;
+ const text = `${prefix} · ${reply.agent_id} · 委托 ${String(reply.request_id).slice(0, 8)}\n\n${collaborationResultText(reply)}`;
return returnResponse({ action: privateConversation ? "private" : verify ? "verify" : "send",
text, message_id: returnMessageId(command), attempt: verify ? savedAttempt(state.attempt) : null, attachments: reply.attachments ?? [],
diff --git a/loopx/control_plane/collaboration/turn_context.ts b/loopx/control_plane/collaboration/turn_context.ts
index 5f69f64114..82e1c0d7fe 100644
--- a/loopx/control_plane/collaboration/turn_context.ts
+++ b/loopx/control_plane/collaboration/turn_context.ts
@@ -29,10 +29,10 @@ export function projectDelegationTurnItem(value: unknown): JsonObject | null {
let step: string;
if (kind === "receipt_unavailable" || row.evidence_unavailable === true) step = "recover_evidence";
else if (cancelled) step = "reconcile_reservation";
+ else if (direction === "outgoing" && unconsumed.length) step = "consume_peer_result";
else if (kind !== "settled") step = direction === "outgoing" ? "await_peer_result"
: receipt.recorded_decision == null ? "assess_request"
: receipt.recorded_decision === "adopt" ? "review_request_work" : "return_answer";
- else if (direction === "outgoing" && unconsumed.length) step = "consume_peer_result";
else if (incomplete.length) step = "await_return_delivery";
else if (request.source_kind === "peer" && unconsumed.length) step = "await_peer_consumption";
else return null;
@@ -49,6 +49,7 @@ export function projectDelegationTurnItem(value: unknown): JsonObject | null {
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,
+ ...(receipt.remaining_work ? {remaining_work: receipt.remaining_work} : {}),
...(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,
From 97225ffee0d3921f52f0a3eddb647122b3709b6e Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 3 Nov 2026 01:36:55 +0800
Subject: [PATCH 2/7] docs(collaboration): explain partial results and
compatible recovery
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
loopx/capabilities/manager_context/README.md | 44 +++++++++++++++++++-
1 file changed, 42 insertions(+), 2 deletions(-)
diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md
index 8dbb6c0f50..8b1b6b3945 100644
--- a/loopx/capabilities/manager_context/README.md
+++ b/loopx/capabilities/manager_context/README.md
@@ -622,8 +622,9 @@ before the last position. Peer results retain their separate consumption flow
and are not paginated by this cursor. `status --offset/--limit` remains separate.
After `acknowledge`, the request remains visible until the answer obligation is
-resolved. `pending.items` retains its existing receipt-only semantics: it stops
-including a request once its conclusion exists. The separate unfinished agenda
+resolved. `pending.items` describes the current reply obligation: an ordinary
+conclusion resolves it, while an explicitly reported remainder keeps it open.
+Neither case certifies fulfillment of the user's requirements. The unfinished agenda
keeps undelivered conclusions visible, and keeps peer work until the requester
explicitly consumes the current result. The worker uses `link` for canonical Todo/evidence lineage
and `report` to publish the answer intended for the original audience:
@@ -716,6 +717,45 @@ conclude with the adopted/rejected planning decision; deferred or blocked work
must explain the concrete condition and next action. Completion of this exchange
is separate from completion of the receiving Goal.
+A partial conclusion can return a useful result while explicitly retaining work
+on this request. Use the existing result publication, without creating a Todo:
+
+```sh
+loopx manager-inbox report --goal-id research --agent-id worker \
+ --request-id --reply-text 'The implementation is ready; installed validation remains.' \
+ --remaining-work 'Validate the repair in the isolated installed package.' \
+ --resume-when 'The isolated package is available to the receiving Agent.'
+```
+
+MCP uses `return_result(..., remaining_work={"summary": "...", "resume_when": "..."})`.
+Both fields are required, each at most 600 Unicode characters; the rendered reply
+still fits the existing 20,000-character budget. They belong to the receiver's
+immutable result for its exact request revision. They do not transfer ownership,
+schedule execution or create a second completion state. The receiver retains the
+remainder under its existing authority; linked Todo status cannot resolve it.
+`resume_when` here describes a condition, unlike a typed Todo scheduler rule.
+
+The pending Inbox, Turn agenda and App delegation reader retain this request
+after the partial reply is delivered or consumed. The original conversation and
+provider reply also show the remainder and continuation condition. Queued replies
+remain distinct from delivered ones. A peer can consume the partial result once
+and still await the final result. Other requests sharing the same Todo retain
+their independent scope. Source correction and cancellation keep their existing
+request-control rules; an old revision's remainder cannot settle a newer request.
+
+When the remaining scope is actually resolved, append a conclusion using a stable
+`--update-id` and omit `--remaining-work`/`--resume-when`. This resolves the reply
+obligation after normal return processing, not independent acceptance. Retry with
+the same text and remainder: changing either under a committed result key is a
+conflict. Legacy unmarked results keep their existing behavior and unknown
+qualification; this path neither reclassifies old answers nor batch-closes work.
+Before rolling back to a reader without remainder support, stop new partial
+publications and reconcile the retained result chains with the compatible
+reader. Do not rewrite their receipts or treat an older reader's empty queue as
+fulfilled work. No installation, host wake or permission change is implied.
+普通短答仍可直接返回，无需创建任务；阶段结果保留原负责人和具体续接条件，
+不能用 Todo done、回传或送达替代原要求的验收。
+
If a returned blocker or draft is followed by a materially changed fact, append
a conclusion with a stable `--update-id`; do not replace the first conclusion or
create another request merely to return its result:
From 6b78d9e951aa3c2bb61b0a3c9880851a3ccbbd32 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 3 Nov 2026 01:36:55 +0800
Subject: [PATCH 3/7] test(collaboration): cover partial result continuity and
independent requests
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
.../smoke/delegation-search-browser-smoke.mjs | 15 ++
.../steward_requests_fixture.py | 17 ++-
.../collaboration_inbox_receipts.test.ts | 68 ++++++++-
.../collaboration_result_publication.test.ts | 22 +++
tests/test_collaboration_mcp.py | 13 +-
tests/test_delegation_remaining_work.py | 136 ++++++++++++++++++
tests/test_manager_context_tracking.py | 2 +-
7 files changed, 266 insertions(+), 7 deletions(-)
create mode 100644 tests/test_delegation_remaining_work.py
diff --git a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs
index 6eae2bb0b1..1aae7bfabe 100644
--- a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs
+++ b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs
@@ -239,6 +239,21 @@ try {
assert.equal(await english.getByRole('button', {name: 'Load next page', exact: true}).count(), 0);
assert.equal(await page.evaluate(() => document.activeElement?.textContent), '13 / 13 shown');
await english.getByRole('button', {name: 'Awaiting follow-up', exact: true}).click(); await rows(11);
+ // A useful reply can retain actual work even after delivery. Its later
+ // append-only result resolves this request, without settling its neighbors.
+ await fixture('report-partial');
+ await english.getByRole('button', {name: 'Refresh delegations', exact: true}).click();
+ await english.getByRole('searchbox', {name: 'Search delegations'}).fill('核对账单导出的字段映射');
+ await english.getByRole('searchbox', {name: 'Search delegations'}).press('Enter');
+ await rows(1);
+ await english.getByText('Partial result returned', {exact: true}).waitFor();
+ await english.locator('.conversation-request-row > summary').click();
+ await english.getByRole('heading', {name: 'Remaining work', exact: true}).waitFor();
+ assert.ok(await english.getByText('在隔离安装包中验证账单字段映射与失败恢复。', {exact: true}).isVisible());
+ await fixture('complete-partial');
+ await english.getByRole('button', {name: 'Refresh delegations', exact: true}).click(); await rows(0);
+ await english.getByRole('button', {name: 'All records', exact: true}).click(); await rows(1);
+ await english.getByText('Returned', {exact: true}).waitFor();
assert.equal(writes.some(path => /request-(?:inventory|control)|team-work/.test(path)), false);
assert.deepEqual(errors, []);
console.log('PASS: packaged search, pre-pagination match, count, complete original across real pages, long body, corrupt journal and later-page failure/recovery, close/reopen, clear/paging, changed-snapshot More, returned history, stale response, Escape/focus, workspace isolation, narrow layout, English, source refusal/recovery and read-only receipts; original-reader POST delta zero, Chat/receipt state unchanged and no new Turns.');
diff --git a/examples/personal-workspace-browser/steward_requests_fixture.py b/examples/personal-workspace-browser/steward_requests_fixture.py
index be861296ea..bb5eb215b5 100644
--- a/examples/personal-workspace-browser/steward_requests_fixture.py
+++ b/examples/personal-workspace-browser/steward_requests_fixture.py
@@ -5,7 +5,8 @@
decisions and return delivery below use the production owners in a temp runtime.
Commands on stdin: inspect, fault, fault-window, deliver-window, report-window, recover,
deliver, source-unavailable, source-recover, local-source-unavailable,
-local-source-recover, history-corrupt/history-recover (--long-history only), stop. Window faults/returns apply
+local-source-recover, history-corrupt/history-recover (--long-history only),
+report-partial, complete-partial, stop. Window faults/returns apply
after the first page to challenge multi-window refresh readback.
"""
from __future__ import annotations
@@ -222,6 +223,20 @@ def source_hashes():
drain(runtime, registry, store, lambda *_: None)
elif command == "deliver":
drain(runtime, registry, store, lambda *_: None)
+ elif command == "report-partial":
+ prior = _root(runtime) / "replies" / first_receipt["request_id"] / "conclusion.json"
+ report(runtime, "billing-exports", "builder", first_receipt["request_id"], "conclusion",
+ "字段映射的修改已完成，安装后的核验仍需继续。", registry=registry,
+ caller_goal_ref=first_receipt["goal_ref"],
+ update_id="remaining-installed-check" if prior.exists() else None,
+ remaining_work={"summary": "在隔离安装包中验证账单字段映射与失败恢复。",
+ "resume_when": "隔离安装包就绪后，由原负责人继续核验并返回实际结果。"})
+ drain(runtime, registry, store, lambda *_: None)
+ elif command == "complete-partial":
+ report(runtime, "billing-exports", "builder", first_receipt["request_id"], "conclusion",
+ "隔离包的字段映射与失败恢复已核验，原检查记录保留。", registry=registry,
+ caller_goal_ref=first_receipt["goal_ref"], update_id="qualified-installed-check")
+ drain(runtime, registry, store, lambda *_: None)
elif command == "conversation-arrival":
store.append_message(session["session_id"], role="assistant", text="新的委派阅读测试消息。", origin="manager_followup")
elif command == "source-unavailable":
diff --git a/tests/control_plane_ts/collaboration_inbox_receipts.test.ts b/tests/control_plane_ts/collaboration_inbox_receipts.test.ts
index 3ab82ccec4..3f1df19e19 100644
--- a/tests/control_plane_ts/collaboration_inbox_receipts.test.ts
+++ b/tests/control_plane_ts/collaboration_inbox_receipts.test.ts
@@ -1,5 +1,9 @@
import assert from "node:assert/strict";
import test from "node:test";
+import {createHash} from "node:crypto";
+import type {JsonObject} from "../../loopx/control_plane/effect_program.ts";
+import {planRequestControl, requestControlView} from "../../loopx/control_plane/collaboration/request_control.ts";
+import {planCollaborationResult} from "../../loopx/control_plane/collaboration/result_publication.ts";
import { inspectCollaborationInboxReceipts, projectReceiverFollowthrough } from "../../loopx/control_plane/collaboration/inbox_receipts.ts";
const identity = { request_id: "a".repeat(64), goal_id: "delivery", agent_id: "receiver", source_id: "peer:request" };
@@ -77,9 +81,11 @@ test("receiver advice keeps assessment, linked work and return distinct", () =>
}
const done = project({ recorded_decision: "adopt", kind: "awaiting_conclusion",
linked_todos: [{ todo_id: "todo_delivery", status: "done" }] })[0];
- assert.equal(done.step, "return_answer");
+ assert.equal(done.step, "review_request_work");
assert.equal(done.answer_owed, true);
assert.equal(done.request_completion, "not_established_by_receipts_or_todo_status");
+ assert.equal(project({kind: "settled", recorded_decision: "adopt",
+ linked_todos: [{todo_id: "todo_delivery", status: "open"}]})[0].step, "inspect_recorded_result");
for (const decision of ["defer", "reject", "no_change"]) {
assert.equal(project({ recorded_decision: decision, kind: "awaiting_conclusion" })[0].step, "return_answer");
}
@@ -92,3 +98,63 @@ test("receiver advice keeps assessment, linked work and return distinct", () =>
assert.throws(() => project({ recorded_decision: "in_progress" }), /unsupported/);
assert.throws(() => projectReceiverFollowthrough({ observations: Array(21).fill({}) }), /at most 20/);
});
+
+test("partial scope follows the exact current request revision and append-only result", () => {
+ const at = "2026-01-01T00:00:00Z";
+ const goal_ref = {goal_id: "delivery", goal_instance_id: "ginst_" + "b".repeat(32)};
+ const brief = {schema_version: "collaboration_brief_v0", purpose: "Repair the export",
+ context: "Keep the installed validation", constraints: [], inputs: [],
+ acceptance: ["The installed repair works"], return_requirement: "Return the checked repair"};
+ let entry: JsonObject = {...identity, goal_ref, schema_version: "loopx_manager_context_entry_v2",
+ source_kind: "peer", source_agent_id: "requester", brief, message: brief.purpose, delivered_at: at};
+ const legacy = {...decision, goal_ref, reason: "Accept original scope", decided_at: at};
+ const change = (operation: string, patch: JsonObject, actor = "source") => {
+ const current = requestControlView(entry, legacy);
+ const result = planRequestControl({entry, input: {operation, ...(operation === "assess" ? {} : {
+ expected_revision: current.revision, expected_digest: current.revision_digest}), ...patch}, source: {actor,
+ actor_agent_id: actor === "source" ? "requester" : "receiver", observed_at: at,
+ goal_active: true, legacy_assessment: legacy}});
+ entry = result.entry as JsonObject;
+ };
+ change("revise", {operation_id: "correct-2", brief: {...brief, context: "Use the corrected export"}, reason: "Corrected input"});
+ const view = requestControlView(entry, legacy);
+ change("assess", {operation_id: "adopt-2", revision: view.revision, revision_digest: view.revision_digest,
+ decision: "adopt", reason: "Own this exact correction"}, "receiver");
+ const remaining_work = {summary: "Validate the installed export", resume_when: "The package is available"};
+ const results: JsonObject[] = [];
+ const publish = (update_id?: string, remainder: unknown = remaining_work) => {
+ const plan = planCollaborationResult({request: entry, request_revision: 2, legacy_assessment: legacy,
+ phase: "conclusion", text: update_id ? "The installed export passed" : "The code is ready",
+ remaining_work: remainder, update_id, results});
+ const value = plan.value as JsonObject;
+ results.push({key: plan.result_key, value, text_chars: [...String(value.text)].length,
+ text_sha256: createHash("sha256").update(String(value.text)).digest("hex")});
+ };
+ const inspect = () => (inspectCollaborationInboxReceipts({observations: [{request: entry,
+ route_present: true, decision: read(legacy), read: absent, results}]}).items as JsonObject[])[0];
+ publish();
+ assert.equal(inspect().kind, "awaiting_conclusion");
+ assert.deepEqual(inspect().remaining_work, remaining_work);
+ publish("checked", null);
+ assert.equal(inspect().kind, "settled");
+ assert.equal(inspect().remaining_work, undefined);
+ assert.deepEqual((results[0].value as JsonObject).remaining_work, remaining_work);
+ change("revise", {operation_id: "correct-3", brief: {...brief, purpose: "Check the new export"}, reason: "New requirement"});
+ assert.equal(inspect().kind, "pending");
+ assert.equal(inspect().remaining_work, undefined);
+ change("cancel", {operation_id: "cancel-3", reason: "Stop only this request"});
+ assert.equal(inspect().kind, "settled");
+});
+
+test("an orphaned or unreadable update cannot hide an outstanding request", () => {
+ const original = {...conclusion, remaining_work: {summary: "Validate the repair", resume_when: "Package ready"}};
+ const update = {...identity, decision: "adopt", phase: "conclusion", text: "Later result",
+ result_key: "conclusion-00000001", previous_result_key: "conclusion", update_id: "later"};
+ const observed = {key: update.result_key, value: update, text_chars: update.text.length,
+ text_sha256: createHash("sha256").update(update.text).digest("hex")};
+ for (const results of [[observed], [{unavailable: true}]]) {
+ const row = (state({decision: read(decision), conclusion: absent, results}) as JsonObject[])[0];
+ assert.equal(row.kind, "receipt_unavailable");
+ }
+ assert.equal((state({decision: read(decision), conclusion: read(original)}) as JsonObject[])[0].kind, "awaiting_conclusion");
+});
diff --git a/tests/control_plane_ts/collaboration_result_publication.test.ts b/tests/control_plane_ts/collaboration_result_publication.test.ts
index 2bef755a84..0190231487 100644
--- a/tests/control_plane_ts/collaboration_result_publication.test.ts
+++ b/tests/control_plane_ts/collaboration_result_publication.test.ts
@@ -106,3 +106,25 @@ test("explicit result files participate in immutable retry identity", () => {
assert.throws(() => planCollaborationResult({...source, attachments: [{...attachment, ...patch}]}), /attachment/);
}
});
+
+test("a reported remainder is bounded immutable result scope, not a completion switch", () => {
+ const remaining_work = {summary: "Qualify the installed repair", resume_when: "The isolated package is available"};
+ const source = {...input, update_id: null, results: [], remaining_work};
+ const plan = planCollaborationResult(source);
+ const replay = {...source, results: [observation("conclusion", plan.value as JsonObject)]};
+ assert.equal(planCollaborationResult(replay).replay, true);
+ assert.deepEqual((plan.value as JsonObject).remaining_work, remaining_work);
+ for (const change of [null, {...remaining_work, summary: "Everything done"}, {...remaining_work, resume_when: "Now"}]) {
+ assert.throws(() => planCollaborationResult({...replay, remaining_work: change}), /conflicting/);
+ }
+ for (const invalid of [{summary: "Only a summary"}, {...remaining_work, owner: "another-agent"},
+ {...remaining_work, summary: " "}, {...remaining_work, resume_when: "😀".repeat(601)},
+ {...remaining_work, summary: "\ud800"}]) {
+ assert.throws(() => planCollaborationResult({...source, remaining_work: invalid}), /remaining work/);
+ }
+ assert.throws(() => planCollaborationResult({...source, phase: "decision"}), /conclusion/);
+ assert.throws(() => planCollaborationResult({...source, text: "x".repeat(20000)}), /result budget/);
+ const final = planCollaborationResult({...input, results: replay.results});
+ assert.equal(Object.hasOwn(final.value as JsonObject, "remaining_work"), false);
+ assert.deepEqual((replay.results[0].value as JsonObject).remaining_work, remaining_work);
+});
diff --git a/tests/test_collaboration_mcp.py b/tests/test_collaboration_mcp.py
index 724af737bb..088e9cd1ef 100644
--- a/tests/test_collaboration_mcp.py
+++ b/tests/test_collaboration_mcp.py
@@ -216,10 +216,15 @@ async def call(name, args):
cli("todo", "archive-completed", "--goal-id", "goal-a",
"--max-active-done", "0", "--execute")
resumed = (await call("read_context", {}))["items"][0]["receiver_followthrough"]
- assert resumed["step"] == "return_answer" and resumed["answer_owed"]
+ assert resumed["step"] == "review_request_work" and resumed["answer_owed"]
assert [(t["todo_id"], t["status"]) for t in resumed["linked_todos"]] == [(own, "done")]
assert path.read_bytes() == original
- await call("return_result", {"request_id": rid, "text": conclusion})
+ remainder = {"summary": "Check the installed repair", "resume_when": "The isolated package is ready"}
+ await call("return_result", {"request_id": rid, "text": "Implementation ready", "remaining_work": remainder})
+ pending = (await call("read_context", {}))["items"][0]
+ assert pending["remaining_work"] == remainder
+ assert pending["receiver_followthrough"]["step"] == "review_request_work"
+ await call("return_result", {"request_id": rid, "text": conclusion, "update_id": "checked-repair"})
assert (await call("read_context", {}))["items"] == []
# A short factual answer requires neither a fabricated task nor a link.
quick = request(root, registry, "goal-a", "agent-b", "agent-a", "short-answer", brief)["request_id"]
@@ -239,7 +244,7 @@ async def call(name, args):
asyncio.run(exercise())
received = read_inbox(root, registry, "goal-a", "agent-b")["peer_returns"]["items"]
- assert len(received) == 2
- assert next(item for item in received if item["request_id"] == rid)["text"] == conclusion
+ assert len(received) == 3
+ assert next(item for item in received if item.get("result_key") == "conclusion-00000001")["text"] == conclusion
assert [row["todo_id"] for row in cli("todo", "list", "--goal-id", "goal-a")["todos"]] == [foreign]
assert cli("todo", "list", "--goal-id", "goal-a", "--todo-id", own)["todo"]["archive_state"] == "archive"
diff --git a/tests/test_delegation_remaining_work.py b/tests/test_delegation_remaining_work.py
new file mode 100644
index 0000000000..6a685739fc
--- /dev/null
+++ b/tests/test_delegation_remaining_work.py
@@ -0,0 +1,136 @@
+"""A returned partial result retains its request without creating work authority."""
+
+import json
+import subprocess
+import sys
+
+import pytest
+
+from loopx.capabilities.manager_context import _root, acknowledge
+from loopx.capabilities.manager_context.roundtrip import drain, report
+from loopx.control_plane.collaboration.peers import consume_return, read_inbox, request
+from loopx.control_plane.collaboration.turn_context_observation import read_delegation_turn_context
+import test_manager_context_roundtrip as roundtrip_tests
+
+
+REMAINDER = {"summary": "Validate the installed repair", "resume_when": "The isolated package is available"}
+
+
+@pytest.fixture
+def flow(tmp_path):
+ return roundtrip_tests.flow.__wrapped__(tmp_path)
+
+
+def agenda(root, registry, agent="worker"):
+ return read_delegation_turn_context(root, registry, "research", agent)["unfinished_delegations"]
+
+
+@pytest.mark.parametrize("external", [False, True], ids=["app", "provider"])
+def test_partial_return_keeps_its_scope_after_delivery_and_restart(flow, external):
+ root, registry, store, create = flow
+ session, _, receipt = create(external=external)
+ rid = receipt["request_id"]
+ acknowledge(root, "research", "worker", rid, "adopt", "Repair and validate the original request")
+ report(root, "research", "worker", rid, "conclusion", "Implementation is ready.", remaining_work=REMAINDER)
+ original = (_root(root) / "replies" / rid / "conclusion.json").read_bytes()
+ sent = []
+
+ def sender(_route, _session, _turn, text):
+ sent.append(text)
+ return {"reply_verified": True, "external_write_performed": True}
+
+ drain(root, registry, store, sender)
+ # A fresh native read, independently of a model or a Todo, retains the work.
+ for _ in range(2):
+ item = agenda(root, registry)["items"][0]
+ assert item["request_id"] == rid and item["step"] == "review_request_work"
+ assert item["remaining_work"] == REMAINDER
+ assert item["pending_return_count"] == 0
+ returned = [m for m in store.messages(session["session_id"]) if m.get("origin") == "manager_followup"]
+ assert len(returned) == 1 and REMAINDER["resume_when"] in returned[0]["text"]
+ if external:
+ assert len(sent) == 1 and REMAINDER["summary"] in sent[0]
+ report(root, "research", "worker", rid, "conclusion", "The isolated repair passed the agreed validation.", update_id="qualified-result")
+ assert agenda(root, registry)["items"][0]["step"] == "await_return_delivery"
+ drain(root, registry, store, sender)
+ assert agenda(root, registry)["items"] == []
+ assert (_root(root) / "replies" / rid / "conclusion.json").read_bytes() == original
+ assert len([m for m in store.messages(session["session_id"]) if m.get("origin") == "manager_followup"]) == 2
+
+
+def test_partial_peer_result_is_consumed_once_while_work_stays_owed(tmp_path):
+ root, registry, _, _ = roundtrip_tests.flow.__wrapped__(tmp_path)
+ data = json.loads(registry.read_text())
+ data["goals"][0]["coordination"]["registered_agents"].append("reviewer")
+ registry.write_text(json.dumps(data))
+ brief = {"schema_version": "collaboration_brief_v0", "purpose": "Repair and independently validate",
+ "context": "Preserve the acceptance requirement", "constraints": [], "inputs": [],
+ "acceptance": ["The installed repair works"], "return_requirement": "Return the checked repair"}
+ rid = request(root, registry, "research", "worker", "reviewer", "repair", brief)["request_id"]
+ acknowledge(root, "research", "reviewer", rid, "adopt", "Own the repair and validation")
+ report(root, "research", "reviewer", rid, "conclusion", "The implementation is ready.", remaining_work=REMAINDER)
+ assert agenda(root, registry)["items"][0]["step"] == "consume_peer_result"
+ result = read_inbox(root, registry, "research", "worker")["peer_returns"]["items"][0]
+ assert result["remaining_work"] == REMAINDER
+ consume_return(root, "research", "worker", rid, registry=registry)
+ assert agenda(root, registry)["items"][0]["step"] == "await_peer_result"
+ assert agenda(root, registry, "reviewer")["items"][0]["step"] == "review_request_work"
+
+
+@pytest.mark.parametrize("provider", ["file", "sqlite"])
+def test_cli_partial_request_and_shared_todo_have_independent_lifetimes(tmp_path, monkeypatch, provider):
+ from tests.control_plane.canonical_authority_fixture import isolate_sqlite_runtime, promoted_create_fixture
+ from loopx.capabilities.manager_context import deliver
+
+ isolate_sqlite_runtime(tmp_path, monkeypatch)
+ registry, root, _ = promoted_create_fixture(tmp_path, provider=provider)
+
+ def cli(*args, ok=True):
+ process = subprocess.run([sys.executable, "-m", "loopx.entrypoint", "--format", "json",
+ "--registry", str(registry), "--runtime-root", str(root), *args], capture_output=True, text=True, timeout=45)
+ assert process.returncode == (0 if ok else 1), (process.stdout, process.stderr)
+ result = json.loads(process.stdout)
+ assert result["ok"] is ok
+ return result
+
+ work = cli("todo", "add", "--goal-id", "goal-a", "--role", "agent", "--text", "Ongoing umbrella work", "--claimed-by", "agent-a")
+ tid = work["todo_id"]
+ request_ids = []
+ for index in range(2):
+ rid = deliver(root, registry, session={"session_id": "original", "channel_id": "manager"},
+ turn={"client_turn_id": f"request-{index}", "origin": "web", "message": "Return this request's actual result"},
+ request={"goal_id": "goal-a", "agent_id": "agent-a"})["request_id"]
+ request_ids.append(rid)
+ scope = ("--goal-id", "goal-a", "--agent-id", "agent-a", "--request-id", rid)
+ cli("manager-inbox", "acknowledge", *scope, "--decision", "adopt", "--reason", "Keep each request independent")
+ cli("manager-inbox", "link", *scope, "--related-todo-id", tid)
+ cli("manager-inbox", "report", *scope, "--reply-text", "A partial result", "--remaining-work", REMAINDER["summary"], "--resume-when", REMAINDER["resume_when"])
+ cli("todo", "complete", "--goal-id", "goal-a", "--todo-id", tid, "--agent-id", "agent-a", "--evidence", "The umbrella's separate scope passed", "--no-follow-up")
+ pending = cli("manager-inbox", "read", "--goal-id", "goal-a", "--agent-id", "agent-a")["items"]
+ assert {item["request_id"] for item in pending} == set(request_ids)
+ assert all(item["remaining_work"] == REMAINDER and item["receiver_followthrough"]["step"] == "review_request_work" for item in pending)
+ scope = ("--goal-id", "goal-a", "--agent-id", "agent-a", "--request-id", request_ids[0])
+ cli("manager-inbox", "report", *scope, "--reply-text", "First request's checked result", "--update-id", "verified-first")
+ pending = cli("manager-inbox", "read", "--goal-id", "goal-a", "--agent-id", "agent-a")["items"]
+ assert [item["request_id"] for item in pending] == [request_ids[1]]
+ assert cli("todo", "list", "--goal-id", "goal-a", "--todo-id", tid)["todo"]["status"] == "done"
+ failure = cli("manager-inbox", "report", *scope, "--reply-text", "Invalid", "--remaining-work", "Unpaired", ok=False)
+ assert "require" in failure["error"]
+
+
+def test_ordinary_short_answer_needs_no_work_record_and_external_remainder_is_scanned(flow):
+ root, registry, store, create = flow
+ _, _, receipt = create()
+ rid = receipt["request_id"]
+ acknowledge(root, "research", "worker", rid, "no_change", "A direct answer is sufficient")
+ report(root, "research", "worker", rid, "conclusion", "The current public PR is merged.")
+ drain(root, registry, store, lambda *_: pytest.fail("local return"))
+ assert agenda(root, registry)["items"] == []
+ assert not (_root(root) / "links" / f"{rid}.json").exists()
+ _, _, external = create(external=True)
+ rid = external["request_id"]
+ acknowledge(root, "research", "worker", rid, "adopt", "An audience-safe return")
+ with pytest.raises(ValueError, match="private boundary"):
+ report(root, "research", "worker", rid, "conclusion", "Safe visible answer", remaining_work={
+ "summary": "/Users/private/customer/raw.txt", "resume_when": "Ready"})
+ assert not (_root(root) / "replies" / rid / "conclusion.json").exists()
diff --git a/tests/test_manager_context_tracking.py b/tests/test_manager_context_tracking.py
index ee95dcfb10..c1a426d04d 100644
--- a/tests/test_manager_context_tracking.py
+++ b/tests/test_manager_context_tracking.py
@@ -346,7 +346,7 @@ def add(owner, goal="goal-a"):
assert archived["todo"]["archive_state"] == "archive"
assert archived["todo"]["status"] == "done"
follow = cli("manager-inbox", "read", "--goal-id", "goal-a", "--agent-id", "agent-a")["items"][0]["receiver_followthrough"]
- assert follow["step"] == "return_answer"
+ assert follow["step"] == "review_request_work"
assert [(t["todo_id"], t["title"], t["status"]) for t in follow["linked_todos"]] == [
(tid, "[P1] Verified current work", "done")
]
From 2334ac936937c8920cf109abae24d68746b90525 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 3 Nov 2026 02:30:02 +0800
Subject: [PATCH 4/7] perf(collaboration): reuse result observations within
agenda reads
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
loopx/control_plane/collaboration/inbox_read.ts | 16 ++++++++++------
1 file changed, 10 insertions(+), 6 deletions(-)
diff --git a/loopx/control_plane/collaboration/inbox_read.ts b/loopx/control_plane/collaboration/inbox_read.ts
index 411efe7b24..76e43259b7 100644
--- a/loopx/control_plane/collaboration/inbox_read.ts
+++ b/loopx/control_plane/collaboration/inbox_read.ts
@@ -225,11 +225,12 @@ async function pendingPage(source: InboxSource, params: JsonObject): Promise
- resultObservation(candidate, readRecord(join(folder, candidate + ".json")))) : [resultObservation(key, reply)];
+ const results = observations ?? (reply.update_id != null || reply.phase === "decision" ? resultKeys(folder).map(candidate =>
+ resultObservation(candidate, readRecord(join(folder, candidate + ".json")))) : [resultObservation(key, reply)]);
const legacy = Object.hasOwn(row, "request_control") || Object.hasOwn(reply, "request_revision")
? observe(join(root, "decisions", String(row.request_id) + ".json")) : {};
const inspected = planCollaborationResult({request: row, phase: reply.phase, text: reply.text,
@@ -330,10 +331,13 @@ function replyStatuses(root: string, row: JsonObject): JsonObject[] {
function agendaResults(root: string, row: JsonObject): JsonObject[] {
const folder = join(root, "replies", requestId(row.request_id));
const statuses = new Map(replyStatuses(root, row).map(status => [status.result_key ?? status.phase, status.status]));
- return resultKeys(folder).flatMap(key => {
- const reply = readRecord(join(folder, key + ".json"));
+ const replies = resultKeys(folder).map(key => ({key, reply: readRecord(join(folder, key + ".json"))}));
+ // Reuse immutable request-local observations; each result still goes through
+ // the publication validator and its current delivery/consumption readback.
+ const observations = replies.map(({key, reply}) => resultObservation(key, reply));
+ return replies.flatMap(({key, reply}) => {
if (reply.phase !== "conclusion") return [];
- if (!inspectResult(root, row, key, reply)) fail("delegation result identity unavailable");
+ if (!inspectResult(root, row, key, reply, observations)) fail("delegation result identity unavailable");
return [{result_key: key, request_revision: reply.request_revision ?? 1, status: statuses.get(key),
consumed: row.source_kind === "peer" && returnConsumed(root, key, row, reply, row.source_agent_id)}];
});
From 25a9fb1a5cedaad899b1752680c53ebb0290ba78 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 3 Nov 2026 02:30:03 +0800
Subject: [PATCH 5/7] test(collaboration): retain partial scope across real
guard transports
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
.../inbox_read_transaction.test.ts | 45 +++++++++++++++++++
tests/test_delegation_turn_context.py | 8 +++-
2 files changed, 52 insertions(+), 1 deletion(-)
diff --git a/tests/control_plane_ts/inbox_read_transaction.test.ts b/tests/control_plane_ts/inbox_read_transaction.test.ts
index 6aeeee0ce3..c9ea35acf2 100644
--- a/tests/control_plane_ts/inbox_read_transaction.test.ts
+++ b/tests/control_plane_ts/inbox_read_transaction.test.ts
@@ -220,3 +220,48 @@ test("a large private agenda lets another runtime request finish before the scan
assert.equal(finished, false);
assert.equal(((await scanning).unfinished_delegations as JsonObject).incoming_count, 256);
});
+
+for (const profile of ["legacy", "exact"] as const) {
+ test(`${profile}: a complete result chain preserves its remainder, delivery and corruption boundaries`, async t => {
+ const f = fixture(t, profile), row = f.publish(1);
+ const identity = Object.fromEntries(["request_id", "goal_id", "agent_id", "source_id", "goal_ref"]
+ .filter(key => Object.hasOwn(row, key)).map(key => [key, row[key as keyof typeof row]]));
+ const replies = join(f.base, "replies", row.request_id);
+ for (const folder of [replies, join(f.base, "decisions"), join(f.base, "roundtrips")]) mkdirSync(folder, {recursive: true});
+ writeFileSync(join(f.base, "decisions", row.request_id + ".json"), JSON.stringify({...identity, decision: "adopt"}));
+ writeFileSync(join(f.base, "roundtrips", row.request_id + ".json"), JSON.stringify(identity));
+ const remaining_work = {summary: "Validate the packaged repair", resume_when: "The isolated package is available"};
+ const key = (n: number) => n ? "conclusion-" + String(n).padStart(8, "0") : "conclusion";
+ for (let n = 0; n < 4; n++) {
+ writeFileSync(join(replies, key(n) + ".json"), JSON.stringify({...identity, phase: "conclusion", decision: "adopt",
+ text: "Recorded evidence " + n + "😀".repeat(2000), remaining_work,
+ ...(n ? {result_key: key(n), previous_result_key: key(n - 1), update_id: "evidence-" + n} : {})}));
+ writeFileSync(join(replies, key(n) + ".delivery.json"), JSON.stringify({status: "delivered"}));
+ }
+ const original = sourceBytes(replies);
+ const pending = (await f.call("pending")).result as JsonObject;
+ assert.deepEqual((pending.items as JsonObject[])[0].remaining_work, remaining_work);
+ const agenda = (await f.call("turn_context")).unfinished_delegations as JsonObject;
+ assert.equal(agenda.incoming_count, 1);
+ assert.equal((agenda.items as JsonObject[])[0].step, "review_request_work");
+ assert.deepEqual(sourceBytes(replies), original);
+
+ const damaged = join(replies, key(2) + ".json"), bytes = readFileSync(damaged, "utf8");
+ const corrupt = JSON.parse(bytes);
+ corrupt.previous_result_key = "conclusion";
+ writeFileSync(damaged, JSON.stringify(corrupt));
+ const failed = (await f.call("pending")).result as JsonObject;
+ assert.equal((failed.items as JsonObject[])[0].inbox_state, "receipt_unavailable");
+ const unavailable = (await f.call("turn_context")).unfinished_delegations as JsonObject;
+ assert.equal((unavailable.items as JsonObject[])[0].step, "recover_evidence");
+ writeFileSync(damaged, bytes);
+ assert.deepEqual(sourceBytes(replies), original);
+
+ writeFileSync(join(replies, key(4) + ".json"), JSON.stringify({...identity, phase: "conclusion", decision: "adopt",
+ text: "The requested package passed validation.", result_key: key(4), previous_result_key: key(3), update_id: "validated"}));
+ const queued = (await f.call("turn_context")).unfinished_delegations as JsonObject;
+ assert.equal((queued.items as JsonObject[])[0].step, "await_return_delivery");
+ writeFileSync(join(replies, key(4) + ".delivery.json"), JSON.stringify({status: "delivered"}));
+ assert.deepEqual(((await f.call("turn_context")).unfinished_delegations as JsonObject).items, []);
+ });
+}
diff --git a/tests/test_delegation_turn_context.py b/tests/test_delegation_turn_context.py
index cd88e6be24..b44e98cba6 100644
--- a/tests/test_delegation_turn_context.py
+++ b/tests/test_delegation_turn_context.py
@@ -96,7 +96,8 @@ def test_separate_agent_has_no_access_to_another_receivers_agenda(flow):
@pytest.mark.parametrize("provider", ["file", "sqlite"])
@pytest.mark.parametrize("delivery", ["host_envelope", pytest.param("native_files",
marks=pytest.mark.skipif(os.name != "posix", reason="native context delivery requires POSIX"))])
-def test_actual_cli_guards_and_host_envelopes_preserve_the_agenda(tmp_path, monkeypatch, provider, delivery):
+@pytest.mark.parametrize("partial", [False, True], ids=["unanswered", "partial"])
+def test_actual_cli_guards_and_host_envelopes_preserve_the_agenda(tmp_path, monkeypatch, provider, delivery, partial):
from control_plane.canonical_authority_fixture import initialize_canonical_authority, isolate_sqlite_runtime
from loopx.control_plane.coordination.runtime_shadow import build_todo_runtime_shadow_projection
from loopx.control_plane.testing.canary_harness import run_json_cli_result, write_fixture_registry
@@ -125,6 +126,10 @@ def test_actual_cli_guards_and_host_envelopes_preserve_the_agenda(tmp_path, monk
acknowledge(root, "research", "worker", receipt["request_id"], "adopt", "Continue the linked work", registry=registry)
from loopx.control_plane.collaboration.links import link
link(root, registry, "research", "worker", receipt["request_id"], ["todo_artifact"], [])
+ remaining_work = {"summary": "Validate the packaged repair", "resume_when": "The isolated package is available"}
+ if partial:
+ report(root, "research", "worker", receipt["request_id"], "conclusion", "The implementation is ready.",
+ remaining_work=remaining_work, registry=registry)
before = {p.relative_to(root): p.read_bytes() for p in _root(root).rglob("*.json")}
if delivery == "native_files":
from loopx.control_plane.turn_driver.codex_context import codex_context_directory, codex_context_hook
@@ -168,6 +173,7 @@ def test_actual_cli_guards_and_host_envelopes_preserve_the_agenda(tmp_path, monk
row = content["unfinished_delegations"]["items"][0]
assert row["request_id"] == receipt["request_id"]
assert row["todo_ids"] == ["todo_artifact"] and row["step"] == "review_request_work"
+ assert row.get("remaining_work") == (remaining_work if partial else None)
assert any("manager-inbox read" in item["command"] for item in required_reads)
assert "Continue the linked work" not in json.dumps(content)
assert {p.relative_to(root): p.read_bytes() for p in _root(root).rglob("*.json")} == before
From a2c4388e328b0db19a9a149ff1d80c56eac8b287 Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 3 Nov 2026 02:39:39 +0800
Subject: [PATCH 6/7] docs(collaboration): clarify result and validation
boundaries
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
.../dashboard/smoke/delegation-search-browser-smoke.mjs | 2 +-
loopx/capabilities/manager_context/README.md | 5 +++--
2 files changed, 4 insertions(+), 3 deletions(-)
diff --git a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs
index 1aae7bfabe..8984adf27e 100644
--- a/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs
+++ b/apps/presentation/dashboard/smoke/delegation-search-browser-smoke.mjs
@@ -256,7 +256,7 @@ try {
await english.getByText('Returned', {exact: true}).waitFor();
assert.equal(writes.some(path => /request-(?:inventory|control)|team-work/.test(path)), false);
assert.deepEqual(errors, []);
- console.log('PASS: packaged search, pre-pagination match, count, complete original across real pages, long body, corrupt journal and later-page failure/recovery, close/reopen, clear/paging, changed-snapshot More, returned history, stale response, Escape/focus, workspace isolation, narrow layout, English, source refusal/recovery and read-only receipts; original-reader POST delta zero, Chat/receipt state unchanged and no new Turns.');
+ console.log('PASS: packaged search, pre-pagination match, count, complete original across real pages, long body, corrupt journal and later-page failure/recovery, close/reopen, clear/paging, changed-snapshot More, partial/final returned history, stale response, Escape/focus, workspace isolation, narrow layout, English and source refusal/recovery; explicit fixture reports update return state, original-reader POST delta zero and no new Turns.');
}
} catch (error) {
if (page) {await page.screenshot({path: resolve(output, 'failure.png')}).catch(() => {}); await writeFile(resolve(output, 'failure.txt'), await page.locator('body').innerText());}
diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md
index 8b1b6b3945..1a2bfef80c 100644
--- a/loopx/capabilities/manager_context/README.md
+++ b/loopx/capabilities/manager_context/README.md
@@ -747,8 +747,9 @@ When the remaining scope is actually resolved, append a conclusion using a stabl
`--update-id` and omit `--remaining-work`/`--resume-when`. This resolves the reply
obligation after normal return processing, not independent acceptance. Retry with
the same text and remainder: changing either under a committed result key is a
-conflict. Legacy unmarked results keep their existing behavior and unknown
-qualification; this path neither reclassifies old answers nor batch-closes work.
+conflict. Valid unmarked legacy results retain their ordinary reply state and
+unknown fulfillment qualification; unreadable or conflicting result chains stay
+unavailable. This path does not batch-close old work.
Before rolling back to a reader without remainder support, stop new partial
publications and reconcile the retained result chains with the compatible
reader. Do not rewrite their receipts or treat an older reader's empty queue as
From 75b0fbe9a58260d7f5f98b1beb37efde88cd0e1f Mon Sep 17 00:00:00 2001
From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
Date: Tue, 3 Nov 2026 03:26:13 +0800
Subject: [PATCH 7/7] test(collaboration): qualify partial Lark return recovery
Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com>
---
.../test_lark_private_manager_returns.py | 86 ++++++++++++++++++-
1 file changed, 84 insertions(+), 2 deletions(-)
diff --git a/tests/extensions/test_lark_private_manager_returns.py b/tests/extensions/test_lark_private_manager_returns.py
index 1b7ff7ca09..87f6130dd9 100644
--- a/tests/extensions/test_lark_private_manager_returns.py
+++ b/tests/extensions/test_lark_private_manager_returns.py
@@ -1,5 +1,6 @@
"""Worker returns use the original Core-bound private App, including recovery."""
import json
+from datetime import datetime, timedelta, timezone
from types import SimpleNamespace
import pytest
@@ -11,6 +12,7 @@
from loopx.chat_store import ChatSessionStore
from loopx.capabilities.manager_context import acknowledge
from loopx.capabilities.manager_context.roundtrip import drain, report, reply_status
+from loopx.control_plane.collaboration.turn_context_observation import read_delegation_turn_context
@pytest.fixture(params=["shared", "separate"])
@@ -82,7 +84,7 @@ def runner(args, cwd=None, timeout=None):
_write(_root(return_root) / "policy.json", {"schema_version": POLICY_SCHEMA, "sources": {}})
else:
replies.append(list(args))
- ref = "om_out_worker"
+ ref = f"om_out_worker_{len(replies)}"
provider.messages[ref] = {"message_id": ref, "msg_type": "post", "body": {"content": content}}
data = {"data": {"message_id": ref}}
return {"returncode": 0, "stdout": json.dumps(data)}
@@ -192,7 +194,6 @@ def test_production_pump_returns_to_original_private_source_after_restart(privat
assert reply_status(sender.root, route)[0]["status"] == "verification_required"
provider.verify_replies = True
# Reload the actual Chat store; the persisted attempt drives read-only recovery.
- from datetime import datetime, timedelta, timezone
recovered = ChatSessionStore(store.root.parent)
drain(sender.root, controller.registry_path, recovered, sender, now=datetime.now(timezone.utc) + timedelta(minutes=10))
assert reply_status(sender.root, route)[0]["status"] == "delivered"
@@ -208,3 +209,84 @@ def test_production_pump_returns_to_original_private_source_after_restart(privat
assert "Receiver independently" not in returned[0]["text"]
assert sorted((store.root / "sessions" / session["session_id"] / "turns").glob("*.json")) == before_turns
assert capture.read_bytes() == before_host_calls
+
+
+@pytest.mark.parametrize("revoke_before_recovery", [False, True])
+def test_partial_return_survives_saved_attempt_recovery_until_final_delivery(private_return, revoke_before_recovery):
+ sender, session, turn, route, _, provider, transport, replies = private_return
+ controller = transport.core.controller
+ store = controller.store
+ capture = store.root.parent.parent / "requests.jsonl"
+ before_host_calls = capture.read_bytes()
+ original_turn = store._turn_path(session["session_id"], turn["turn_id"]).read_bytes()
+ turns = sorted((store.root / "sessions" / session["session_id"] / "turns").glob("*.json"))
+ remainder = {"summary": "Validate the installed repair", "resume_when": "The isolated package is available"}
+ rid = route["request_id"]
+ acknowledge(sender.root, route["goal_id"], route["agent_id"], rid, "adopt", "Repair and validate")
+ report(sender.root, route["goal_id"], route["agent_id"], rid, "conclusion",
+ "The source repair is ready.", remaining_work=remainder)
+ partial = _root(sender.root) / "replies" / rid / "conclusion.json"
+ original_result = partial.read_bytes()
+ provider.verify_replies = False
+ drain(sender.root, controller.registry_path, store, sender)
+ assert len(replies) == 1
+ assert reply_status(sender.root, route)[0]["status"] == "verification_required"
+ partial_content = replies[0][replies[0].index("--content") + 1]
+ assert all(value in partial_content for value in remainder.values())
+
+ # Reload the saved attempt. Revocation still wins over that historical proof.
+ recovered = ChatSessionStore(store.root.parent)
+ resumed_sender = LarkManagerReturnTransport(sender.server, sender.root)
+ provider.verify_replies = True
+ clock = datetime.now(timezone.utc) + timedelta(minutes=10)
+ if revoke_before_recovery:
+ policy_path = _root(sender.root) / "policy.json"
+ policy = policy_path.read_bytes()
+ _write(policy_path, {"schema_version": POLICY_SCHEMA, "sources": {}})
+ drain(sender.root, controller.registry_path, recovered, resumed_sender, now=clock)
+ status = reply_status(sender.root, route)[0]
+ assert (status["status"], status["error"]) == ("explicit_unverified", "return_authorization_unavailable")
+ assert len(replies) == 1
+ policy_path.write_bytes(policy)
+ # Restoring access does not automatically resume a stopped publication.
+ clock += timedelta(minutes=10)
+ drain(sender.root, controller.registry_path, recovered, resumed_sender, now=clock)
+ partial_status = "explicit_unverified" if revoke_before_recovery else "delivered"
+ assert reply_status(sender.root, route)[0]["status"] == partial_status
+ assert len(replies) == 1
+ agenda = read_delegation_turn_context(sender.root, controller.registry_path,
+ route["goal_id"], route["agent_id"])["unfinished_delegations"]
+ assert [(item["request_id"], item["step"], item.get("remaining_work")) for item in agenda["items"]] == [
+ (rid, "review_request_work", remainder)]
+
+ report(sender.root, route["goal_id"], route["agent_id"], rid, "conclusion",
+ "The isolated repair passed the agreed validation.", update_id="qualified-result")
+ drain(sender.root, controller.registry_path, recovered, resumed_sender, now=clock + timedelta(minutes=20))
+ drain(sender.root, controller.registry_path, ChatSessionStore(store.root.parent), resumed_sender,
+ now=clock + timedelta(minutes=30))
+ statuses = [item["status"] for item in reply_status(sender.root, route)]
+ agenda = read_delegation_turn_context(sender.root, controller.registry_path,
+ route["goal_id"], route["agent_id"])["unfinished_delegations"]["items"]
+ returned = [item for item in recovered.messages(session["session_id"])
+ if item.get("origin") == "manager_followup"]
+ if revoke_before_recovery:
+ # A newer result cannot overtake an uncertain predecessor, even after
+ # access returns. Preserve the exact failure instead of resending it.
+ assert len(replies) == 1 and statuses == ["explicit_unverified", "queued"]
+ assert [(item["request_id"], item["step"], item["pending_return_count"]) for item in agenda] == [
+ (rid, "await_return_delivery", 2)]
+ assert len(returned) == 1
+ else:
+ assert len(replies) == 2 and statuses == ["delivered", "delivered"]
+ assert agenda == []
+ assert len(returned) == 2
+ final_content = replies[1][replies[1].index("--content") + 1]
+ assert "The isolated repair passed" in final_content
+ assert "The isolated repair passed" in returned[1]["text"]
+ assert remainder["summary"] not in returned[1]["text"]
+ assert all(item["turn_id"] == turn["turn_id"] for item in returned)
+ assert all(value in returned[0]["text"] for value in remainder.values())
+ assert partial.read_bytes() == original_result
+ assert store._turn_path(session["session_id"], turn["turn_id"]).read_bytes() == original_turn
+ assert sorted((store.root / "sessions" / session["session_id"] / "turns").glob("*.json")) == turns
+ assert capture.read_bytes() == before_host_calls