+ {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/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/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md
index 8dbb6c0f50..1a2bfef80c 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,46 @@ 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. 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
+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:
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..76e43259b7 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,
@@ -218,15 +225,17 @@ 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,
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 +290,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]])),
@@ -321,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)}];
});
@@ -364,6 +377,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,
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/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/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
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_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
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")
]