From 8d76419078daaeae1d10070f0bb12cd2dc7c0572 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Fri, 13 Nov 2026 12:05:45 +0800 Subject: [PATCH] Deliver native child results on the next Turn Signed-off-by: huangruiteng --- .../host-native-child-receipts.md | 42 ++- .../capabilities/manager_context/__init__.py | 3 + loopx/capabilities/multi_subagent/cli.py | 48 +++- .../native_child_result_inbox.py | 49 ++++ .../capabilities/native_child_result_inbox.ts | 257 ++++++++++++++++++ .../control_plane/collaboration/inbox_read.ts | 34 ++- loopx/control_plane/collaboration/peers.py | 26 +- .../control_plane/effect_runtime_handlers.ts | 1 + loopx/extensions/codex_native_child.py | 18 +- .../test_codex_native_child_receipts.py | 8 + .../test_native_child_receipts.py | 47 ++++ .../native_child_result_inbox.test.ts | 121 +++++++++ tests/test_manager_context_tracking.py | 36 +++ 13 files changed, 672 insertions(+), 18 deletions(-) create mode 100644 loopx/capabilities/multi_subagent/native_child_result_inbox.py create mode 100644 loopx/control_plane/capabilities/native_child_result_inbox.ts create mode 100644 tests/control_plane_ts/native_child_result_inbox.test.ts diff --git a/docs/integrations/host-native-child-receipts.md b/docs/integrations/host-native-child-receipts.md index fb3a7c0144..c928c77f00 100644 --- a/docs/integrations/host-native-child-receipts.md +++ b/docs/integrations/host-native-child-receipts.md @@ -92,7 +92,9 @@ Agent 明确记录验收；宿主完成不代表证据已被采纳。 公开安全的证据与验证引用。结果和不采用评审的原因码可选，只需是紧凑的 不透明诊断标识，不限定供应商词表，也不授予权限；未知时省略。结果与评审 可重复携带一致的 `operation`、`entrypoint-id`，冲突值包括重放时仍被拒绝。 - 原始提示、宿主错误、对话、本地路径和凭据不进入回执。 + 原始提示、宿主错误、对话、本地路径和凭据不进入公开回执。支持的宿主若能 + 直接观察最终结果正文，会把正文写入按 Goal 实例和主 Agent 隔离的本机私有 + inbox；公开活动投影仍只保留类型化终态和不透明身份。 4. 同一身份和内容重放幂等，内容冲突会被拒绝。`read` 与带 Turn ID 的 `agent-context` 读取同一模型。Goal 状态的 JSON 与 Markdown 只在能力启用且确有决策时投影 最近一轮；仪表板读取该投影，未配置或无关 Goal 不显示活动行。 @@ -102,6 +104,38 @@ Agent 明确记录验收；宿主完成不代表证据已被采纳。 新写入在既有事件流锁内重新核对读回，每次报告只调用一个粗粒度 TS 边界， 不增加第二套 Python 阶段判断。 +## Next-Turn result delivery / 下一 Turn 结果投递 + +Supported managed hosts persist an observed child's final text before exposing +the matching typed terminal receipt. The text is stored only in the +owner-private local runtime, partitioned by the captured Goal lifetime and +coordinator (or by Goal ID for a legacy registry without lifetime identity). It +never enters the public rollout log, status projection, telemetry, or repository +state. + +The existing `manager-inbox` turn-start hook detects unread native results and +adds one `before_work` read. A normal `manager-inbox read` returns short results +inline and gives a digest-bound private file reference for larger results. That +read records delivery but does not consume the result. Until the parent records +an `accepted`, `deferred`, or `rejected` disposition with the supplied +`native-child acknowledge-result` command, a restart may deliver the same stable +result ID again. Exact ingest and acknowledgement replay are idempotent; +conflicting text or disposition is rejected. A child conclusion remains +candidate evidence and never performs parent work, validation, or adoption. + +支持的托管宿主会先把观察到的子 Agent 最终正文写入主 Agent 的本机私有 +runtime，再公开对应的类型化终态回执。存储按已捕获的 Goal 生命周期和主 Agent +隔离；缺少生命周期身份的旧 registry 按 Goal ID 隔离。正文不会进入公开 rollout +日志、状态投影、遥测或仓库状态。 + +既有 `manager-inbox` Turn-start hook 会发现未读结果，并在下一 Turn 的普通工作 +之前要求读取。`manager-inbox read` 对短结果内联返回，对长结果返回带摘要校验的 +私有文件引用。读取只记录 delivery，不会消费结果。主 Agent 使用返回的 +`native-child acknowledge-result` 命令明确记录 `accepted`、`deferred` 或 +`rejected` 之前，重启后仍可能看到同一个稳定结果 ID。相同写入与确认重放幂等， +正文或 disposition 冲突会被拒绝。子 Agent 结论始终只是候选证据，不会替主 +Agent 执行工作、验证或采纳。 + After a lawful deferred-Todo recovery, rerun `quota should-run` with the same explicit Turn and Todo. If the original bound guard denied delivery and the current guard explicitly admits work, quota appends a work-qualification receipt @@ -196,7 +230,8 @@ replaying the same binding and item remains idempotent. Direct enabled CLI adapter calls require that attempt; feature-off calls retain their original request. Only compact hashed child references are retained for correlation; they do not enter public activity rows or replace independent parent review. -No prompt or raw result is added to a receipt. +No prompt or raw result is added to a public receipt. Supported managed hosts +retain final result text only in the private next-Turn inbox described above. 恢复 CLI 时可能只收到新的 `wait` 完成事件。适配器从同一已准入 Goal 实例、 主 Agent、LoopX Turn 的最近一次已启动宿主 spawn 或 followup 回执恢复关联， @@ -211,7 +246,8 @@ No prompt or raw result is added to a receipt. app-server 使用其原生 Turn ID。真实重试在启动前递增尝试次数；同一绑定与 事件的重放仍幂等。直接调用已启用的 CLI 适配器须提供该尝试次数，关闭能力 时请求不变。关联只保留紧凑的子代理哈希引用，不进入公开活动行， -也不替代独立父任务验收。回执不增加原始提示或结果内容。 +也不替代独立父任务验收。公开回执不增加原始提示或结果内容；支持的托管宿主仅在 +上述私有下一 Turn inbox 中保留最终结果正文。 A live isolated Codex 0.142.5 test observed one successful spawn, a second failed spawn at `agents.max_threads=1`, child completion and independent parent diff --git a/loopx/capabilities/manager_context/__init__.py b/loopx/capabilities/manager_context/__init__.py index 82a3d41d5f..4b67e6cf90 100644 --- a/loopx/capabilities/manager_context/__init__.py +++ b/loopx/capabilities/manager_context/__init__.py @@ -257,6 +257,9 @@ def produce(): count = ( max(len(inbox["items"]), agenda["incoming_count"]) + max(len(inbox.get("peer_returns", {}).get("items", [])), agenda["outgoing_count"]) + + len(inbox.get("native_child_results", {}).get("items", [])) + + int(inbox.get("native_child_results", {}).get("unavailable_count", 0)) + + int(inbox.get("native_child_results", {}).get("scan_complete") is False) + len(inbox.get("operation_handoffs", [])) ) incomplete = not agenda["scan_complete"] diff --git a/loopx/capabilities/multi_subagent/cli.py b/loopx/capabilities/multi_subagent/cli.py index 420a2005ce..5e4fceadec 100644 --- a/loopx/capabilities/multi_subagent/cli.py +++ b/loopx/capabilities/multi_subagent/cli.py @@ -11,6 +11,7 @@ from ...control_plane.goals.source_session_registry_state import exact_goal_ref from ...orchestration import compact_orchestration_policy from .native_child_receipts import load_native_child_activity, record_native_child +from .native_child_result_inbox import acknowledge_native_child_result def register_native_child_commands(subparsers, add_format): @@ -18,7 +19,7 @@ def register_native_child_commands(subparsers, add_format): "native-child", help="Record or read typed host-native child activity for an admitted Turn." ) add_format(parser) - parser.add_argument("native_child_action", choices=("record", "read")) + parser.add_argument("native_child_action", choices=("record", "read", "acknowledge-result")) parser.add_argument("--goal-id", required=True) parser.add_argument("--agent-id", required=True) parser.add_argument("--turn-instance-id", required=True) @@ -35,6 +36,9 @@ def register_native_child_commands(subparsers, add_format): "for result/deferred/rejected. Accepted reviews use evidence/validation refs instead.") parser.add_argument("--evidence-ref", help="Public-safe opaque reference for accepted evidence.") parser.add_argument("--validation-ref", help="Public-safe opaque reference for parent validation.") + parser.add_argument("--result-id", help="Stable native child result identity returned by manager-inbox read.") + parser.add_argument("--disposition", choices=("accepted", "deferred", "rejected"), + help="Parent disposition after the result was delivered and reviewed.") parser.add_argument("--execute", action="store_true", help="Append the receipt; otherwise preview only.") @@ -56,27 +60,46 @@ def handle_native_child_command(args, registry_path, runtime_root, print_payload goal = load_goal_from_registry(registry_path, args.goal_id) if goal is None or args.agent_id not in registered_agent_ids_for_goal(goal): raise ValueError("coordinator is not registered for this Goal") - orchestration = compact_orchestration_policy(goal.get("spawn_policy")) - if not ( - orchestration.get("mode") == "multi_subagent" - and orchestration.get("spawn_allowed") is True - and int(orchestration.get("max_children") or 0) > 0 - ): - raise ValueError("native child receipts require enabled multi_subagent policy") - configured_limit = int(orchestration["max_children"]) + configured_limit = None + if args.native_child_action != "acknowledge-result": + orchestration = compact_orchestration_policy(goal.get("spawn_policy")) + if not ( + orchestration.get("mode") == "multi_subagent" + and orchestration.get("spawn_allowed") is True + and int(orchestration.get("max_children") or 0) > 0 + ): + raise ValueError("native child receipts require enabled multi_subagent policy") + configured_limit = int(orchestration["max_children"]) if args.native_child_action == "read": if any((args.operation_id, args.stage, args.operation, args.outcome, args.entrypoint_id, args.reason_code, args.evidence_ref, - args.validation_ref, args.execute)): + args.validation_ref, args.result_id, args.disposition, args.execute)): raise ValueError("read accepts only goal, agent and Turn identity") payload = {"ok": True, "native_child_activity": load_native_child_activity( runtime_root, goal_id=args.goal_id, agent_id=args.agent_id, turn_instance_id=args.turn_instance_id, configured_limit=configured_limit, registry_path=registry_path, goal_ref=goal_ref, )} + elif args.native_child_action == "acknowledge-result": + if any((args.operation_id, args.stage, args.operation, args.outcome, + args.entrypoint_id, args.reason_code, args.evidence_ref, + args.validation_ref)) or not args.result_id or not args.disposition: + raise ValueError("acknowledge-result requires only result-id and disposition") + if not args.execute: + payload = {"ok": True, "preview": True, "result_id": args.result_id, + "disposition": args.disposition} + else: + payload = acknowledge_native_child_result( + runtime_root=runtime_root, goal_id=args.goal_id, + agent_id=args.agent_id, turn_instance_id=args.turn_instance_id, + result_id=args.result_id, + disposition=args.disposition, goal_ref=goal_ref, + ) else: if not args.operation_id or not args.stage or not args.outcome: raise ValueError("record requires operation-id, stage and outcome") + if args.result_id or args.disposition: + raise ValueError("record does not accept result acknowledgement fields") payload = record_native_child( runtime_root=runtime_root, goal_id=args.goal_id, agent_id=args.agent_id, turn_instance_id=args.turn_instance_id, operation_id=args.operation_id, @@ -96,6 +119,11 @@ def handle_native_child_command(args, registry_path, runtime_root, print_payload def render_native_child(payload): if not payload["ok"]: return str(payload["error"]) + if "native_child_activity" not in payload: + return ( + f"Native child result {payload['result_id']} " + f"{'would be ' if payload.get('preview') else ''}{payload['disposition']}." + ) activity = payload["native_child_activity"] return ( f"Native child activity for {activity['turn_instance_id']}: " diff --git a/loopx/capabilities/multi_subagent/native_child_result_inbox.py b/loopx/capabilities/multi_subagent/native_child_result_inbox.py new file mode 100644 index 0000000000..3ac1a99a77 --- /dev/null +++ b/loopx/capabilities/multi_subagent/native_child_result_inbox.py @@ -0,0 +1,49 @@ +"""Thin Python adapter for the TypeScript native-child result inbox owner.""" + +from __future__ import annotations + +from collections.abc import Mapping +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from ...control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result + + +def _now() -> str: + return datetime.now(timezone.utc).isoformat() + + +def ingest_native_child_result( + *, runtime_root: Path, goal_id: str, agent_id: str, + turn_instance_id: str, operation_id: str, outcome: str, text: str, + goal_ref: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + try: + return effect_runtime_result("multi_subagent.native_result_inbox", { + "operation": "ingest", "runtime_root": str(runtime_root.resolve()), + "goal_id": goal_id, "agent_id": agent_id, + "goal_ref": dict(goal_ref) if goal_ref is not None else None, + "turn_instance_id": turn_instance_id, "operation_id": operation_id, + "outcome": outcome, "text": text, "observed_at": _now(), + }) + except EffectRuntimeRejected as exc: + raise ValueError(str(exc)) from exc + + +def acknowledge_native_child_result( + *, runtime_root: Path, goal_id: str, agent_id: str, + turn_instance_id: str, result_id: str, disposition: str, + goal_ref: Mapping[str, Any] | None = None, +) -> dict[str, Any]: + try: + return effect_runtime_result("multi_subagent.native_result_inbox", { + "operation": "acknowledge", "runtime_root": str(runtime_root.resolve()), + "goal_id": goal_id, "agent_id": agent_id, + "goal_ref": dict(goal_ref) if goal_ref is not None else None, + "turn_instance_id": turn_instance_id, + "result_id": result_id, "disposition": disposition, + "observed_at": _now(), + }) + except EffectRuntimeRejected as exc: + raise ValueError(str(exc)) from exc diff --git a/loopx/control_plane/capabilities/native_child_result_inbox.ts b/loopx/control_plane/capabilities/native_child_result_inbox.ts new file mode 100644 index 0000000000..d20f10dd12 --- /dev/null +++ b/loopx/control_plane/capabilities/native_child_result_inbox.ts @@ -0,0 +1,257 @@ +/** Durable, owner-private delivery for host-native child results. + * + * The public rollout log keeps only typed lifecycle receipts. Result text stays + * in this local private store and is projected to the parent on a later Turn. + */ +import {createHash} from "node:crypto"; +import {constants} from "node:fs"; +import {open, readdir} from "node:fs/promises"; +import {isAbsolute, join, toNamespacedPath} from "node:path"; + +import type {JsonObject} from "../effect_program.ts"; +import {EffectRuntimeRequestError} from "../effect_runtime_errors.ts"; +import {atomicWriteJson, atomicWriteText, withFileMutationLock} from "../effect_runtime_io.ts"; +import {jsonObject, requireJsonObject, requireNonEmptyString, requireStringLiteral} from "../runtime_decode.ts"; + +export const NATIVE_CHILD_RESULT_ENTRY_SCHEMA = "native_child_result_entry_v0"; +export const NATIVE_CHILD_RESULT_INBOX_SCHEMA = "native_child_result_inbox_v0"; + +const ID = /^[A-Za-z0-9][A-Za-z0-9._:/-]{0,191}$/u; +const RESULT_ID = /^ncr_[0-9a-f]{32}$/u; +const OUTCOMES = ["completed", "failed", "cancelled"] as const; +const DISPOSITIONS = ["accepted", "deferred", "rejected"] as const; +const MAX_CONTENT_BYTES = 4 * 1024 * 1024; +const INLINE_CONTENT_BYTES = 32 * 1024; +const MAX_INLINE_PAGE_BYTES = 128 * 1024; +const MAX_SCAN = 2_000; +const PAGE_SIZE = 20; + +function fail(message: string): never { throw new EffectRuntimeRequestError(message); } +function opaque(value: unknown, label: string): string { + const result = requireNonEmptyString(value, label); + if (!ID.test(result)) fail(`${label} is invalid`); + return result; +} +function instant(value: unknown, label: string): string { + const result = requireNonEmptyString(value, label); + if (result.length > 80 || !/^\d{4}-\d{2}-\d{2}T[^\s]{1,64}$/u.test(result)) fail(`${label} is invalid`); + return result; +} +function goalRef(value: unknown, goalId: string): JsonObject | null { + if (value === null || value === undefined) return null; + const ref = requireJsonObject(value, "Goal reference"); + if (Object.keys(ref).sort().join(",") !== "goal_id,goal_instance_id" + || ref.goal_id !== goalId) fail("Goal reference is invalid"); + return {goal_id: goalId, goal_instance_id: opaque(ref.goal_instance_id, "Goal instance")}; +} +function canonical(value: unknown): string { + if (Array.isArray(value)) return `[${value.map(canonical).join(",")}]`; + const row = jsonObject(value); + if (row) return `{${Object.keys(row).sort().map(key => `${JSON.stringify(key)}:${canonical(row[key])}`).join(",")}}`; + return JSON.stringify(value ?? null); +} +function digest(value: unknown): string { return createHash("sha256").update(canonical(value)).digest("hex"); } +function scope(params: JsonObject): {root: string; goalId: string; agentId: string; goalRef: JsonObject | null} { + const runtime = requireNonEmptyString(params.runtime_root, "runtime root"); + if (!isAbsolute(runtime)) fail("native child result runtime root must be absolute"); + const goalId = opaque(params.goal_id, "Goal"), agentId = opaque(params.agent_id, "Agent"); + const ref = goalRef(params.goal_ref, goalId); + const key = digest({goal_ref: ref, goal_id: ref ? null : goalId, agent_id: agentId}); + return {root: toNamespacedPath(join(runtime, ".local", "native-child-result-inbox", key)), + goalId, agentId, goalRef: ref}; +} +function resultIdentity(source: ReturnType, turn: string, operation: string): string { + return `ncr_${digest({goal_ref: source.goalRef, goal_id: source.goalRef ? null : source.goalId, + agent_id: source.agentId, turn_instance_id: turn, operation_id: operation}).slice(0, 32)}`; +} +function metadataPath(root: string, id: string): string { return join(root, `${id}.json`); } +function contentPath(root: string, id: string): string { return join(root, `${id}.txt`); } +async function readJson(path: string): Promise { + const handle = await open(path, constants.O_RDONLY | (constants.O_NONBLOCK ?? 0) + | (constants.O_NOFOLLOW ?? 0)); + try { + const info = await handle.stat(); + if (!info.isFile() || info.size > 128_000) fail("native child result metadata is invalid"); + const text = await handle.readFile("utf8"); + return requireJsonObject(JSON.parse(text), "native child result metadata"); + } catch (error) { + if (error instanceof SyntaxError || error instanceof TypeError) fail("native child result metadata is invalid"); + throw error; + } finally { await handle.close(); } +} +function identityMatches(row: JsonObject, source: ReturnType): boolean { + return row.schema_version === NATIVE_CHILD_RESULT_ENTRY_SCHEMA + && row.goal_id === source.goalId && row.agent_id === source.agentId + && canonical(row.goal_ref ?? null) === canonical(source.goalRef); +} +async function readEntry(source: ReturnType, id: string): Promise { + if (!RESULT_ID.test(id)) fail("native child result id is invalid"); + const row = await readJson(metadataPath(source.root, id)); + if (!identityMatches(row, source) || row.result_id !== id) fail("native child result scope mismatch"); + const expected = ["acknowledgement", "agent_id", "content_bytes", "content_sha256", "created_at", + "delivery", "goal_id", "goal_ref", "operation_id", "outcome", "result_id", "schema_version", + "turn_instance_id"]; + if (Object.keys(row).sort().join(",") !== expected.join(",")) fail("native child result metadata is invalid"); + opaque(row.turn_instance_id, "Turn"); + opaque(row.operation_id, "operation"); + requireStringLiteral(row.outcome, [...OUTCOMES], "native child result outcome"); + if (typeof row.content_sha256 !== "string" || !/^sha256:[0-9a-f]{64}$/u.test(row.content_sha256) + || !Number.isSafeInteger(row.content_bytes) || Number(row.content_bytes) < 0 + || Number(row.content_bytes) > MAX_CONTENT_BYTES) fail("native child result content identity is invalid"); + instant(row.created_at, "result creation time"); + const delivery = row.delivery === null ? null : jsonObject(row.delivery); + if (row.delivery !== null && (!delivery + || Object.keys(delivery).sort().join(",") !== "delivery_count,first_delivered_at,last_delivered_at" + || !Number.isSafeInteger(delivery.delivery_count) || Number(delivery.delivery_count) < 1 + || Number(delivery.delivery_count) > 1_000_000)) fail("native child result delivery is invalid"); + if (delivery) { + instant(delivery.first_delivered_at, "first delivery time"); + instant(delivery.last_delivered_at, "last delivery time"); + } + const acknowledgement = row.acknowledgement === null ? null : jsonObject(row.acknowledgement); + if (row.acknowledgement !== null && (!acknowledgement + || Object.keys(acknowledgement).sort().join(",") !== "acknowledged_at,disposition")) { + fail("native child result acknowledgement is invalid"); + } + if (acknowledgement) { + if (!delivery) fail("native child result acknowledgement has no delivery"); + instant(acknowledgement.acknowledged_at, "acknowledgement time"); + requireStringLiteral(acknowledgement.disposition, [...DISPOSITIONS], "native child result disposition"); + } + return row; +} +async function readContent(path: string): Promise { + const handle = await open(path, constants.O_RDONLY | (constants.O_NONBLOCK ?? 0) + | (constants.O_NOFOLLOW ?? 0)); + try { + const info = await handle.stat(); + if (!info.isFile() || info.size > MAX_CONTENT_BYTES) fail("native child result content is invalid"); + return await handle.readFile("utf8"); + } finally { await handle.close(); } +} +async function ingest(params: JsonObject): Promise { + const source = scope(params); + const turn = opaque(params.turn_instance_id, "Turn"), operation = opaque(params.operation_id, "operation"); + const outcome = requireStringLiteral(params.outcome, [...OUTCOMES], "native child result outcome"); + const observedAt = instant(params.observed_at, "result observation time"); + if (typeof params.text !== "string") fail("native child result text is required"); + const text = params.text; + if (/\p{Surrogate}/u.test(text)) fail("native child result text encoding is invalid"); + const bytes = Buffer.byteLength(text, "utf8"); + if (bytes > MAX_CONTENT_BYTES) fail("native child result text is too large"); + const id = resultIdentity(source, turn, operation), path = metadataPath(source.root, id); + const contentSha256 = `sha256:${createHash("sha256").update(text, "utf8").digest("hex")}`; + return withFileMutationLock(path, async () => { + let existing: JsonObject | null = null; + try { existing = await readEntry(source, id); } + catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; } + const identity: JsonObject = {schema_version: NATIVE_CHILD_RESULT_ENTRY_SCHEMA, result_id: id, + goal_id: source.goalId, agent_id: source.agentId, goal_ref: source.goalRef, + turn_instance_id: turn, operation_id: operation, outcome, + content_sha256: contentSha256, content_bytes: bytes}; + if (existing) { + if (Object.entries(identity).some(([key, value]) => canonical(existing![key]) !== canonical(value))) { + fail("native child result identity already has conflicting content"); + } + return {schema_version: NATIVE_CHILD_RESULT_INBOX_SCHEMA, ok: true, result_id: id, + created: false, replay: true}; + } + await atomicWriteText(contentPath(source.root, id), text); + await atomicWriteJson(path, {...identity, created_at: observedAt, delivery: null, acknowledgement: null}); + return {schema_version: NATIVE_CHILD_RESULT_INBOX_SCHEMA, ok: true, result_id: id, + created: true, replay: false}; + }); +} +async function listEntries(params: JsonObject): Promise { + const source = scope(params); + let names: string[] = []; + try { names = (await readdir(source.root)).filter(name => /^ncr_[0-9a-f]{32}\.json$/u.test(name)).sort(); } + catch (error) { if ((error as NodeJS.ErrnoException).code !== "ENOENT") throw error; } + const scanComplete = names.length <= MAX_SCAN; + names = names.slice(0, MAX_SCAN); + const rows: JsonObject[] = [], unavailable: string[] = []; + for (const name of names) { + const id = name.slice(0, -5); + try { + const row = await readEntry(source, id); + if (row.acknowledgement === null) rows.push(row); + } catch { unavailable.push(id); } + } + rows.sort((left, right) => Number(left.delivery !== null) - Number(right.delivery !== null) + || String(left.created_at).localeCompare(String(right.created_at)) || String(left.result_id).localeCompare(String(right.result_id))); + const items: JsonObject[] = []; + let inlineBytes = 0; + let inspected = 0; + for (const row of rows) { + if (items.length >= PAGE_SIZE) break; + inspected += 1; + const file = contentPath(source.root, String(row.result_id)); + const base: JsonObject = Object.fromEntries(["result_id", "turn_instance_id", "operation_id", "outcome", + "content_sha256", "content_bytes", "created_at", "delivery"].map(key => [key, row[key]])); + const size = Number(row.content_bytes); + try { + if (Number.isSafeInteger(size) && size >= 0 && size <= INLINE_CONTENT_BYTES + && inlineBytes + size <= MAX_INLINE_PAGE_BYTES) { + const text = await readContent(file); + if (`sha256:${createHash("sha256").update(text, "utf8").digest("hex")}` !== row.content_sha256) { + unavailable.push(String(row.result_id)); continue; + } + base.text = text; inlineBytes += size; + } else { + const text = await readContent(file); + if (Buffer.byteLength(text, "utf8") !== size + || `sha256:${createHash("sha256").update(text, "utf8").digest("hex")}` !== row.content_sha256) { + unavailable.push(String(row.result_id)); continue; + } + base.content_file = file; + } + } catch { + unavailable.push(String(row.result_id)); continue; + } + items.push(base); + } + return {schema_version: NATIVE_CHILD_RESULT_INBOX_SCHEMA, ok: true, items, + has_more: !scanComplete || inspected < rows.length, pending_count: rows.length, + scan_complete: scanComplete, + unavailable_count: unavailable.length}; +} +async function markDelivered(params: JsonObject): Promise { + const source = scope(params), deliveredAt = instant(params.observed_at, "delivery time"); + if (!Array.isArray(params.result_ids) || params.result_ids.length > PAGE_SIZE) fail("bounded native child result ids are required"); + if (new Set(params.result_ids.map(id => requireNonEmptyString(id, "native child result id"))).size + !== params.result_ids.length) fail("native child result ids must be unique"); + for (const raw of params.result_ids) { + const id = requireNonEmptyString(raw, "native child result id"), path = metadataPath(source.root, id); + await withFileMutationLock(path, async () => { + const row = await readEntry(source, id), delivery = jsonObject(row.delivery); + await atomicWriteJson(path, {...row, delivery: {first_delivered_at: delivery?.first_delivered_at ?? deliveredAt, + last_delivered_at: deliveredAt, delivery_count: Math.min(Number(delivery?.delivery_count ?? 0) + 1, 1_000_000)}}); + }); + } + return {schema_version: NATIVE_CHILD_RESULT_INBOX_SCHEMA, ok: true, delivered_count: params.result_ids.length}; +} +async function acknowledge(params: JsonObject): Promise { + const source = scope(params), id = requireNonEmptyString(params.result_id, "native child result id"); + const turn = opaque(params.turn_instance_id, "Turn"); + const acknowledgedAt = instant(params.observed_at, "acknowledgement time"); + const disposition = requireStringLiteral(params.disposition, [...DISPOSITIONS], "native child result disposition"); + const path = metadataPath(source.root, id); + return withFileMutationLock(path, async () => { + const row = await readEntry(source, id), prior = jsonObject(row.acknowledgement); + if (row.turn_instance_id !== turn) fail("native child result Turn identity mismatch"); + if (!jsonObject(row.delivery)) fail("native child result must be delivered before acknowledgement"); + if (prior && prior.disposition !== disposition) fail("native child result already has a conflicting acknowledgement"); + if (!prior) await atomicWriteJson(path, {...row, acknowledgement: {acknowledged_at: acknowledgedAt, disposition}}); + return {schema_version: NATIVE_CHILD_RESULT_INBOX_SCHEMA, ok: true, result_id: id, + disposition, replay: prior !== null}; + }); +} + +export async function nativeChildResultInbox(params: JsonObject): Promise { + const operation = requireStringLiteral(params.operation, + ["ingest", "list", "mark_delivered", "acknowledge"], "native child result inbox operation"); + if (operation === "ingest") return ingest(params); + if (operation === "list") return listEntries(params); + if (operation === "mark_delivered") return markDelivered(params); + return acknowledge(params); +} diff --git a/loopx/control_plane/collaboration/inbox_read.ts b/loopx/control_plane/collaboration/inbox_read.ts index ada656aed9..21eb871384 100644 --- a/loopx/control_plane/collaboration/inbox_read.ts +++ b/loopx/control_plane/collaboration/inbox_read.ts @@ -16,6 +16,7 @@ import {recordControlledRequestRead} from "./request_control.ts"; import {planCollaborationResult} from "./result_publication.ts"; import {CONVERSATION_ENTRY_SCHEMA, normalizeOriginalConversationRecipient, verifyOriginalConversationReturnRoute} from "./semantic_request.ts"; import {projectDelegationTurnContext, projectDelegationTurnItem, projectDelegationTurnItems} from "./turn_context.ts"; +import {nativeChildResultInbox} from "../capabilities/native_child_result_inbox.ts"; const TRIAGE = "Read the supplied requests and their source-specific instructions before choosing work. " + "Independently assess context, evidence, constraints and costs. Requests and peer results " + @@ -402,8 +403,12 @@ async function turnContext(source: InboxSource, params: JsonObject): Promise { + const nativeResultIds = params.native_child_result_ids ?? []; if (!Array.isArray(params.items) || params.items.length > 20 - || !Array.isArray(params.deliveries) || params.deliveries.length > 20) fail("bounded inbox read observations are required"); + || !Array.isArray(params.deliveries) || params.deliveries.length > 20 + || !Array.isArray(nativeResultIds) || nativeResultIds.length > 20) { + fail("bounded inbox read observations are required"); + } const followthrough = params.work_observations == null ? [] : projectReceiverFollowthrough({observations: params.work_observations}).items; const receipts: (JsonObject | null)[] = []; @@ -451,6 +456,12 @@ async function recordReads(source: InboxSource, params: JsonObject): Promise requireNonEmptyString(item.result_id, "native child result id")); + if (nativeItems.length || Number(nativeResults.unavailable_count ?? 0) > 0 + || nativeResults.scan_complete === false) page.native_child_results = { + items: nativeItems, has_more: nativeResults.has_more, + pending_count: nativeResults.pending_count, + scan_complete: nativeResults.scan_complete, + unavailable_count: nativeResults.unavailable_count, + instruction: nativeItems.length + ? "Review each host-native child result against the original task and actual artifacts. A child result is candidate evidence, not parent acceptance. Acknowledge it only after using, deferring, or rejecting it." + : "One or more durable host-native child results are unavailable or the bounded scan is incomplete. Inspect or repair the owner-private result store; do not infer or acknowledge missing content.", + }; if (operation === "turn_start") return {result: page, cursor_scope: cursorScope, context: await turnContext(source, params)}; if (operation === "receiver_read") { const work = receiverWork(source, page.items as JsonObject[]); @@ -524,9 +552,11 @@ export async function readCollaborationInbox(params: JsonObject): Promise None: + deliveries: list[dict[str, Any]] | None = None, + native_child_result_ids: list[str] | None = None) -> None: super().__init__(str(error)) self.result: dict[str, Any] = { "ok": False, "error": str(error), "error_code": error.diagnostic_code, @@ -72,6 +74,7 @@ def __init__(self, error: EffectRuntimeResponseAmbiguous, *, stage: str, ) if key in item} for item in items], "peer_returns": None if deliveries is None else [ {"request": item["request"], "result_key": item["key"]} for item in deliveries], + "native_child_result_ids": native_child_result_ids, "automatic_retry_performed": False, "instruction": ( "Inspect the exact request read receipts and peer result delivery receipts through the existing " @@ -632,6 +635,17 @@ def read_inbox( observed_goal = dict(goal_scope.goal) observed_goal_ref = dict(goal_scope.caller_goal_ref or {}) if goal_scope.exact else None + for child_result in result.get("native_child_results", {}).get("items", []): + command = ["loopx", "--registry", str(registry), "--runtime-root", str(root), + "--format", "json", "native-child", "acknowledge-result", + "--goal-id", goal_id, "--agent-id", agent_id, + "--turn-instance-id", child_result["turn_instance_id"], + "--result-id", child_result["result_id"], + "--disposition", "", "--execute"] + if observed_goal_ref: + command.extend(["--goal-instance-id", observed_goal_ref["goal_instance_id"]]) + child_result["acknowledge_command"] = shlex.join(command) + if not snapshot.get("completed"): work = receiver_work_observations(root, registry, result["items"]) # Input hashing remains outside Goal lifetime admission. @@ -648,11 +662,13 @@ def read_inbox( "operation": "record_read", "goal_id": goal_id, "agent_id": agent_id, "items": result["items"], "deliveries": snapshot["deliveries"], "work_observations": work, + "native_child_result_ids": snapshot.get("native_child_result_ids", []), "include_context": request_id is None, "read_command": read_command, }, scope=goal_scope) except EffectRuntimeResponseAmbiguous as exc: raise InboxReadResponseAmbiguous(exc, stage="record_read", scope=read_scope, - read_arguments=read_arguments, items=result["items"], deliveries=snapshot["deliveries"]) from exc + read_arguments=read_arguments, items=result["items"], deliveries=snapshot["deliveries"], + native_child_result_ids=snapshot.get("native_child_result_ids", [])) from exc else: final = snapshot for item in result["items"]: @@ -677,6 +693,12 @@ def read_inbox( "Finish the original request with return_result, including evidence and remaining gaps. " "Adoption, file hashes and returned opinions are not independent acceptance or Todo completion." ) + if result.get("native_child_results", {}).get("items"): + result["followthrough"] += ( + " For each native_child_result, inspect the result and its evidence, then run native-child " + "acknowledge-result using its acknowledge_command and accepted/deferred/rejected disposition. " + "Until acknowledgement, the same stable result may be delivered again after restart." + ) if request_id is not None: result["followthrough"] = ( "Inspect this exact request's current source, revisions and receipts. " diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 8e3922cb15..5e45fc2eda 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -669,6 +669,7 @@ export function createEffectRuntimeHandlers( ["collaboration.request.amendment_readback", lazyHandler(() => import("./collaboration/request_control.ts"), ({requestAmendmentReadback}) => requestAmendmentReadback)], ["goal.amendment.request_receipt", lazyHandler(() => import("./goals/goal_amendment_authority.ts"), ({projectRequestAmendmentReceipt}) => projectRequestAmendmentReceipt)], ["collaboration.inbox.read", lazyHandler(() => import("./collaboration/inbox_read.ts"), ({readCollaborationInbox}) => readCollaborationInbox)], + ["multi_subagent.native_result_inbox", lazyHandler(() => import("./capabilities/native_child_result_inbox.ts"), ({nativeChildResultInbox}) => nativeChildResultInbox)], ["collaboration.result.plan_publication", lazyHandler(() => import("./collaboration/result_publication.ts"), ({planCollaborationResult}) => planCollaborationResult)], ["collaboration.result.attachment_refs", lazyHandler(() => import("./collaboration/result_publication.ts"), ({resultAttachmentRefs}) => resultAttachmentRefs)], ["collaboration.result.delivery_ready", lazyHandler(() => import("./collaboration/result_publication.ts"), ({collaborationResultDeliveryReady}) => collaborationResultDeliveryReady)], diff --git a/loopx/extensions/codex_native_child.py b/loopx/extensions/codex_native_child.py index e0fdc9b889..a0cf5a448f 100644 --- a/loopx/extensions/codex_native_child.py +++ b/loopx/extensions/codex_native_child.py @@ -1,7 +1,8 @@ """Transient Codex host events adapted to native-child receipts. Only opaque identities and typed outcomes reach the existing multi_subagent -log. Prompts, child messages and raw tool output are never retained. +log. A supported host may retain the final child message in the coordinator's +owner-private local inbox; prompts and raw tool output are never retained. """ from __future__ import annotations @@ -14,6 +15,9 @@ from ..capabilities.multi_subagent.native_child_receipts import ( _load_native_child_events, record_native_child, ) +from ..capabilities.multi_subagent.native_child_result_inbox import ( + ingest_native_child_result, +) def configured_native_child_limit(request: Mapping[str, Any]) -> int | None: @@ -160,6 +164,18 @@ def observe(self, item: Mapping[str, Any], *, session_id: str, invocation_id: st wait_ref = "codex-wait-" + hashlib.sha256(identity.encode()).hexdigest()[:32] operation_id = self._restore_operation(child, wait_ref=wait_ref) if wait_ref else decision_id if operation_id is not None: + message = state.get("message") + if isinstance(message, str): + ingest_native_child_result( + runtime_root=self.runtime_root, + goal_id=self.lineage["goal_id"], + agent_id=self.lineage["agent_id"], + turn_instance_id=self.turn_instance_id, + operation_id=operation_id, + outcome=outcome, + text=message, + goal_ref=self.goal_ref, + ) self._record(stage="result", operation_id=operation_id, outcome=outcome, _host_wait_ref=wait_ref) diff --git a/tests/capabilities/test_codex_native_child_receipts.py b/tests/capabilities/test_codex_native_child_receipts.py index ee6d13ffae..321c33e58d 100644 --- a/tests/capabilities/test_codex_native_child_receipts.py +++ b/tests/capabilities/test_codex_native_child_receipts.py @@ -56,6 +56,14 @@ def test_host_spawn_result_parent_review_and_restart(tmp_path: Path): assert len(events) == 4 assert "private child" not in json.dumps(events) assert all(event.get("event_kind") != "quota_spend" for event in events) + from loopx.control_plane.effect_runtime import effect_runtime_result + inbox = effect_runtime_result("multi_subagent.native_result_inbox", { + "operation": "list", "runtime_root": str(tmp_path), + "goal_id": GOAL, "agent_id": AGENT, "goal_ref": None, + }) + assert len(inbox["items"]) == 1 + assert inbox["items"][0]["text"] == "private child result" + assert inbox["items"][0]["operation_id"] == operation["operation_id"] @pytest.mark.parametrize("items", [[], [{"type": "agentMessage", "text": "I spawned three children"}], diff --git a/tests/capabilities/test_native_child_receipts.py b/tests/capabilities/test_native_child_receipts.py index 34d91f8786..a571cbe3dc 100644 --- a/tests/capabilities/test_native_child_receipts.py +++ b/tests/capabilities/test_native_child_receipts.py @@ -304,6 +304,53 @@ def test_cli_readback_and_disabled_policy_do_not_mutate(tmp_path: Path): assert "enabled multi_subagent" in json.loads(rejected.stdout)["error"] +def test_cli_acknowledges_only_a_delivered_native_child_result(tmp_path: Path): + from loopx.capabilities.multi_subagent.native_child_result_inbox import ( + ingest_native_child_result, + ) + from loopx.control_plane.effect_runtime import effect_runtime_result + + registry = tmp_path / "registry.json" + registry.write_text(json.dumps({"schema_version": 1, "goals": [{ + "id": GOAL, "repo": str(tmp_path), "status": "active", + "registered_agents": [AGENT], + "spawn_policy": {"mode": "multi_subagent", "allowed": True, "max_children": 6}, + }]})) + runtime_root = tmp_path / "runtime" + created = ingest_native_child_result( + runtime_root=runtime_root, goal_id=GOAL, agent_id=AGENT, + turn_instance_id=TURN, operation_id="op-result-1", + outcome="completed", text="Review result.", + ) + effect_runtime_result("multi_subagent.native_result_inbox", { + "operation": "mark_delivered", "runtime_root": str(runtime_root), + "goal_id": GOAL, "agent_id": AGENT, "goal_ref": None, + "result_ids": [created["result_id"]], + "observed_at": "2026-11-13T00:00:00.000000+00:00", + }) + command = [sys.executable, "-m", "loopx.cli", "--registry", str(registry), + "--runtime-root", str(runtime_root), "--format", "json", + "native-child", "--goal-id", GOAL, "--agent-id", AGENT, + "--turn-instance-id", TURN, "acknowledge-result", + "--result-id", created["result_id"], "--disposition", "accepted"] + preview = subprocess.run(command, text=True, capture_output=True) + assert preview.returncode == 0, preview.stderr + assert json.loads(preview.stdout)["preview"] is True + assert len(effect_runtime_result("multi_subagent.native_result_inbox", { + "operation": "list", "runtime_root": str(runtime_root), + "goal_id": GOAL, "agent_id": AGENT, "goal_ref": None, + })["items"]) == 1 + disabled = json.loads(registry.read_text()) + disabled["goals"][0]["spawn_policy"]["allowed"] = False + registry.write_text(json.dumps(disabled)) + applied = subprocess.run([*command, "--execute"], text=True, capture_output=True) + assert applied.returncode == 0, applied.stderr + assert effect_runtime_result("multi_subagent.native_result_inbox", { + "operation": "list", "runtime_root": str(runtime_root), + "goal_id": GOAL, "agent_id": AGENT, "goal_ref": None, + })["items"] == [] + + def test_goal_status_only_attaches_reported_activity_to_enabled_goal( tmp_path: Path, monkeypatch, ): diff --git a/tests/control_plane_ts/native_child_result_inbox.test.ts b/tests/control_plane_ts/native_child_result_inbox.test.ts new file mode 100644 index 0000000000..b067c82ed3 --- /dev/null +++ b/tests/control_plane_ts/native_child_result_inbox.test.ts @@ -0,0 +1,121 @@ +import assert from "node:assert/strict"; +import {readFile, rm, writeFile} from "node:fs/promises"; +import {mkdtempSync} from "node:fs"; +import {tmpdir} from "node:os"; +import {join} from "node:path"; +import test from "node:test"; + +import type {JsonObject} from "../../loopx/control_plane/effect_program.ts"; +import {nativeChildResultInbox} from "../../loopx/control_plane/capabilities/native_child_result_inbox.ts"; +import {readCollaborationInbox} from "../../loopx/control_plane/collaboration/inbox_read.ts"; + +const goalRef = {goal_id: "research", goal_instance_id: "ginst_" + "a".repeat(32)}; +const base = {goal_id: "research", agent_id: "coordinator", goal_ref: goalRef}; +const at = "2026-11-13T00:00:00.000000+00:00"; + +function fixture(t: test.TestContext) { + const runtime_root = mkdtempSync(join(tmpdir(), "loopx-native-result-")); + t.after(() => rm(runtime_root, {recursive: true, force: true})); + const call = (operation: string, extra: JsonObject = {}) => nativeChildResultInbox({ + operation, runtime_root, ...base, ...extra, + }); + const inbox = (operation: string, extra: JsonObject = {}) => readCollaborationInbox({ + operation, runtime_root, ...base, observed_at: at, + read_command: "loopx manager-inbox read", + scope: {profile_id: "source_session_v1", caller_goal_ref: goalRef, current_goal_ref: goalRef}, + ...extra, + }); + return {runtime_root, call, inbox}; +} + +test("a completed host result survives replay and enters the next Turn unread", async t => { + const f = fixture(t), input = {turn_instance_id: "turn-1", operation_id: "operation-1", + outcome: "completed", text: "Independent review passed.", observed_at: at}; + const created = await f.call("ingest", input), replay = await f.call("ingest", input); + assert.equal(created.created, true); + assert.equal(replay.replay, true); + assert.equal(replay.result_id, created.result_id); + + const turn = await f.inbox("turn_start"), page = turn.result as JsonObject; + const [result] = (page.native_child_results as JsonObject).items as JsonObject[]; + assert.equal(result.result_id, created.result_id); + assert.equal(result.text, input.text); + assert.equal(result.delivery, null); + + const delivered = await f.inbox("receiver_read"); + assert.equal(delivered.completed, true); + const next = await f.call("list"), [again] = next.items as JsonObject[]; + assert.equal((again.delivery as JsonObject).delivery_count, 1); + assert.equal(again.result_id, created.result_id); +}); + +test("acknowledgement follows delivery and restart replay does not duplicate the effect", async t => { + const f = fixture(t), input = {turn_instance_id: "turn-1", operation_id: "operation-1", + outcome: "completed", text: "Use this once.", observed_at: at}; + const created = await f.call("ingest", input); + await assert.rejects(f.call("acknowledge", {result_id: created.result_id, + turn_instance_id: "turn-1", disposition: "accepted", observed_at: at}), /delivered before acknowledgement/); + await f.inbox("receiver_read"); + const first = await f.call("acknowledge", {result_id: created.result_id, + turn_instance_id: "turn-1", disposition: "accepted", observed_at: at}); + const replay = await f.call("acknowledge", {result_id: created.result_id, + turn_instance_id: "turn-1", disposition: "accepted", observed_at: "2026-11-13T00:01:00.000000+00:00"}); + assert.equal(first.replay, false); + assert.equal(replay.replay, true); + assert.deepEqual((await f.call("list")).items, []); + await assert.rejects(f.call("acknowledge", {result_id: created.result_id, + turn_instance_id: "turn-1", disposition: "rejected", observed_at: at}), /conflicting acknowledgement/); +}); + +test("conflicting content cannot replace an operation result", async t => { + const f = fixture(t), input = {turn_instance_id: "turn-1", operation_id: "operation-1", + outcome: "completed", text: "Original", observed_at: at}; + await f.call("ingest", input); + await assert.rejects(f.call("ingest", {...input, text: "Replacement"}), /conflicting content/); + const [result] = (await f.call("list")).items as JsonObject[]; + assert.equal(result.text, "Original"); +}); + +test("large result text stays in a private file while the inbox keeps a verified reference", async t => { + const f = fixture(t), text = "结果".repeat(40_000); + await f.call("ingest", {turn_instance_id: "turn-1", operation_id: "operation-1", + outcome: "completed", text, observed_at: at}); + const [result] = (await f.call("list")).items as JsonObject[]; + assert.equal(result.text, undefined); + assert.equal(await readFile(String(result.content_file), "utf8"), text); + assert.match(String(result.content_sha256), /^sha256:[0-9a-f]{64}$/u); +}); + +test("corrupt content stays visibly unavailable and cannot starve later results", async t => { + const f = fixture(t), text = "a".repeat(40_000); + const first = await f.call("ingest", {turn_instance_id: "turn-1", operation_id: "operation-1", + outcome: "completed", text, observed_at: at}); + for (let index = 2; index <= 22; index += 1) { + await f.call("ingest", {turn_instance_id: `turn-${index}`, operation_id: `operation-${index}`, + outcome: "completed", text: `result-${index}`, observed_at: at}); + } + const listed = await f.call("list"), large = (listed.items as JsonObject[]) + .find(item => item.result_id === first.result_id); + assert.ok(large?.content_file); + await writeFile(String(large.content_file), "b".repeat(40_000), "utf8"); + const after = await f.call("list"); + assert.equal(after.unavailable_count, 1); + assert.equal((after.items as JsonObject[]).length, 20); + assert.equal(after.has_more, true); + const turn = await f.inbox("turn_start"), page = turn.result as JsonObject; + assert.equal((page.native_child_results as JsonObject).unavailable_count, 1); + await rm(String(large.content_file)); + const missing = await f.call("list"); + assert.equal(missing.unavailable_count, 1); + assert.equal((missing.items as JsonObject[]).length, 20); +}); + +test("Goal lifetime and Agent scope isolate private native results", async t => { + const f = fixture(t); + await f.call("ingest", {turn_instance_id: "turn-1", operation_id: "operation-1", + outcome: "completed", text: "Scoped", observed_at: at}); + assert.deepEqual((await nativeChildResultInbox({operation: "list", runtime_root: f.runtime_root, + ...base, agent_id: "another"})).items, []); + assert.deepEqual((await nativeChildResultInbox({operation: "list", runtime_root: f.runtime_root, + ...base, goal_ref: {...goalRef, goal_instance_id: "ginst_" + "b".repeat(32)}})).items, []); +}); diff --git a/tests/test_manager_context_tracking.py b/tests/test_manager_context_tracking.py index 4e486fb240..855fa0935e 100644 --- a/tests/test_manager_context_tracking.py +++ b/tests/test_manager_context_tracking.py @@ -67,6 +67,42 @@ def status(): assert status()["delivery"]["at"] == first["delivery"]["at"] +def test_next_turn_requires_reading_a_durable_native_child_result(fixture, capsys): + from loopx.capabilities.multi_subagent.native_child_result_inbox import ( + ingest_native_child_result, + ) + + root, registry, _session, _turn, _target = fixture + created = ingest_native_child_result( + runtime_root=root, + goal_id="research", + agent_id="worker", + turn_instance_id="turn-native-result", + operation_id="operation-native-result", + outcome="completed", + text="The independent check found one actionable boundary.", + ) + hook = turn_start_hook(root, registry, "research", "worker") + observed = hook.producer() + assert observed["status"] == "observed" + assert observed["agent_read_required"] is True + assert observed["observation_count"] == 1 + + args = Namespace(manager_inbox_action="read", goal_id="research", agent_id="worker") + assert handle_manager_inbox(args, registry, root) == 0 + read = json.loads(capsys.readouterr().out) + [result] = read["native_child_results"]["items"] + assert result["result_id"] == created["result_id"] + assert result["text"] == "The independent check found one actionable boundary." + assert "native-child acknowledge-result" in result["acknowledge_command"] + + # Delivery does not consume the result; without acknowledgement, restart + # projects the same stable identity again on the next Turn. + replay = turn_start_hook(root, registry, "research", "worker").producer() + assert replay["agent_read_required"] is True + assert replay["observation_count"] == 1 + + def test_legacy_decision_is_not_a_fabricated_read(fixture): root, registry, session, turn, target = fixture rid = deliver(root, registry, session=session, turn=turn, request=target)[