From bbc14682561f5ae71f606db05e391072a0cbb3fa Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Mon, 19 Oct 2026 04:11:39 +0000 Subject: [PATCH 1/6] refactor(collaboration): converge return delivery and restart recovery in TypeScript Signed-off-by: aihop <4408344+hhyykk@users.noreply.github.com> --- .../typescript-control-plane-migration-v0.md | 14 +- ...script-control-plane-migration-v0.zh-CN.md | 12 +- loopx/capabilities/manager_context/README.md | 18 +- .../capabilities/manager_context/roundtrip.py | 1156 +++-------------- .../collaboration/return_delivery.ts | 398 +++++- .../control_plane/effect_runtime_handlers.ts | 4 +- .../manager_return_transaction.test.ts | 228 ++++ tests/test_collaboration_goal_instance.py | 66 +- tests/test_manager_context_roundtrip.py | 113 +- 9 files changed, 936 insertions(+), 1073 deletions(-) create mode 100644 tests/control_plane_ts/manager_return_transaction.test.ts diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md index e3b01a16a5..8cdf66d055 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.md @@ -4,7 +4,7 @@ - Supersedes / closes: none - Proposed by: LoopX maintainers - Date: 2026-08-15 -- Last revised: 2026-10-16 +- Last revised: 2026-10-19 - Scope: an incremental, replacement-first migration of the LoopX control-plane core from Python to TypeScript without maintaining two semantic implementations @@ -46,6 +46,18 @@ Each page reuses the existing executable-revision request scope; operation records remain live reads, and the next page resolves the current code again. +Manager result return now uses the existing `collaboration/return_delivery.ts` +transaction for admission, provider-attempt checkpoints, retry/verification reduction +and atomic delivery receipts. The two Python exact/legacy delivery coordinators +and the private attempt-normalization RPC are retired together. Python retains +original Chat/authority observations, Host fences and real provider IO. Bounded +scan hints are rechecked under the write lock; an accepted provider attempt is +persisted before verification, without a second queue or a Python decision fallback. +Original request/result/route identities, historical codecs and public readback +remain. This implements one R3/T1–T2 return transaction cutover, not Inbox acceptance, +source-grant migration, installed adoption or live-provider qualification. + + ## Current delivery frontier (post-merge, 2026-10-16) diff --git a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md index eb238403b9..db25317c89 100644 --- a/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md +++ b/docs/architecture/rfcs/typescript-control-plane-migration-v0.zh-CN.md @@ -4,7 +4,7 @@ - 替代 / 关闭：无 - Proposed by：LoopX maintainers - Date：2026-08-15 -- Last revised：2026-10-16 +- Last revised：2026-10-19 - Scope：LoopX 控制面核心从 Python 到 TypeScript 的增量、replacement-first 迁移；不长期维护两份语义实现 - Tracking issue：[#3225](https://github.com/loopx-project/loopx/issues/3225) @@ -39,6 +39,16 @@ Python 保留有界地址扫描、请求者注册检查和实时 `Delegations.re 每页复用既有代码版本请求作用域；操作记录仍实时读取，下一页重新解析当前代码。 +管家结果回传现由既有 `collaboration/return_delivery.ts` 事务负责准入、provider +尝试检查点、重试／核验消解和投递回执的原子落盘。Python 的 exact／legacy 两套 +投递编排与私有 attempt 规范化 RPC 同时退役。Python 保留原始 Chat／授权事实读取、 +Host fence 和真实 provider IO。有界扫描只提供读取提示，写入锁内重新检查；provider +接受写入后的尝试记录先落盘，再继续核验，不增加第二份队列或 Python 决策回退。 +原 request／result／route 身份、历史 codec 和公开回读保留。这实现 R3/T1–T2 的 +一个完整回传事务改造，不等于 Inbox 接收、来源授权迁移、安装采用或真实 provider +资格已完成。 + + ## 当前交付边界（合入后，2026-10-16） diff --git a/loopx/capabilities/manager_context/README.md b/loopx/capabilities/manager_context/README.md index b621a77a66..2784a98e1f 100644 --- a/loopx/capabilities/manager_context/README.md +++ b/loopx/capabilities/manager_context/README.md @@ -609,10 +609,20 @@ read-only verification. A matching message advances the original delivery to message, or verified mismatch becomes `explicit_unverified`. CLI, Manager read, and Chat expose the same public-safe state and reason without returning the provider locator. The Lark adapter keeps locator interpretation and provider -readback; `manager-context` remains the sole result/delivery writer. The typed -`control_plane/collaboration/return_delivery.ts` boundary owns provider-neutral -attempt validation and verification classification; Python retains file-lock, -persistence and adapter orchestration only. +readback. The existing `control_plane/collaboration/return_delivery.ts` owner now +performs delivery admission, state transitions, backoff and atomic delivery-receipt +writes for exact-instance and legacy routes. It reuses the request, result and +Goal-lifetime owners. Python observes original Chat/authority facts, holds the +existing Host fences and invokes the actual transcript/provider adapters. + +The obsolete Python exact/legacy delivery coordinators and attempt-normalization +RPC are removed. The shared transaction accepts bounded scan hints, rechecks each +selected receipt under its mutation lock, and uses source/revision and exact +admission-token fences before persistence. Attempt-aware providers persist their +acceptance checkpoint before verification continues; it is not deferred to a final callback. +The same stored request/result/route identities and public delivery readback remain. +This retires this delivery orchestration family, not Inbox acceptance, source grants, +all Python IO or installed/live-provider qualification. Unclassified readback exceptions retain the saved attempt and retry with backoff; their wording never establishes revoked authority or a missing route. Adapters diff --git a/loopx/capabilities/manager_context/roundtrip.py b/loopx/capabilities/manager_context/roundtrip.py index bf0e2c56bd..df5a369d3c 100644 --- a/loopx/capabilities/manager_context/roundtrip.py +++ b/loopx/capabilities/manager_context/roundtrip.py @@ -8,9 +8,9 @@ import logging import threading -from contextlib import ExitStack -from datetime import datetime, timezone, timedelta -from uuid import uuid4 +from itertools import islice +from contextlib import contextmanager, ExitStack +from datetime import datetime, timezone from . import _root, _read, _write, _hash, authority, target_authority from .tracking import _entry, _now @@ -29,13 +29,11 @@ load_project_registry, ) from ...presentation.public_safety import scan_public_boundary_text -from ...control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_result +from ...control_plane.effect_runtime import EffectRuntimeRejected, effect_runtime_request_scope, effect_runtime_result from ...control_plane.collaboration.inbox import ( _request_lock, iter_result_paths, - inspect_result_identity, - prior_result_delivered, DELIVERY_STATUSES as DELIVERY_STATUSES, DELIVERY_ERRORS as DELIVERY_ERRORS, reply_status as reply_status, @@ -44,8 +42,6 @@ from ...control_plane.collaboration.source_chat_observation import original_delivery_facts from ...chat import ChatMessagePayloadConflictError -EXACT_DELIVERY_ADMISSION_SECONDS = 60 - class ReturnResolutionBlocked(ValueError, RuntimeError): """A return cannot be resolved; ``reason`` is the provider-neutral code. @@ -63,32 +59,6 @@ def __init__(self, reason, message): super().__init__(message) -def _delivery_attempt(value): - return dict( - effect_runtime_result( - "manager.return_delivery.normalize_attempt", {"attempt": value} - ) - ) - - -def _attempt_locator(value): - """The provider locator a recorded attempt can be verified against. - - ``None`` covers both a record no normalization can read and the typed state - where the provider accepted the write without reporting a message id. The - attempt still proves a write happened; it just cannot name a readback - target, and that is what stops the pump from treating the return as unsent. - """ - - if value is None: - return None - try: - message_ref = _delivery_attempt(value).get("message_ref") - except (ValueError, EffectRuntimeRejected): - return None - return message_ref if isinstance(message_ref, str) and message_ref.strip() else None - - def _verification_decision(outcome): return dict( effect_runtime_result( @@ -97,15 +67,6 @@ def _verification_decision(outcome): ) -def _verification_exception_error(exc): - decision = _verification_decision({ - "verification_performed": False, - "reply_verified": False, - "blocker": exc.reason if isinstance(exc, ReturnResolutionBlocked) else None, - }) - return decision["error"] if decision["status"] == "explicit_unverified" else None - - def _register_unlocked(root, row, session, turn, *, source_store=None): value = { key: row[key] @@ -333,13 +294,7 @@ def project_chat_session_snapshot(root, store, session_id, *, registry): -def _exact_return_scope(registry, reply): - return collaboration_goal_scope( - registry, - goal_id=reply["goal_id"], - agents=(reply["agent_id"],), - caller_goal_ref=reply["goal_ref"], - ) +RETURN_DELIVERY_TRANSACTION = "manager_return_delivery_transaction_v0" def _hold_return_source_fence(root, route, effects): @@ -351,259 +306,140 @@ def _hold_return_source_fence(root, route, effects): ) -def _exact_return_context(root, registry, store, path, state_path, now, effects): - reply = _read(path) - if not isinstance(reply.get("goal_ref"), dict): - raise ValueError("exact return reply is missing goal_ref") - # Observe the frozen route only to select the fence, before Goal/request - # locks. Validate it against the exact request again under those locks. - frozen_route = _read(_root(root) / "roundtrips" / (path.parent.name + ".json")) - _hold_return_source_fence(root, frozen_route, effects) - with _exact_return_scope(registry, reply) as scope: - row = _entry( - root, - reply["goal_id"], - reply["agent_id"], - reply["request_id"], - scope=scope, - ) - inspected = inspect_result_identity(reply, row, path) - if ( - path.parent.name != row["request_id"] - or inspected is None - or any( - reply.get(key) != row.get(key) - for key in ("source_id", "goal_ref") - ) - ): - raise ValueError("return_reply_identity_mismatch") - route = _route(root, row, scope=scope) - if route != frozen_route: - raise ValueError("context return route changed during admission") - if route.get("kind") == "peer": - return None - session = store.load_session(route["session_id"]) - turn = store.turn_for_client(route["session_id"], route["client_turn_id"]) - if ( - not session - or session.get("status") == "closed" - or session.get("channel_id") != route["channel_id"] - or not turn - ): - raise ValueError("original_conversation_unavailable") - target = {key: row[key] for key in ("goal_id", "agent_id")} - grant = target_authority( - root, - session=session, - turn=turn, - target=target, - ) - if ( - target not in grant["targets"] - or grant.get("source_id") != row["source_id"] - ): - raise ValueError("return_authorization_unavailable") - decide_collaboration_lifecycle( - scope, - operation="original_return_admit", - record=row, - route=route, - initial_delivery=original_delivery_facts(row, route, turn, source_id=grant.get("source_id")), - ) - with _request_lock( - root, - row["request_id"], - scope, - path.with_suffix(".lock"), - ): - if not prior_result_delivered(path, reply): - return None - state = _read(state_path) if state_path.exists() else {} - if state.get("status") in { - "delivered", - "superseded", - "explicit_unverified", - }: - return None - admission = state.get("admission") - if ( - isinstance(admission, dict) - and isinstance(admission.get("expires_at"), str) - and now.isoformat() < admission["expires_at"] - ): - return None - if ( - state.get("status") == "admitted" - and state.get("attempt") is None - and not conversation_scope(session)["private_conversation"] - ): - _write( - state_path, - { - "status": "explicit_unverified", - "error": "provider_delivery_unverified", - "goal_ref": row["goal_ref"], - }, - ) - return None - if ( - state.get("retry_at") - and state.get("status") != "admitted" - and now.isoformat() < state["retry_at"] - ): - return None - if inspected["interim_superseded"]: - _write( - state_path, - { - "status": "superseded", - "reason": "conclusion_ready", - "goal_ref": row["goal_ref"], - }, - ) - return None - token = uuid4().hex - prior_status = ( - admission.get("prior_status") - if isinstance(admission, dict) - else state.get("status", "queued") - ) - admitted = { - **state, - "status": "admitted", - "goal_ref": row["goal_ref"], - "admission": { - "token": token, - "prior_status": prior_status, - "admitted_at": now.isoformat(), - "expires_at": ( - now + timedelta(seconds=EXACT_DELIVERY_ADMISSION_SECONDS) - ).isoformat(), - }, +@contextmanager +def _return_source(host, path, frozen_route, original_grant=None): + """Observe original Chat/provider sources under their existing lifetime fence.""" + root, registry, store, exact = (host[key] for key in ("root", "registry", "store", "exact")) + with ExitStack() as guards: + try: + reply = _read(path) + scope = guards.enter_context(collaboration_goal_scope( + registry, goal_id=reply["goal_id"], + agents=(reply["agent_id"],) if exact else (), + caller_goal_ref=reply.get("goal_ref") if exact else None, + )) + if exact: + guards.enter_context(_request_lock(root, reply["request_id"], scope, path.with_suffix(".lock"))) + row = _entry(root, reply["goal_id"], reply["agent_id"], reply["request_id"], scope=scope) + route = _route(root, row, scope=scope) + if route != frozen_route: + raise ValueError("context return route changed during admission") + session = store.load_session(route["session_id"]) + turn = store.turn_for_client(route["session_id"], route["client_turn_id"]) + grant = original_grant + if grant is None and session and turn: + grant = (target_authority(root, session=session, turn=turn, + target={key: row[key] for key in ("goal_id", "agent_id")}) + if exact else authority(root, registry, session, turn)) + target = scope.target(reply["agent_id"]) + facts = { + "entry": row, + "entry_path": str(_root(root) / "entries" / _hash(target) / (reply["request_id"] + ".json")), + "route": route, "session": session, "turn": turn, + "grant": grant or {"targets": []}, + "scope": {"profile_id": scope.profile_id, + "caller_goal_ref": scope.caller_goal_ref, + "current_goal_ref": scope.current_goal_ref}, + "initial_delivery": original_delivery_facts(row, route, turn, + source_id=(grant or {}).get("source_id")) if turn else None, } - admitted.pop("retry_at", None) - _write(state_path, admitted) - return { - "reply": reply, - "row": row, - "route": route, - "session": session, - "turn": turn, - "store": store, - "state_path": state_path, - "state": admitted, - "token": token, - "source_id": grant["source_id"], - } - - -def _write_exact_return_state( - root, - registry, - context, - value, - *, - preserve_admission=False, -): - reply = context["reply"] - with _exact_return_scope(registry, reply) as scope: - row = _entry( - root, - reply["goal_id"], - reply["agent_id"], - reply["request_id"], - scope=scope, - ) - route = _route(root, row, scope=scope) - turn = context["store"].turn_for_client( - route["session_id"], - route["client_turn_id"], - ) - decide_collaboration_lifecycle( - scope, - operation="original_return_settle", - record=row, - route=route, - initial_delivery=( - original_delivery_facts(row, route, turn, source_id=context["source_id"]) - if isinstance(turn, dict) else None - ), - ) - with _request_lock( - root, - row["request_id"], - scope, - context["state_path"].with_suffix(".lock"), - ): - current = _read(context["state_path"]) - admission = current.get("admission") - if ( - not isinstance(admission, dict) - or admission.get("token") != context["token"] - ): - raise ValueError("exact return delivery admission changed") - result = {**value, "goal_ref": row["goal_ref"], - **({"result_key": reply["result_key"]} if "result_key" in reply else {})} - # A refusal does not erase evidence that the provider already wrote. - # Terminal state remains terminal; preserving its locator is no retry. - if result.get("status") == "explicit_unverified" and current.get("attempt") is not None: - result["attempt"] = current["attempt"] - if preserve_admission: - result["admission"] = admission - else: - result.pop("admission", None) - _write(context["state_path"], result) - if (result.get("status") == "delivered" and result.get("reply_verified") is True - and current.get("status") != "delivered"): - from ...usage_ping import observe_verified_return - observe_verified_return() - - -def _retry_state(state, now, *, error): - attempts = int(state.get("attempts", 0)) + 1 - result = { - key: value - for key, value in state.items() - if key not in {"admission", "delivered_at", "message_id", "verification"} - } - result.update( - status="retry_pending", - attempts=attempts, - error=error, - retry_at=( - now + timedelta(seconds=min(300, 5 * 2 ** min(attempts, 6))) - ).isoformat(), - ) + except (OSError, ValueError, KeyError, TypeError, RuntimeError) as exc: + facts = {"error": exc.reason if isinstance(exc, ReturnResolutionBlocked) else "return_source_unavailable"} + yield facts + + +def _return_transaction(host, path, phase, request): + root, now, exact = (host[key] for key in ("root", "now", "exact")) + payload = {key: value for key, value in request.items() if key != "facts"} + observed = dict(request["facts"]) + for field, keys in { + "session": ("session_id", "goal_id", "channel_id", "status", "project_context", "steward_context"), + "turn": ("turn_id", "client_turn_id", "status"), + "grant": ("source_id", "targets"), + }.items(): + if isinstance(observed.get(field), dict): + observed[field] = {key: observed[field][key] for key in keys if key in observed[field]} + result = effect_runtime_result("manager.return_delivery.transact", { + "schema_version": RETURN_DELIVERY_TRANSACTION, + "runtime_root": str(root.resolve()), "request_id": path.parent.name, + "result_key": path.stem, "phase": phase, "now": now.isoformat(), + "exact": exact, "facts": observed, **payload, + }, retry_safe=False) + if (not isinstance(result, dict) or result.get("schema_version") != RETURN_DELIVERY_TRANSACTION + or result.get("action") not in {"skip", "private", "verify", "send"} + or type(result.get("processed")) is not int or result["processed"] not in {0, 1}): + raise RuntimeError("return delivery transaction result shape mismatch") return result -def _transport_route(route, reply, external_sender): - attachments = reply.get("attachments") or [] - if not attachments: - return route - if getattr(external_sender, "supports_result_files", False) is not True: - raise ValueError("return_transport_unavailable") - # Ephemeral transport payload, never part of the persisted audience/grant. - return {**route, "result_attachments": attachments} - - -def _resolve_delivery_sender(external_sender): - """Resolve a manager return transport to ``(sender, attempt_aware)``. +def _return_provider(plan, source, sender, record_attempt): + route, session, turn = (source[key] for key in ("route", "session", "turn")) + attachments = plan.get("attachments") or [] + if attachments: + if getattr(sender, "supports_result_files", False) is not True: + raise ValueError("return_transport_unavailable") + route = {**route, "result_attachments": attachments} + if plan["action"] == "private": + return {"kind": "private"} + if plan["action"] == "verify": + attempt = plan.get("attempt") + if not attempt or attempt.get("message_ref") is None: + return {"kind": "locator_unavailable"} + verifier = getattr(sender, "verify", None) + if not callable(verifier): + return {"kind": "verify_unavailable"} + return {"kind": "verified", "verification": verifier(route, session, turn, plan["text"], attempt)} + provider = getattr(sender, "send_with_attempt", None) + if callable(provider): + sent = provider(route, session, turn, plan["text"], record_attempt) + elif callable(sender): + sent = sender(route, session, turn, plan["text"]) + else: + return {"kind": "sender_unavailable"} + return {"kind": "sent", "reply_verified": sent.get("reply_verified"), + "external_write_performed": sent.get("external_write_performed"), + "idempotency_key": sent.get("idempotency_key")} + + +def _deliver_return(host, path, sender, frozen_route): + store, cancelled = host["store"], host["cancelled"] + with _return_source(host, path, frozen_route) as facts: + plan = _return_transaction(host, path, "prepare", {"facts": facts}) + source = dict(facts) + if plan.get("source_unavailable") is True: + logging.getLogger(__name__).warning("Exact manager return admission unavailable") + if plan["action"] == "skip": + return plan["processed"] + if cancelled(): + return 0 + checkpoint = {key: plan[key] for key in ("revision", "token")} + + def transact(phase, **payload): + with _return_source(host, path, frozen_route, source["grant"]) as current: + return _return_transaction(host, path, phase, + {"facts": current, **checkpoint, **payload}) + + def record_attempt(attempt): + saved = transact("attempt", attempt=dict(attempt)) + checkpoint.update({key: saved[key] for key in ("revision", "token")}) - Two transport shapes are legitimate: an object exposing a callable - ``send_with_attempt`` (the attempt-aware protocol) and a bare callable - transport. The predicate is shape-based on purpose — the transports - share no base class, so an ``isinstance`` check would reject valid - transports. ``(None, False)`` means the transport cannot deliver: - callers must record ``return_transport_unavailable`` before invoking - instead of letting the call raise ``TypeError`` mid-loop. - """ - sender = getattr(external_sender, "send_with_attempt", None) - if callable(sender): - return sender, True - if callable(external_sender): - return external_sender, False - return None, False + try: + store.append_message(source["route"]["session_id"], role="agent", text=plan["text"], + turn_id=source["turn"]["turn_id"], origin="manager_followup", message_id=plan["message_id"]) + if cancelled(): + return 0 + outcome = _return_provider(plan, source, sender, record_attempt) + except ChatMessagePayloadConflictError as exc: + logging.getLogger(__name__).warning("Manager return payload conflict on %s: %s", plan["message_id"], exc.field) + outcome = {"kind": "conflict"} + except (OSError, ValueError, KeyError, TypeError, RuntimeError) as exc: + outcome = {"kind": "verification_error" if plan["action"] == "verify" else "error", + "blocker": exc.reason if isinstance(exc, ReturnResolutionBlocked) else None} + settled = transact("settle", outcome=outcome, message_id=plan["message_id"]) + if settled.get("observe_verified"): + from ...usage_ping import observe_verified_return + observe_verified_return() + return settled["processed"] def _belongs_to_chat_host(root, store, path): @@ -611,9 +447,6 @@ def _belongs_to_chat_host(root, store, path): route_path = _root(root) / "roundtrips" / (path.parent.name + ".json") try: - # Missing legacy routes can only be recovered from the coordination - # root's Chat receipts. Private hosts recover their frozen provenance - # at startup before draining; a foreign host must never claim them. route = _read(route_path) if route_path.exists() else {} return route.get("kind") != "peer" and is_source_chat_host(root, route, store) except (OSError, ValueError, TypeError): @@ -621,699 +454,56 @@ def _belongs_to_chat_host(root, store, path): return False -def _drain_exact(root, registry, store, external_sender, *, now, cancelled): - processed = 0 - for path in iter_result_paths(_root(root) / "replies"): - if cancelled(): - break - if not _belongs_to_chat_host(root, store, path): - continue - state_path = path.with_name(path.stem + ".delivery.json") - effect_locks = ExitStack() - try: - effect_locks.enter_context( - exclusive_file_lock( - _root(root) - / "return-effect-locks" - / path.parent.name - / path.stem, - policy=LockAcquisitionPolicy.SINGLE_FLIGHT, - operation="manager_return_delivery", - ) - ) - except LockAcquireTimeoutError: - continue - try: - context = _exact_return_context( - root, - registry, - store, - path, - state_path, - now, - effect_locks, - ) - except (OSError, ValueError, KeyError, TypeError, RuntimeError): - effect_locks.close() - logging.getLogger(__name__).warning( - "Exact manager return admission unavailable" - ) - continue - if context is None: - effect_locks.close() - continue - row = context["row"] - route = context["route"] - session = context["session"] - turn = context["turn"] - reply = context["reply"] - state = context["state"] - prefix = "处理结论" if reply["phase"] == "conclusion" else "处理进展" - text = ( - f"{prefix} · {row['agent_id']} · 委托 {row['request_id'][:8]}" - f"\n\n{reply['text']}" - ) - mid = "handoff." + _hash([row["request_id"], path.stem]) - try: - if cancelled(): - return processed - route = _transport_route(route, reply, external_sender) - try: - store.append_message( - route["session_id"], - role="agent", - text=text, - turn_id=turn["turn_id"], - origin="manager_followup", - message_id=mid, - ) - except ChatMessagePayloadConflictError as exc: - logging.getLogger(__name__).warning( - "Manager return payload conflict on %s: %s", - mid, - exc.field, - ) - _write_exact_return_state( - root, - registry, - context, - { - "status": "explicit_unverified", - "error": "manager_return_payload_conflict", - }, - ) - processed += 1 - continue - if cancelled(): - return processed - if conversation_scope(session)["private_conversation"]: - _write_exact_return_state( - root, - registry, - context, - { - "status": "delivered", - "delivered_at": now.isoformat(), - "message_id": mid, - }, - ) - processed += 1 - continue - - prior_status = state["admission"]["prior_status"] - if prior_status == "verification_required" or state.get("attempt") is not None: - attempt = state.get("attempt") - if attempt is None or _attempt_locator(attempt) is None: - _write_exact_return_state( - root, - registry, - context, - { - **({"attempt": attempt} if attempt is not None else {}), - "status": "explicit_unverified", - "error": "provider_locator_unavailable", - }, - ) - processed += 1 - continue - normalized_attempt = _delivery_attempt(attempt) - verifier = getattr(external_sender, "verify", None) - if not callable(verifier): - _write_exact_return_state( - root, - registry, - context, - { - "status": "explicit_unverified", - "error": "provider_verifier_unavailable", - }, - ) - processed += 1 - continue - try: - decision = _verification_decision( - verifier( - route, - session, - turn, - text, - normalized_attempt, - ) - ) - except ( - OSError, - ValueError, - KeyError, - TypeError, - RuntimeError, - ) as exc: - error = _verification_exception_error(exc) - next_state = ( - { - "status": "explicit_unverified", - "error": error, - } - if error - else _retry_state( - {"attempt": normalized_attempt, **state}, - now, - error="provider_verification_unavailable", - ) - ) - _write_exact_return_state( - root, - registry, - context, - next_state, - ) - processed += 1 - continue - if decision["status"] == "delivered": - next_state = { - "status": "delivered", - "delivered_at": now.isoformat(), - "message_id": mid, - "provider_receipt": normalized_attempt["provider_receipt"], - "reply_verified": True, - "verification": decision["verification"], - } - elif decision["status"] == "explicit_unverified": - next_state = { - "status": "explicit_unverified", - "error": decision["error"], - } - else: - next_state = _retry_state( - {"attempt": normalized_attempt, **state}, - now, - error=decision["error"], - ) - _write_exact_return_state( - root, - registry, - context, - next_state, - ) - processed += 1 - continue - - def record_attempt(value): - nonlocal state - attempt = _delivery_attempt(value) - existing = state.get("attempt") - if existing is not None and _delivery_attempt(existing) != attempt: - raise ValueError("manager return delivery attempt conflict") - state = {**state, "attempt": attempt} - _write_exact_return_state( - root, - registry, - context, - state, - preserve_admission=True, - ) - - sender, attempt_aware = _resolve_delivery_sender(external_sender) - if sender is None: - _write_exact_return_state( - root, - registry, - context, - _retry_state( - state, - now, - error="return_transport_unavailable", - ), - ) - processed += 1 - continue - sent = ( - sender(route, session, turn, text, record_attempt) - if attempt_aware - else sender(route, session, turn, text) - ) - if sent.get("reply_verified") is not True: - if sent.get("external_write_performed") is True: - attempt = state.get("attempt") - next_state = ( - { - "status": "verification_required", - "error": "provider_delivery_unverified", - "attempt": attempt, - } - if attempt is not None and _attempt_locator(attempt) is not None - else { - **({"attempt": attempt} if attempt is not None else {}), - "status": "explicit_unverified", - "error": "provider_locator_unavailable", - } - ) - _write_exact_return_state( - root, - registry, - context, - next_state, - ) - processed += 1 - continue - raise ValueError("return_transport_unavailable") - _write_exact_return_state( - root, - registry, - context, - { - "status": "delivered", - "delivered_at": now.isoformat(), - "message_id": mid, - "provider_receipt": sent.get("idempotency_key"), - "reply_verified": True, - }, - ) - except (OSError, ValueError, KeyError, TypeError, RuntimeError) as exc: - attempt = state.get("attempt") - error = ( - _verification_exception_error(exc) - if attempt is not None - else None - ) - if error: - next_state = { - "status": "explicit_unverified", - "error": error, - } - elif attempt is not None and _attempt_locator(attempt) is None: - next_state = { - "status": "explicit_unverified", - "error": "provider_locator_unavailable", - "attempt": attempt, - } - else: - next_state = _retry_state( - state, - now, - error="original_route_or_return_delivery_unavailable", - ) - try: - _write_exact_return_state( - root, - registry, - context, - next_state, - ) - except (OSError, ValueError, KeyError, TypeError, RuntimeError): - logging.getLogger(__name__).warning( - "Exact manager return settlement unavailable" - ) - finally: - effect_locks.close() - processed += 1 - if processed >= 20: - break - return processed +def _pending_return_paths(host, paths): + root, now, exact, cancelled = (host[key] for key in ("root", "now", "exact", "cancelled")) + paths = iter(paths) + while not cancelled() and (batch := list(islice(paths, 128))): + candidates = [{"request_id": path.parent.name, "result_key": path.stem} for path in batch] + selected = effect_runtime_result("manager.return_delivery.transact", { + "schema_version": RETURN_DELIVERY_TRANSACTION, "phase": "scan", + "runtime_root": str(root.resolve()), "now": now.isoformat(), "exact": exact, + "candidates": candidates, + }) + if not isinstance(selected, dict) or selected.get("schema_version") != RETURN_DELIVERY_TRANSACTION: + raise RuntimeError("return delivery scan result shape mismatch") + if selected.get("unavailable_count"): + logging.getLogger(__name__).warning("Unreadable manager return receipt") + ready = {(row["request_id"], row["result_key"]) for row in selected["candidates"]} + yield from (path for path in batch if (path.parent.name, path.stem) in ready) +@effect_runtime_request_scope() def drain(root, registry, store, external_sender, *, now=None, cancelled=lambda: False): - """Restart-safe return delivery; transport retries never rerun the worker/model.""" + """Run the typed return transaction with the existing Chat/provider adapters.""" + if cancelled(): + return 0 now = now or datetime.now(timezone.utc) try: - profile_id = load_project_registry(registry).get("profile_id") + exact = load_project_registry(registry).get("profile_id") == SOURCE_SESSION_PROFILE_ID except (OSError, ValueError, TypeError): - profile_id = None - if profile_id == SOURCE_SESSION_PROFILE_ID: - return _drain_exact( - root, - registry, - store, - external_sender, - now=now, - cancelled=cancelled, - ) + exact = False processed = 0 - for path in iter_result_paths(_root(root) / "replies"): + host = {"root": root, "registry": registry, "store": store, "now": now, "exact": exact, "cancelled": cancelled} + paths = iter_result_paths(_root(root) / "replies") + for path in _pending_return_paths(host, paths): if cancelled(): break if not _belongs_to_chat_host(root, store, path): continue - state_path = path.with_name(path.stem + ".delivery.json") - with exclusive_file_lock(path.with_suffix(".lock")), ExitStack() as effects: - try: - state = _read(state_path) if state_path.exists() else {} - except (OSError, ValueError): - # A damaged receipt has unknown delivery state: do not resend - # it blindly, and do not starve unrelated pending returns. - logging.getLogger(__name__).warning("Unreadable manager return receipt") - continue - if state.get("status") in {"delivered", "superseded", "explicit_unverified"}: - continue - if state.get("retry_at") and now.isoformat() < state["retry_at"]: - continue - try: - reply = _read(path) - row = _entry( - root, reply["goal_id"], reply["agent_id"], reply["request_id"] - ) - inspected = inspect_result_identity(reply, row, path) - if ( - path.parent.name != row["request_id"] - or inspected is None - or reply.get("source_id") != row["source_id"] - ): - raise ValueError("return_reply_identity_mismatch") - route = _transport_route(_route(root, row), reply, external_sender) - if route.get("kind") == "peer": - continue # Delivered by the requester inbox, never a Chat audience. - _hold_return_source_fence(root, route, effects) - session = store.load_session(route["session_id"]) - turn = store.turn_for_client( - route["session_id"], route["client_turn_id"] - ) - if ( - not session - or session.get("status") == "closed" - or session.get("channel_id") != route["channel_id"] - or not turn - ): - raise ReturnResolutionBlocked("original_route_unavailable", "original_conversation_unavailable") - grant = authority(root, registry, session, turn) - target = {k: row[k] for k in ("goal_id", "agent_id")} - if ( - target not in grant["targets"] - or grant.get("source_id") != row["source_id"] - ): - raise ReturnResolutionBlocked("return_authorization_unavailable", "return_authorization_unavailable") - with collaboration_goal_scope(registry, goal_id=row["goal_id"], agents=()) as scope: - decide_collaboration_lifecycle( - scope, operation="original_return_admit", record=row, route=route, - initial_delivery=original_delivery_facts(row, route, turn, source_id=grant.get("source_id")), - ) - if inspected["interim_superseded"]: - _write( - state_path, - {"status": "superseded", "reason": "conclusion_ready"}, - ) - continue - if not prior_result_delivered(path, reply): - continue - # A conclusion can be a deferral or rejection. Transport completion - # is not completion of the delegated work. - prefix = "协作回复" if reply["phase"] == "conclusion" else "协作进展" - text = f"{prefix} · {row['agent_id']} · 委托 {row['request_id'][:8]}\n\n{reply['text']}" - # Transcript writes are independently idempotent, including when - # Lark is offline. Keep the original Turn and logical conversation. - mid = "handoff." + _hash([row["request_id"], path.stem]) - if cancelled(): - return processed - try: - store.append_message( - route["session_id"], - role="agent", - text=text, - turn_id=turn["turn_id"], - origin="manager_followup", - message_id=mid, - ) - except ChatMessagePayloadConflictError as exc: - logging.getLogger(__name__).warning( - "Manager return payload conflict on %s: %s", - mid, - exc.field, - ) - _write( - state_path, - { - **state, - "status": "explicit_unverified", - "error": "manager_return_payload_conflict", - }, - ) - processed += 1 - continue - if cancelled(): - return processed - transport = {} - if not conversation_scope(session)["private_conversation"]: - if state.get("status") == "verification_required": - if state.get("attempt") is None: - # A record written before locators were persisted has - # no trustworthy provider identity: never guess and - # never resend the conclusion. - _write( - state_path, - { - **state, - "status": "explicit_unverified", - "error": "provider_locator_unavailable", - }, - ) - continue - try: - attempt = _delivery_attempt(state.get("attempt")) - except (ValueError, EffectRuntimeRejected): - _write( - state_path, - { - **state, - "status": "explicit_unverified", - "error": "provider_locator_unavailable", - }, - ) - continue - if _attempt_locator(state.get("attempt")) is None: - # The attempt records a provider write that carried no - # locator, so no readback can prove it and the provider - # must not be called again. Converging here keeps the - # attempt as the evidence of that write while making - # the return terminal, instead of looping the pump - # through verification attempts forever. - _write( - state_path, - { - **state, - "status": "explicit_unverified", - "error": "provider_locator_unavailable", - }, - ) - continue - verifier = getattr(external_sender, "verify", None) - if not callable(verifier): - _write( - state_path, - { - **state, - "status": "explicit_unverified", - "error": "provider_verifier_unavailable", - }, - ) - continue - try: - verified = verifier(route, session, turn, text, attempt) - decision = _verification_decision(verified) - except ( - OSError, - ValueError, - KeyError, - TypeError, - RuntimeError, - ) as exc: - error = _verification_exception_error(exc) - if error: - _write( - state_path, - {**state, "status": "explicit_unverified", "error": error}, - ) - continue - # An unclassified readback failure keeps the locator and - # stays retryable, but must never re-run the provider on - # every pump without backoff. - attempts = int(state.get("attempts", 0)) + 1 - _write( - state_path, - { - **state, - "status": "verification_required", - "attempts": attempts, - "retry_at": ( - now - + timedelta( - seconds=min(300, 5 * 2 ** min(attempts, 6)) - ) - ).isoformat(), - }, - ) - continue - if decision["status"] == "delivered": - _write( - state_path, - { - "status": "delivered", - **({"result_key": path.stem} if "result_key" in reply else {}), - "delivered_at": now.isoformat(), - "message_id": mid, - "provider_receipt": attempt["provider_receipt"], - "reply_verified": True, - "verification": decision["verification"], - }, - ) - continue - if decision["status"] == "explicit_unverified": - _write( - state_path, - { - **state, - "status": "explicit_unverified", - "error": decision["error"], - }, - ) - continue - attempts = int(state.get("attempts", 0)) + 1 - _write( - state_path, - { - **state, - "status": "verification_required", - "attempts": attempts, - "error": decision["error"], - "retry_at": ( - now - + timedelta( - seconds=min(300, 5 * 2 ** min(attempts, 6)) - ) - ).isoformat(), - }, - ) - continue - - def record_attempt(value): - attempt = _delivery_attempt(value) - current = _read(state_path) if state_path.exists() else {} - existing = current.get("attempt") - if existing is not None and _delivery_attempt(existing) != attempt: - raise ValueError("manager return delivery attempt conflict") - _write( - state_path, - { - "status": "verification_required", - "error": "provider_delivery_unverified", - "attempt": attempt, - }, - ) - - sender, attempt_aware = _resolve_delivery_sender(external_sender) - if sender is None: - attempts = int(state.get("attempts", 0)) + 1 - _write( - state_path, - { - "status": "retry_pending", - "attempts": attempts, - "error": "return_transport_unavailable", - "retry_at": ( - now - + timedelta( - seconds=min(300, 5 * 2 ** min(attempts, 6)) - ) - ).isoformat(), - }, - ) - processed += 1 - continue - sent = ( - sender(route, session, turn, text, record_attempt) - if attempt_aware - else sender(route, session, turn, text) - ) - if sent.get("reply_verified") is not True: - if sent.get("external_write_performed") is True: - current = _read(state_path) if state_path.exists() else {} - if ( - current.get("attempt") is not None - and _attempt_locator(current.get("attempt")) is not None - ): - _write( - state_path, - { - **current, - "status": "verification_required", - "error": "provider_delivery_unverified", - }, - ) - else: - # Either nothing was recorded or the record says - # the provider took the write without a locator. - # Both leave no readback target, so the return is - # terminal rather than retryable: a retry would - # post the same text again. - _write( - state_path, - ( - { - **current, - "status": "explicit_unverified", - "error": "provider_locator_unavailable", - } - if current.get("attempt") is not None - else { - "status": "explicit_unverified", - "error": "provider_locator_unavailable", - } - ), - ) - continue - raise ValueError("return_transport_unavailable") - transport = { - "provider_receipt": sent.get("idempotency_key"), - "reply_verified": True, - } - _write( - state_path, - { - "status": "delivered", - **({"result_key": path.stem} if "result_key" in reply else {}), - "delivered_at": now.isoformat(), - "message_id": mid, - **transport, - }, - ) - except (OSError, ValueError, KeyError, TypeError, RuntimeError) as exc: - current = _read(state_path) if state_path.exists() else {} - if current.get("status") == "verification_required" and current.get( - "attempt" - ) is not None: - error = _verification_exception_error(exc) - if error: - _write( - state_path, - {**current, "status": "explicit_unverified", "error": error}, - ) - elif _attempt_locator(current.get("attempt")) is None: - # The record says the provider took the write and named - # no locator, so there is nothing left to verify and a - # retry would post the same text again. Converge now - # instead of leaving the return in a retryable state. - _write( - state_path, - { - **current, - "status": "explicit_unverified", - "error": "provider_locator_unavailable", - }, - ) - processed += 1 - continue - attempts = int(state.get("attempts", 0)) + 1 - _write( - state_path, - { - "status": "retry_pending", - "attempts": attempts, - "error": "original_route_or_return_delivery_unavailable", - "retry_at": ( - now + timedelta(seconds=min(300, 5 * 2 ** min(attempts, 6))) - ).isoformat(), - }, - ) - processed += 1 - if processed >= 20: - break + lock_path = (_root(root) / "return-effect-locks" / path.parent.name / path.stem + if exact else path.with_suffix(".lock")) + try: + with exclusive_file_lock(lock_path, **({"policy": LockAcquisitionPolicy.SINGLE_FLIGHT, + "operation": "manager_return_delivery"} if exact else {})), ExitStack() as effects: + frozen_route = _read(_root(root) / "roundtrips" / (path.parent.name + ".json")) + _hold_return_source_fence(root, frozen_route, effects) + processed += _deliver_return(host, path, external_sender, frozen_route) + except LockAcquireTimeoutError: + continue + except (OSError, ValueError, KeyError, TypeError, RuntimeError): + logging.getLogger(__name__).warning("Manager return transaction unavailable") + if processed >= 20: + break return processed diff --git a/loopx/control_plane/collaboration/return_delivery.ts b/loopx/control_plane/collaboration/return_delivery.ts index 0859c08d6d..1ee6831953 100644 --- a/loopx/control_plane/collaboration/return_delivery.ts +++ b/loopx/control_plane/collaboration/return_delivery.ts @@ -4,8 +4,19 @@ import { requireBoolean, requireJsonObject, requireNonEmptyString, + requireStringLiteral, } from "../runtime_decode.ts"; -import { ENVELOPED_SHA256_PATTERN } from "../content_digest.ts"; +import { BARE_SHA256_PATTERN, ENVELOPED_SHA256_PATTERN } from "../content_digest.ts"; +import { createHash, randomUUID } from "node:crypto"; +import { constants } from "node:fs"; +import { open, readdir } from "node:fs/promises"; +import { isAbsolute, join, relative, resolve, toNamespacedPath } from "node:path"; +import { isDeepStrictEqual } from "node:util"; +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"; export const MANAGER_RETURN_DELIVERY_ATTEMPT_SCHEMA = "manager_return_delivery_attempt_v0"; @@ -120,3 +131,388 @@ export function classifyManagerReturnVerification(value: unknown): JsonObject { : "provider_delivery_mismatch"; return { status: "explicit_unverified", error, verification: null }; } + +export const RETURN_DELIVERY_TRANSACTION = "manager_return_delivery_transaction_v0"; +const RESULT_KEY = /^(decision|conclusion|conclusion-[0-9]{8}|decision-[0-9]{8})$/; +const TERMINAL = new Set(["delivered", "superseded", "explicit_unverified"]); + +interface ReturnCommand { + request: JsonObject; + folder: string; + key: string; + path: string; + now: string; + exact: boolean; + facts: JsonObject; +} + +function returnCommand(value: unknown): ReturnCommand { + const request = requireJsonObject(value, "return delivery transaction"); + if (request.schema_version !== RETURN_DELIVERY_TRANSACTION) { + throw new EffectRuntimeRequestError("return delivery transaction schema mismatch"); + } + const root = requireNonEmptyString(request.runtime_root, "runtime_root"); + if (!isAbsolute(root)) throw new EffectRuntimeRequestError("return runtime root must be absolute"); + const id = matchingString(request.request_id, "request_id", BARE_SHA256_PATTERN); + const key = matchingString(request.result_key, "result_key", RESULT_KEY); + const now = requireNonEmptyString(request.now, "now"); + if (now.length > 64 || parseTodoTimestampMicros(now) === null) throw new EffectRuntimeRequestError("invalid return clock"); + requireStringLiteral(request.phase, ["prepare", "attempt", "settle"], "return delivery phase"); + const folder = join(root, ".local", "manager-context"); + return { request, folder, key, now, + path: join(folder, "replies", id, `${key}.delivery.json`), + exact: requireBoolean(request.exact, "exact"), + facts: requireJsonObject(request.facts, "return source facts") }; +} + +async function readReturnRecord(path: string, absent: JsonObject | null = null): Promise { + try { + const file = await open(path, constants.O_RDONLY | (constants.O_NONBLOCK ?? 0)); + try { + const metadata = await file.stat(); + if (!metadata.isFile() || metadata.size > 128000) { + throw new EffectRuntimeRequestError("manager context record too large or not regular"); + } + const bytes = Buffer.alloc(Math.min(metadata.size + 1, 128001)); + let length = 0; + while (length < bytes.length) { + const read = await file.read(bytes, length, bytes.length - length, null); + if (!read.bytesRead) break; + length += read.bytesRead; + } + if (length > 128000) throw new EffectRuntimeRequestError("manager context record too large"); + const text = new TextDecoder("utf-8", { fatal: true, ignoreBOM: true }).decode(bytes.subarray(0, length)); + return requireJsonObject(JSON.parse(text), "manager context record"); + } finally { await file.close(); } + } catch (error) { + if ((error as NodeJS.ErrnoException).code === "ENOENT") return absent; + throw error; + } +} + +function returnResponse(fields: JsonObject = {}): JsonObject { + return { schema_version: RETURN_DELIVERY_TRANSACTION, action: "skip", processed: 0, ...fields }; +} + +function unavailableRecord(error: unknown): boolean { + return error instanceof SyntaxError || error instanceof EffectRuntimeRequestError + || typeof (error as NodeJS.ErrnoException)?.code === "string"; +} + +function pendingDelivery(command: ReturnCommand, state: JsonObject): boolean { + if (TERMINAL.has(String(state.status))) return false; + if (!command.exact || state.status !== "admitted") { + if (state.retry_at && command.now < String(state.retry_at)) return false; + } + const admission = state.admission; + return !(command.exact && admission && typeof admission === "object" + && (admission as JsonObject).expires_at + && command.now < String((admission as JsonObject).expires_at)); +} + +function deliveryState(value: unknown): JsonObject { + const state = requireJsonObject(value, "delivery state"); + if (Object.keys(state).length) requireStringLiteral(state.status, + [...TERMINAL, "queued", "retry_pending", "verification_required", "admitted"], "delivery status"); + return state; +} + +async function scanReturns(request: JsonObject): Promise { + if (request.schema_version !== RETURN_DELIVERY_TRANSACTION) { + throw new EffectRuntimeRequestError("return delivery transaction schema mismatch"); + } + if (!Array.isArray(request.candidates) || request.candidates.length > 128) { + throw new EffectRuntimeRequestError("return scan requires at most 128 candidates"); + } + const commands = request.candidates.map(candidate => returnCommand({ ...request, + ...requireJsonObject(candidate, "return candidate"), phase: "prepare", facts: {} })); + const candidates = []; + let unavailable = 0; + for (const command of commands) { + try { + const state = deliveryState(await readReturnRecord(command.path, {})); + if (pendingDelivery(command, state)) candidates.push({ request_id: command.request.request_id, result_key: command.key }); + } catch (error) { if (!unavailableRecord(error)) throw error; unavailable += 1; } + } + return returnResponse({ candidates, unavailable_count: unavailable }); +} + +function shiftedClock(now: string, seconds: number): string { + // Preserve Python's offset/fraction representation, used by existing lexical retry fences. + const match = /^(\d{4}-\d\d-\d\dT\d\d:\d\d:\d\d)(\.\d+)?(Z|[+-]\d\d:\d\d(?::\d\d(?:\.\d+)?)?)?$/.exec(now); + if (!match) throw new EffectRuntimeRequestError("invalid return clock representation"); + return new Date(Date.parse(`${match[1]}Z`) + seconds * 1000).toISOString().slice(0, 19) + + (match[2] ?? "") + (match[3] ?? ""); +} + +function retryState(state: JsonObject, command: ReturnCommand, error: unknown): JsonObject { + const attempts = Number(state.attempts ?? 0) + 1; + if (!Number.isSafeInteger(attempts) || attempts < 1) { + throw new EffectRuntimeRequestError("invalid return retry count"); + } + const retained = Object.fromEntries(Object.entries(state).filter(([key]) => + !["admission", "delivered_at", "message_id", "verification"].includes(key))); + return { ...retained, status: "retry_pending", attempts, error, + retry_at: shiftedClock(command.now, Math.min(300, 5 * 2 ** Math.min(attempts, 6))) }; +} + +function deliveryRevision(state: JsonObject): string { + return createHash("sha256").update(JSON.stringify(state)).digest("hex"); +} + +async function writeDeliveryState(command: ReturnCommand, state: JsonObject): Promise { + await atomicWriteJson(command.path, state); +} + +function admitReturnLifetime(command: ReturnCommand, operation: string): void { + const facts = command.facts; + const decision = decideCollaborationLifecycle({ + ...requireJsonObject(facts.scope, "return Goal scope"), operation, + record_goal_ref: requireJsonObject(facts.entry, "request").goal_ref ?? null, + route_goal_ref: requireJsonObject(facts.route, "return route").goal_ref ?? null, + initial_delivery: facts.initial_delivery, + }); + if (decision.kind === "reject" || decision.kind === "omit") { + throw new EffectRuntimeRequestError("return Goal lifetime admission rejected"); + } +} + +async function inspectReturnPublication(command: ReturnCommand): Promise<{ reply: JsonObject; inspected: JsonObject }> { + const directory = join(command.folder, "replies", String(command.request.request_id)); + const reply = requireJsonObject(await readReturnRecord(join(directory, `${command.key}.json`)), "return reply"); + const entry = requireJsonObject(command.facts.entry, "request"); + if (entry.request_id !== command.request.request_id) { + throw new EffectRuntimeRequestError("return request identity mismatch"); + } + const entryPath = requireNonEmptyString(command.facts.entry_path, "entry_path"); + if (!isAbsolute(entryPath)) throw new EffectRuntimeRequestError("return entry path must be absolute"); + const address = relative(toNamespacedPath(join(command.folder, "entries")), toNamespacedPath(resolve(entryPath))).split(/[\\/]/); + if (address.length !== 2 || !BARE_SHA256_PATTERN.test(address[0]) + || address[1] !== `${command.request.request_id}.json` + || !isDeepStrictEqual(await readReturnRecord(entryPath), entry)) { + throw new EffectRuntimeRequestError("return request source changed"); + } + const route = requireJsonObject(command.facts.route, "return route"); + if (!isDeepStrictEqual(await readReturnRecord(join(command.folder, "roundtrips", `${entry.request_id}.json`)), route) + || ["request_id", "goal_id", "agent_id", "source_id", "goal_ref"].some(key => + entry[key] !== undefined && !isDeepStrictEqual(entry[key], route[key]))) { + throw new EffectRuntimeRequestError("return route identity changed"); + } + const keys = reply.update_id !== undefined || reply.phase === "decision" + ? (await readdir(directory)).filter(name => name.endsWith(".json") && RESULT_KEY.test(name.slice(0, -5))) + : [`${command.key}.json`]; + const results = []; + for (const name of keys.sort()) { + const observed = requireJsonObject(await readReturnRecord(join(directory, name)), "return result"); + const text = requireNonEmptyString(observed.text, "return result text"); + if (Buffer.from(text, "utf8").toString("utf8") !== text) { + throw new EffectRuntimeRequestError("return result text is not valid UTF-8"); + } + results.push({ key: name.slice(0, -5), value: observed, + text_chars: Array.from(text).length, + text_sha256: createHash("sha256").update(text).digest("hex") }); + } + let legacy: JsonObject | null = null; + try { legacy = await readReturnRecord(join(command.folder, "decisions", `${entry.request_id}.json`)); } + 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, + 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"); + } + return { reply, inspected }; +} + +function savedAttempt(attempt: unknown): JsonObject | null { + if (attempt == null) return null; + try { return normalizeManagerReturnDeliveryAttempt(attempt); } + catch (error) { if (error instanceof EffectRuntimeRequestError) return null; throw error; } +} + +function attemptLocator(attempt: unknown): string | null { + return savedAttempt(attempt)?.message_ref as string | null ?? null; +} + +async function sourceFailure(command: ReturnCommand, state: JsonObject, blocker: unknown = command.facts.error): Promise { + if (command.exact) return returnResponse({ source_unavailable: true }); + if (state.status === "verification_required" && state.attempt != null) { + const decision = classifyManagerReturnVerification({ verification_performed: false, + reply_verified: false, blocker }); + if (decision.status === "explicit_unverified" || attemptLocator(state.attempt) === null) { + await writeDeliveryState(command, { ...state, status: "explicit_unverified", + error: decision.status === "explicit_unverified" ? decision.error : "provider_locator_unavailable" }); + } + } else await writeDeliveryState(command, retryState({ attempts: state.attempts ?? 0 }, command, "original_route_or_return_delivery_unavailable")); + return returnResponse({ processed: 1 }); +} + +function returnAuthorityBlocker(command: ReturnCommand): string | null { + const session = requireJsonObject(command.facts.session, "source session"); + requireJsonObject(command.facts.turn, "source turn"); + const route = requireJsonObject(command.facts.route, "source route"); + const target = requireJsonObject(command.facts.entry, "request"); + const allowed = requireJsonObject(command.facts.grant, "return grant"); + if (session.status === "closed" || session.channel_id !== route.channel_id) return "original_route_unavailable"; + if (allowed.source_id !== target.source_id || !Array.isArray(allowed.targets) + || !allowed.targets.some(raw => isDeepStrictEqual(raw, { goal_id: target.goal_id, agent_id: target.agent_id }))) { + return "return_authorization_unavailable"; + } + return null; +} + +async function prepareReturn(command: ReturnCommand, state: JsonObject): Promise { + if (!pendingDelivery(command, state)) return returnResponse(); + if (command.facts.error) return sourceFailure(command, state); + let source: Awaited>; + try { + source = await inspectReturnPublication(command); + const blocker = returnAuthorityBlocker(command); + if (blocker) return sourceFailure(command, state, blocker); + if (requireJsonObject(command.facts.route, "source route").kind === "peer") return returnResponse(); + admitReturnLifetime(command, "original_return_admit"); + } catch (error) { + if (error instanceof EffectRuntimeRequestError || error instanceof SyntaxError + || (error as NodeJS.ErrnoException).code) return sourceFailure(command, state); + throw error; + } + const { reply, inspected } = source; + // Decode the persisted predecessor before using it as a filesystem address. + collaborationResultDeliveryReady({ result: reply, previous_delivery: null }); + const previous = reply.previous_result_key == null ? null + : await readReturnRecord(join(command.folder, "replies", String(reply.request_id), `${reply.previous_result_key}.delivery.json`)); + if (!collaborationResultDeliveryReady({ result: reply, previous_delivery: previous }).ready) return returnResponse(); + const privateConversation = resolveConversationScope(requireJsonObject(command.facts.session, "session")).private_conversation; + const admission = state.admission == null ? null : requireJsonObject(state.admission, "return admission"); + if (command.exact) { + if (state.status === "admitted" && state.attempt == null && !privateConversation) { + await writeDeliveryState(command, { status: "explicit_unverified", error: "provider_delivery_unverified", goal_ref: reply.goal_ref }); + return returnResponse(); + } + } + if (inspected.interim_superseded) { + await writeDeliveryState(command, { status: "superseded", reason: "conclusion_ready", + ...(command.exact ? { goal_ref: reply.goal_ref } : {}) }); + return returnResponse(); + } + const priorStatus = (command.exact ? admission?.prior_status : null) ?? state.status ?? "queued"; + if (command.exact) { + state = { ...state, status: "admitted", goal_ref: reply.goal_ref, + admission: { token: randomUUID().replaceAll("-", ""), prior_status: priorStatus, admitted_at: command.now, + expires_at: shiftedClock(command.now, 60) } }; + delete state.retry_at; + await writeDeliveryState(command, state); + } + 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}`; + + return returnResponse({ action: privateConversation ? "private" : verify ? "verify" : "send", + text, message_id: returnMessageId(command), attempt: verify ? savedAttempt(state.attempt) : null, attachments: reply.attachments ?? [], + revision: deliveryRevision(state), token: (state.admission as JsonObject | undefined)?.token ?? null }); +} + +function returnMessageId(command: ReturnCommand): string { + // Retain Python json.dumps' spaces in the existing ASCII-only message identity. + const identity = JSON.stringify([command.request.request_id, command.key]).replace(",", ", "); + return "handoff." + createHash("sha256").update(identity).digest("hex"); +} + +function deliveryOutcome(command: ReturnCommand, state: JsonObject): JsonObject { + const outcome = requireJsonObject(command.request.outcome, "return provider outcome"); + requireStringLiteral(outcome.kind, ["private", "sent", "conflict", "verify_unavailable", "locator_unavailable", + "verified", "error", "verification_error", "sender_unavailable"], "return outcome kind"); + if (outcome.kind === "private" || outcome.kind === "sent" && outcome.reply_verified === true) { + return { status: "delivered", delivered_at: command.now, message_id: returnMessageId(command), + ...(outcome.kind === "sent" ? { provider_receipt: outcome.idempotency_key, reply_verified: true } : {}) }; + } + if (outcome.kind === "conflict") return { status: "explicit_unverified", error: "manager_return_payload_conflict" }; + if (outcome.kind === "verify_unavailable" || outcome.kind === "locator_unavailable") { + return { status: "explicit_unverified", error: outcome.kind === "verify_unavailable" + ? "provider_verifier_unavailable" : "provider_locator_unavailable" }; + } + if (outcome.kind === "verified") { + let checked: JsonObject; + try { checked = classifyManagerReturnVerification(outcome.verification); } + catch (error) { + if (!(error instanceof EffectRuntimeRequestError)) throw error; + return { ...retryState(state, command, "provider_verification_unavailable"), + status: command.exact ? "retry_pending" : "verification_required" }; + } + return checked.status === "delivered" + ? { status: "delivered", delivered_at: command.now, message_id: returnMessageId(command), + provider_receipt: normalizeManagerReturnDeliveryAttempt(state.attempt).provider_receipt, + reply_verified: true, verification: checked.verification } + : checked.status === "explicit_unverified" ? { status: checked.status, error: checked.error } + : { ...retryState(state, command, checked.error), status: command.exact ? "retry_pending" : "verification_required" }; + } + if (outcome.kind === "sent" && outcome.external_write_performed === true) { + return { ...(state.attempt == null ? {} : { attempt: state.attempt }), + status: attemptLocator(state.attempt) === null ? "explicit_unverified" : "verification_required", + error: attemptLocator(state.attempt) === null ? "provider_locator_unavailable" : "provider_delivery_unverified" }; + } + const checked = classifyManagerReturnVerification({ verification_performed: false, + reply_verified: false, blocker: outcome.blocker ?? null }); + if (state.attempt != null && (checked.status === "explicit_unverified" || attemptLocator(state.attempt) === null)) { + return { status: "explicit_unverified", + error: checked.status === "explicit_unverified" ? checked.error : "provider_locator_unavailable" }; + } + if (!command.exact && state.status === "verification_required" && state.attempt != null) { + if (outcome.kind === "error") return state; + return { ...retryState(state, command, "provider_verification_unavailable"), status: "verification_required" }; + } + return retryState(command.exact ? state : { attempts: state.attempts ?? 0 }, command, outcome.kind === "sender_unavailable" + ? "return_transport_unavailable" : "original_route_or_return_delivery_unavailable"); +} + +async function finishReturn(command: ReturnCommand, state: JsonObject): Promise { + if (command.request.revision !== deliveryRevision(state)) throw new EffectRuntimeRequestError("return delivery state changed"); + const source = await inspectReturnPublication(command); + admitReturnLifetime(command, "original_return_settle"); + if (command.exact) { + if ((state.admission as JsonObject | undefined)?.token !== command.request.token) { + throw new EffectRuntimeRequestError("exact return delivery admission changed"); + } + } + if (command.request.phase === "attempt") { + const attempt = normalizeManagerReturnDeliveryAttempt(command.request.attempt); + if (state.attempt != null && !isDeepStrictEqual(normalizeManagerReturnDeliveryAttempt(state.attempt), attempt)) { + throw new EffectRuntimeRequestError("manager return delivery attempt conflict"); + } + const checkpoint = command.exact ? { ...state, attempt } + : { status: "verification_required", error: "provider_delivery_unverified", attempt }; + await writeDeliveryState(command, checkpoint); + return returnResponse({ revision: deliveryRevision(checkpoint), token: command.request.token }); + } + if (command.request.phase !== "settle") throw new EffectRuntimeRequestError("unsupported return delivery phase"); + const settled = deliveryOutcome(command, state); + if (requireJsonObject(command.request.outcome, "return outcome").kind === "private" + && !resolveConversationScope(requireJsonObject(command.facts.session, "source session")).private_conversation) { + throw new EffectRuntimeRequestError("external return requires a provider outcome"); + } + if (settled.status === "explicit_unverified" && state.attempt != null) settled.attempt = state.attempt; + if (command.exact) settled.goal_ref = requireJsonObject(command.facts.entry, "request").goal_ref; + if (source.reply.result_key !== undefined) settled.result_key = source.reply.result_key; + if (!isDeepStrictEqual(settled, state)) await writeDeliveryState(command, settled); + const outcome = requireJsonObject(command.request.outcome, "return outcome"); + return returnResponse({ processed: command.exact || settled.status === "delivered" + || ["error", "conflict", "sender_unavailable"].includes(String(outcome.kind)) ? 1 : 0, + observe_verified: command.exact && settled.status === "delivered" && settled.reply_verified === true }); +} + +/** One locked owner for admission, provider-attempt checkpoints and delivery receipts. */ +export async function transactManagerReturnDelivery(value: unknown): Promise { + const request = requireJsonObject(value, "return delivery transaction"); + if (request.phase === "scan") return scanReturns(request); + const command = returnCommand(value); + return withFileMutationLock(command.path, async () => { + let state: JsonObject; + try { state = deliveryState(await readReturnRecord(command.path, {})); } + catch (error) { + if (command.request.phase === "prepare" && unavailableRecord(error)) return returnResponse(); + throw error; + } + return command.request.phase === "prepare" ? prepareReturn(command, state) : finishReturn(command, state); + }); +} diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 809f5ee933..e1dc2e1634 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -695,8 +695,8 @@ export function createEffectRuntimeHandlers( ["reward_memory.app_outcome.bind", lazyHandler(() => import("./capabilities/reward_memory_app_outcome.ts"), ({bindRewardMemoryAppOutcome}) => bindRewardMemoryAppOutcome)], ["reward_memory.app_outcome.identity", lazyHandler(() => import("./capabilities/reward_memory_app_outcome.ts"), ({rewardMemoryAppOutcomeIdentity}) => rewardMemoryAppOutcomeIdentity)], [ - "manager.return_delivery.normalize_attempt", - lazyHandler(() => import("./collaboration/return_delivery.ts"), ({normalizeManagerReturnDeliveryAttempt}) => (params) => normalizeManagerReturnDeliveryAttempt(params.attempt)), + "manager.return_delivery.transact", + lazyHandler(() => import("./collaboration/return_delivery.ts"), ({transactManagerReturnDelivery}) => transactManagerReturnDelivery), ], [ "manager.return_delivery.classify_verification", diff --git a/tests/control_plane_ts/manager_return_transaction.test.ts b/tests/control_plane_ts/manager_return_transaction.test.ts new file mode 100644 index 0000000000..9188d406be --- /dev/null +++ b/tests/control_plane_ts/manager_return_transaction.test.ts @@ -0,0 +1,228 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import { mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises"; +import { tmpdir } from "node:os"; +import { join } from "node:path"; +import type { JsonObject } from "../../loopx/control_plane/effect_program.ts"; +import { RETURN_DELIVERY_TRANSACTION, transactManagerReturnDelivery } from "../../loopx/control_plane/collaboration/return_delivery.ts"; + +const id = "a".repeat(64); +const now = "2026-01-02T03:04:05.123456+00:00"; +const ref = { goal_id: "research", goal_instance_id: "ginst_" + "b".repeat(32) }; +const attempt = { schema_version: "manager_return_delivery_attempt_v0", provider: "lark", + message_ref: "om_result", intent_digest: "sha256:" + "c".repeat(64), + provider_receipt: "sha256:" + "d".repeat(64) }; + +async function fixture(exact: boolean, external = false) { + const root = await mkdtemp(join(tmpdir(), "loopx-return-transaction-")); + const folder = join(root, ".local", "manager-context"); + const entry: JsonObject = { request_id: id, goal_id: "research", agent_id: "worker", source_id: "source:original", + ...(exact ? { goal_ref: ref } : {}) }; + const route = { ...entry, session_id: "session", client_turn_id: "original", channel_id: external ? "manager.external.fixture" : "manager" }; + const reply = { ...entry, decision: "adopt", phase: "conclusion", text: "Independent findings." }; + const entryPath = join(folder, "entries", "e".repeat(64), id + ".json"); + const statePath = join(folder, "replies", id, "conclusion.delivery.json"); + async function put(path: string, body: unknown) { + await mkdir(join(path, ".."), { recursive: true }); + await writeFile(path, JSON.stringify(body)); + } + await put(entryPath, entry); + await put(join(folder, "roundtrips", id + ".json"), route); + await put(join(folder, "replies", id, "conclusion.json"), reply); + const facts: JsonObject = { entry, entry_path: entryPath, route, + session: { session_id: "session", goal_id: "loopx-manager", channel_id: route.channel_id, status: "open" }, + turn: { turn_id: "turn", client_turn_id: "original", status: "completed" }, + grant: { source_id: entry.source_id, targets: [{ goal_id: entry.goal_id, agent_id: entry.agent_id }] }, + scope: { profile_id: exact ? "source_session_v1" : null, + caller_goal_ref: exact ? ref : null, current_goal_ref: exact ? ref : null }, + initial_delivery: { request: entry, route, authorized_source_id: entry.source_id, + turn: { client_turn_id: "original", status: "completed" } } }; + const command = { schema_version: RETURN_DELIVERY_TRANSACTION, runtime_root: root, request_id: id, + result_key: "conclusion", exact, now, facts }; + return { root, folder, entryPath, statePath, facts, command, put, + state: async () => JSON.parse(await readFile(statePath, "utf8")), + run: (phase: string, fields: JsonObject = {}) => transactManagerReturnDelivery({ ...command, phase, ...fields }), + close: () => rm(root, { recursive: true, force: true }) }; +} + +for (const exact of [false, true]) { + test(`private return settles once and retains the original message identity (${exact})`, async () => { + const f = await fixture(exact); + try { + const plan = await f.run("prepare"); + assert.equal(plan.action, "private"); + assert.match(String(plan.message_id), /^handoff\.[a-f0-9]{64}$/); + const settled = await f.run("settle", { ...plan, outcome: { kind: "private" } }); + assert.equal(settled.processed, 1); + assert.equal((await f.state()).status, "delivered"); + const bytes = await readFile(f.statePath); + assert.equal((await f.run("prepare")).action, "skip"); + assert.deepEqual(await readFile(f.statePath), bytes); + } finally { await f.close(); } + }); + + test(`provider acceptance survives response loss and resumes by verification (${exact})`, async () => { + const f = await fixture(exact, true); + try { + const plan = await f.run("prepare"); + assert.equal(plan.action, "send"); + await f.run("attempt", { ...plan, attempt }); + const resumed = await f.run("prepare", { now: "2026-01-02T03:06:05.123456+00:00" }); + assert.equal(resumed.action, "verify"); + assert.deepEqual(resumed.attempt, attempt); + await f.run("settle", { ...resumed, now: "2026-01-02T03:06:05.123456+00:00", outcome: { kind: "verified", + verification: { verification_performed: true, reply_verified: true } } }); + assert.equal((await f.state()).status, "delivered"); + assert.equal((await f.state()).provider_receipt, attempt.provider_receipt); + assert.equal((await f.run("prepare")).action, "skip"); + } finally { await f.close(); } + }); + + test(`an accepted write without a locator is terminal and retains its evidence (${exact})`, async () => { + const f = await fixture(exact, true); + try { + const plan = await f.run("prepare"); + const locatorless = { ...attempt, message_ref: null }; + const saved = await f.run("attempt", { ...plan, attempt: locatorless }); + await f.run("settle", { ...saved, outcome: { kind: "sent", external_write_performed: true } }); + assert.equal((await f.state()).status, "explicit_unverified"); + assert.deepEqual((await f.state()).attempt, locatorless); + assert.equal((await f.run("prepare")).action, "skip"); + } finally { await f.close(); } + }); + + test(`corrupt delivery evidence and changed source bytes cannot be overwritten (${exact})`, async () => { + const f = await fixture(exact); + try { + await writeFile(f.statePath, "{broken"); + assert.equal((await f.run("prepare")).action, "skip"); + assert.equal(await readFile(f.statePath, "utf8"), "{broken"); + await f.put(f.statePath, { status: "unknown_delivery_state" }); + assert.equal((await f.run("prepare")).action, "skip"); + assert.deepEqual(await f.state(), { status: "unknown_delivery_state" }); + await rm(f.statePath); + await f.put(f.entryPath, { ...(f.facts.entry as JsonObject), source_id: "another-source" }); + assert.equal((await f.run("prepare")).action, "skip"); + if (exact) await assert.rejects(readFile(f.statePath)); + else assert.equal((await f.state()).status, "retry_pending"); + } finally { await f.close(); } + }); + + test(`a malformed saved attempt never supplies an external verification address (${exact})`, async () => { + const f = await fixture(exact, true); + try { + const corrupt = { ...attempt, intent_digest: "not-a-digest" }; + await f.put(f.statePath, { status: "verification_required", attempt: corrupt }); + const plan = await f.run("prepare"); + assert.equal(plan.action, "verify"); + assert.equal(plan.attempt, null); + await f.run("settle", { ...plan, outcome: { kind: "locator_unavailable" } }); + assert.equal((await f.state()).status, "explicit_unverified"); + assert.deepEqual((await f.state()).attempt, corrupt); + } finally { await f.close(); } + }); +} + +test("concurrent exact admission is single flight and stale settlement cannot overwrite a checkpoint", async () => { + const f = await fixture(true, true); + try { + const plans = await Promise.all([f.run("prepare"), f.run("prepare")]); + assert.equal(plans.filter(plan => plan.action === "send").length, 1); + const plan = plans.find(plan => plan.action === "send")!; + await f.run("attempt", { ...plan, attempt }); + const bytes = await readFile(f.statePath); + await assert.rejects(f.run("settle", { ...plan, outcome: { kind: "sent", reply_verified: true } }), /state changed/); + assert.deepEqual(await readFile(f.statePath), bytes); + } finally { await f.close(); } +}); + +test("an ambiguous expired external admission never dispatches another send", async () => { + const f = await fixture(true, true); + try { + assert.equal((await f.run("prepare")).action, "send"); + assert.equal((await f.run("prepare", { now: "2026-01-02T03:06:05.123456+00:00" })).action, "skip"); + assert.equal((await f.state()).status, "explicit_unverified"); + assert.equal((await f.state()).error, "provider_delivery_unverified"); + } finally { await f.close(); } +}); + +test("invalid transaction addresses and phases refuse before changing evidence", async () => { + const f = await fixture(false); + try { + for (const fields of [{ request_id: "../escape" }, { result_key: "../escape" }, + { schema_version: "unsupported" }, { phase: "unknown" }, { now: "not-a-clock" }]) { + await assert.rejects(transactManagerReturnDelivery({ ...f.command, phase: "prepare", ...fields })); + } + await assert.rejects(readFile(f.statePath)); + } finally { await f.close(); } +}); + +test("scan skips terminal and corrupt records but locked admission rechecks every hint", async () => { + const f = await fixture(false); + const scan = (candidates = [{ request_id: id, result_key: "conclusion" }]) => + transactManagerReturnDelivery({ ...f.command, phase: "scan", candidates }); + try { + assert.deepEqual((await scan()).candidates, [{ request_id: id, result_key: "conclusion" }]); + await f.put(f.statePath, { status: "delivered" }); + assert.deepEqual((await scan()).candidates, []); + assert.equal((await f.run("prepare")).action, "skip"); + await writeFile(f.statePath, "{broken"); + const unreadable = await scan(); + assert.deepEqual(unreadable.candidates, []); + assert.equal(unreadable.unavailable_count, 1); + assert.equal(await readFile(f.statePath, "utf8"), "{broken"); + await assert.rejects(scan(Array(129).fill({ request_id: id, result_key: "conclusion" })), /at most 128/); + await assert.rejects(transactManagerReturnDelivery({ ...f.command, phase: "scan", schema_version: "old", candidates: [] })); + } finally { await f.close(); } +}); + +test("provider exception after a locatorless checkpoint cannot become a retry", async () => { + const f = await fixture(false, true); + try { + const plan = await f.run("prepare"); + const saved = await f.run("attempt", { ...plan, attempt: { ...attempt, message_ref: null } }); + await f.run("settle", { ...saved, outcome: { kind: "error" } }); + assert.equal((await f.state()).status, "explicit_unverified"); + assert.equal((await f.state()).error, "provider_locator_unavailable"); + assert.equal((await f.run("prepare")).action, "skip"); + } finally { await f.close(); } +}); + +test("invalid UTF-8 and unexpected BOMs cannot become a repaired receipt or delivery instruction", async () => { + const f = await fixture(false); + try { + for (const bytes of [Buffer.from([0x7b, 0x22, 0xff, 0x22, 0x7d]), Buffer.from('\ufeff{"status":"queued"}')]) { + await writeFile(f.statePath, bytes); + assert.equal((await f.run("prepare")).action, "skip"); + assert.deepEqual(await readFile(f.statePath), bytes); + } + } finally { await f.close(); } +}); + +test("a surrogate result cannot be silently replaced while hashing or delivering", async () => { + const f = await fixture(false); + try { + const path = join(f.folder, "replies", id, "conclusion.json"); + const reply = JSON.parse(await readFile(path, "utf8")); + await f.put(path, { ...reply, text: "An incomplete character: \ud800" }); + const bytes = await readFile(path); + assert.equal((await f.run("prepare")).action, "skip"); + assert.deepEqual(await readFile(path), bytes); + } finally { await f.close(); } +}); + + +for (const [clock, retry] of [ + ["2026-01-02T03:04:05.123456", "2026-01-02T03:04:15.123456"], + ["2026-01-02T23:59:55+08:00", "2026-01-03T00:00:05+08:00"], + ["2026-01-02T03:04:05.123456+08:00:30", "2026-01-02T03:04:15.123456+08:00:30"], +]) { + test(`retry retains the Python clock's offset and fractional precision (${clock})`, async () => { + const f = await fixture(false, true); + try { + const plan = await f.run("prepare", { now: clock }); + await f.run("settle", { ...plan, now: clock, outcome: { kind: "sender_unavailable" } }); + assert.equal((await f.state()).retry_at, retry); + } finally { await f.close(); } + }); +} diff --git a/tests/test_collaboration_goal_instance.py b/tests/test_collaboration_goal_instance.py index fa1c7e8210..4467ff9d89 100644 --- a/tests/test_collaboration_goal_instance.py +++ b/tests/test_collaboration_goal_instance.py @@ -22,7 +22,6 @@ register_ingress, ) from loopx.capabilities.manager_context.roundtrip import ( - EXACT_DELIVERY_ADMISSION_SECONDS, ReturnResolutionBlocked, drain, report, @@ -40,6 +39,7 @@ INSTANCE_A = "ginst_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa" INSTANCE_B = "ginst_bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" +RETURN_ADMISSION_SECONDS = 60 @pytest.mark.skipif(os.open not in os.supports_dir_fd or not hasattr(os, "O_NOFOLLOW"), @@ -668,7 +668,7 @@ def __call__(self, *_args): store, transport, now=admitted_at - + timedelta(seconds=EXACT_DELIVERY_ADMISSION_SECONDS + 1), + + timedelta(seconds=RETURN_ADMISSION_SECONDS + 1), ) assert second.result(timeout=5) == 0 release.set() @@ -711,7 +711,7 @@ def test_exact_external_return_does_not_resend_after_expired_unknown_admission( "admitted_at": admitted_at.isoformat(), "expires_at": ( admitted_at - + timedelta(seconds=EXACT_DELIVERY_ADMISSION_SECONDS) + + timedelta(seconds=RETURN_ADMISSION_SECONDS) ).isoformat(), }, }, @@ -735,7 +735,7 @@ def __call__(self, *_args): store, transport, now=admitted_at - + timedelta(seconds=EXACT_DELIVERY_ADMISSION_SECONDS + 1), + + timedelta(seconds=RETURN_ADMISSION_SECONDS + 1), ) == 0 ) @@ -835,7 +835,7 @@ def verify(self, *_args): store, transport, now=admitted_at - + timedelta(seconds=EXACT_DELIVERY_ADMISSION_SECONDS + 1), + + timedelta(seconds=RETURN_ADMISSION_SECONDS + 1), ) == 1 ) @@ -854,7 +854,7 @@ def verify(self, *_args): assert failed["attempt"] == first["attempt"] # Another immediate pump must honor backoff, even after recreation. drain(tmp_path, registry, ChatSessionStore(tmp_path), transport, - now=admitted_at + timedelta(seconds=EXACT_DELIVERY_ADMISSION_SECONDS + 1)) + now=admitted_at + timedelta(seconds=RETURN_ADMISSION_SECONDS + 1)) assert transport.verify_calls == 1 drain(tmp_path, registry, ChatSessionStore(tmp_path), transport, now=admitted_at + timedelta(days=1)) @@ -1338,3 +1338,57 @@ def test_exact_instance_updates_keep_original_route_and_http_readback(tmp_path): assert messages[-1]["return_delivery"]["status"] == "delivered" with pytest.raises((ValueError, FileNotFoundError)): report(tmp_path, "delivery", "builder", rid, "conclusion", "Wrong generation.", registry=registry, caller_goal_ref={"goal_id": "delivery", "goal_instance_id": INSTANCE_B}, update_id="wrong-instance") + + +def test_native_return_process_exit_recovers_the_original_attempt(tmp_path): + from loopx.capabilities.manager_context import target_authority + from loopx.control_plane.collaboration.source_chat_observation import original_delivery_facts + + registry = _create_source_registry(tmp_path) + store, session, receipt = _external_manager_request(tmp_path, registry) + rid = receipt["request_id"] + folder = _root(tmp_path) + entry_path = folder / "entries" / _hash({"goal_ref": receipt["goal_ref"], "agent_id": "builder"}) / (rid + ".json") + entry = json.loads(entry_path.read_text()) + route = json.loads((folder / "roundtrips" / (rid + ".json")).read_text()) + turn = store.turn_for_client(session["session_id"], "owner-request") + grant = target_authority(tmp_path, session=session, turn=turn, + target={"goal_id": "delivery", "agent_id": "builder"}) + command = { + "schema_version": "manager_return_delivery_transaction_v0", "runtime_root": str(tmp_path), + "request_id": rid, "result_key": "conclusion", "exact": True, + "now": "2026-01-02T03:04:05.123456+00:00", + "facts": {"entry": entry, "entry_path": str(entry_path), "route": route, + "session": session, "turn": turn, "grant": grant, + "scope": {"profile_id": "source_session_v1", "caller_goal_ref": receipt["goal_ref"], + "current_goal_ref": receipt["goal_ref"]}, + "initial_delivery": original_delivery_facts(entry, route, turn, source_id=grant["source_id"])}, + } + module = (Path(__file__).parents[1] / "loopx/control_plane/collaboration/return_delivery.ts").as_uri() + + def invoke(phase, *, crash=False, **payload): + script = (f"import {{transactManagerReturnDelivery}} from {json.dumps(module)};" + "const result=await transactManagerReturnDelivery(JSON.parse(process.argv[1]));" + + ("process.exit(17);" if crash else "process.stdout.write(JSON.stringify(result));")) + child = subprocess.run(["node", "--no-warnings", "--experimental-strip-types", "--input-type=module", + "-e", script, json.dumps({**command, **payload, "phase": phase})], + capture_output=True, text=True, timeout=10) + assert child.returncode == (17 if crash else 0), child.stderr + return None if crash else json.loads(child.stdout) + + plan = invoke("prepare") + assert plan["action"] == "send" + attempt = {"schema_version": "manager_return_delivery_attempt_v0", "provider": "fixture", + "message_ref": "om_original", "intent_digest": "sha256:" + "a" * 64, + "provider_receipt": "sha256:" + "b" * 64} + invoke("attempt", crash=True, **{**plan, "attempt": attempt}) + resumed = invoke("prepare", now="2026-01-02T03:06:05.123456+00:00") + assert resumed["action"] == "verify" and resumed["attempt"] == attempt + invoke("settle", **resumed, now="2026-01-02T03:06:05.123456+00:00", + outcome={"kind": "verified", "verification": {"verification_performed": True, "reply_verified": True}}) + state_path = folder / "replies" / rid / "conclusion.delivery.json" + state = json.loads(state_path.read_text()) + assert state["status"] == "delivered" and state["goal_ref"] == receipt["goal_ref"] + assert state["provider_receipt"] == attempt["provider_receipt"] + assert invoke("prepare")["action"] == "skip" + assert json.loads(state_path.read_text()) == state diff --git a/tests/test_manager_context_roundtrip.py b/tests/test_manager_context_roundtrip.py index 8bfcde4f3d..c4b6328852 100644 --- a/tests/test_manager_context_roundtrip.py +++ b/tests/test_manager_context_roundtrip.py @@ -191,8 +191,7 @@ def sender(route, session, turn, text): assert reply_status(root, receipt)[0]["status"] == "delivered" -def test_exact_return_skips_other_host_before_admission(flow, monkeypatch): - from loopx.capabilities.manager_context import roundtrip +def test_exact_return_skips_other_host_before_admission(flow): from loopx.control_plane.projects.registry_codec import SOURCE_SESSION_PROFILE_ID root, registry, store, create = flow @@ -207,8 +206,9 @@ def test_exact_return_skips_other_host_before_admission(flow, monkeypatch): def unexpected(*args, **kwargs): raise AssertionError("A foreign host must not admit or deliver this return") - monkeypatch.setattr(roundtrip, "_exact_return_context", unexpected) + before = {path: path.read_bytes() for path in (_root(root) / "replies").rglob("*.json")} assert drain(root, registry, foreign, unexpected) == 0 + assert {path: path.read_bytes() for path in before} == before def test_conclusion_coalesces_unsent_intermediate_decision_and_is_immutable(flow): @@ -1276,81 +1276,44 @@ def unexpected(*_): assert len(stored) == 1 and stored[0]["attachments"] == [] -def test_exact_loop_payload_conflict_sets_terminal_state_without_retry( - flow, - monkeypatch, -): - root, registry, store, create = flow - session, turn, receipt = create(True) - rid = receipt["request_id"] - acknowledge(root, "research", "worker", rid, "adopt", "Checked") - report( - root, - "research", - "worker", - rid, - "conclusion", - "Checked with the current evidence and found no gap.", - ) - conflicting_mid = "handoff." + _hash([rid, "conclusion"]) - store.append_message( - session["session_id"], - role="agent", - text="An earlier, different version of this conclusion.", - turn_id=turn["turn_id"], - origin="manager_followup", - message_id=conflicting_mid, - ) - # The exact admission machinery needs a source-session registry fixture - # this module does not build; stub the context construction and the - # settlement writer so the exact loop's own delivery-side conflict - # check is what runs against the real divergent row above. - import loopx.capabilities.manager_context.roundtrip as return_roundtrip - from loopx.control_plane.projects.registry_codec import ( - SOURCE_SESSION_PROFILE_ID, - ) - - settlements = [] - - def fake_settlement(root, registry, context, value, **kwargs): - settlements.append(dict(value)) +def test_exact_loop_payload_conflict_sets_terminal_state_without_retry(tmp_path): + from loopx.control_plane.projects.registry_codec import source_session_registry_transaction - monkeypatch.setattr( - return_roundtrip, "_write_exact_return_state", fake_settlement - ) - - def fake_context(root, registry, store, path, state_path, now, effects): - reply = return_roundtrip._read(path) - return { - "reply": {**reply, "phase": path.stem}, - "row": {"request_id": rid, "agent_id": "worker"}, - "route": {"session_id": session["session_id"]}, - "session": session, - "turn": turn, - "store": store, - "state_path": state_path, - "state": {}, - "token": "fixture", - "source_id": "lark:om_fixture_source", - } - - monkeypatch.setattr(return_roundtrip, "_exact_return_context", fake_context) - profile = json.loads(registry.read_text()) - profile["profile_id"] = SOURCE_SESSION_PROFILE_ID - registry.write_text(json.dumps(profile)) + registry = tmp_path / ".loopx" / "registry.json" + with source_session_registry_transaction(registry, operation="test_exact_return", create=lambda: { + "profile_id": "source_session_v1", + "goals": [{"id": "research", "repo": str(tmp_path), "status": "active", + "goal_instance_id": "ginst_" + "a" * 32, + "coordination": {"registered_agents": ["worker"]}}], + "session_bindings": [], "session_receipts": [], "lifetime_receipts": [], + }) as transaction: + transaction.commit(transaction.payload_copy()) + store = ChatSessionStore(tmp_path) + session = store.create_session(goal_id="loopx-manager", agent_id="codex", adapter_kind="codex_app_server", + upstream_thread_id="fixture", channel_id="manager") + turn, _ = store.create_turn(session["session_id"], client_turn_id="original", + message="Review the evidence.", origin="web") + receipt = deliver(tmp_path, registry, session=session, turn=turn, + request={"goal_id": "research", "agent_id": "worker"}) + store.update_turn(session["session_id"], turn["turn_id"], status="completing", + response={"context_handoff_receipt": receipt}) + store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"]) + rid = receipt["request_id"] + acknowledge(tmp_path, "research", "worker", rid, "adopt", "Checked", registry=registry) + report(tmp_path, "research", "worker", rid, "conclusion", "Checked the evidence.", registry=registry) + mid = "handoff." + _hash([rid, "conclusion"]) + store.append_message(session["session_id"], role="agent", text="An earlier, different conclusion.", + turn_id=turn["turn_id"], origin="manager_followup", message_id=mid) deliveries = [] - - def recording_transport(route, session, turn, text): - deliveries.append(text) - return {"reply_verified": True, "idempotency_key": "sha256:provider-proof"} - - drain(root, registry, store, recording_transport) + assert drain(tmp_path, registry, store, lambda *args: deliveries.append(args)) == 1 assert deliveries == [] - assert len(settlements) == 1 - assert settlements[0] == { - "status": "explicit_unverified", - "error": "manager_return_payload_conflict", - } + state_path = _root(tmp_path) / "replies" / rid / "conclusion.delivery.json" + state = json.loads(state_path.read_text()) + assert state["status"] == "explicit_unverified" + assert state["error"] == "manager_return_payload_conflict" + assert state["goal_ref"] == receipt["goal_ref"] + assert drain(tmp_path, registry, ChatSessionStore(tmp_path), None) == 0 + assert json.loads(state_path.read_text()) == state def test_explicit_channel_return_completes_transcript_only_without_sender(flow): From 5899a4c7a331c50aab9686241e175f8c3fff5f1d Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Mon, 19 Oct 2026 04:44:59 +0000 Subject: [PATCH 2/6] test(collaboration): decode native process receipts as UTF-8 Signed-off-by: aihop <4408344+hhyykk@users.noreply.github.com> --- tests/test_collaboration_goal_instance.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/tests/test_collaboration_goal_instance.py b/tests/test_collaboration_goal_instance.py index 4467ff9d89..7134abee34 100644 --- a/tests/test_collaboration_goal_instance.py +++ b/tests/test_collaboration_goal_instance.py @@ -1372,7 +1372,7 @@ def invoke(phase, *, crash=False, **payload): + ("process.exit(17);" if crash else "process.stdout.write(JSON.stringify(result));")) child = subprocess.run(["node", "--no-warnings", "--experimental-strip-types", "--input-type=module", "-e", script, json.dumps({**command, **payload, "phase": phase})], - capture_output=True, text=True, timeout=10) + capture_output=True, text=True, encoding="utf-8", timeout=10) assert child.returncode == (17 if crash else 0), child.stderr return None if crash else json.loads(child.stdout) From fdbbd91c2b27ca33945a95b8cc8d79d9b7ffc681 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 19 Oct 2026 19:55:01 +0800 Subject: [PATCH 3/6] fix(collaboration): preserve return failure isolation and recovery Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../collaboration/return_delivery.ts | 29 +++++++-- .../manager_return_transaction.test.ts | 46 ++++++++++++++ tests/test_manager_context_roundtrip.py | 63 +++++++++++++++++++ 3 files changed, 132 insertions(+), 6 deletions(-) diff --git a/loopx/control_plane/collaboration/return_delivery.ts b/loopx/control_plane/collaboration/return_delivery.ts index 1ee6831953..d1f102357a 100644 --- a/loopx/control_plane/collaboration/return_delivery.ts +++ b/loopx/control_plane/collaboration/return_delivery.ts @@ -146,6 +146,12 @@ interface ReturnCommand { facts: JsonObject; } +function returnAddress(value: unknown): {request_id: string; result_key: string} { + const row = requireJsonObject(value, "return candidate"); + return {request_id: matchingString(row.request_id, "request_id", BARE_SHA256_PATTERN), + result_key: matchingString(row.result_key, "result_key", RESULT_KEY)}; +} + function returnCommand(value: unknown): ReturnCommand { const request = requireJsonObject(value, "return delivery transaction"); if (request.schema_version !== RETURN_DELIVERY_TRANSACTION) { @@ -153,8 +159,7 @@ function returnCommand(value: unknown): ReturnCommand { } const root = requireNonEmptyString(request.runtime_root, "runtime_root"); if (!isAbsolute(root)) throw new EffectRuntimeRequestError("return runtime root must be absolute"); - const id = matchingString(request.request_id, "request_id", BARE_SHA256_PATTERN); - const key = matchingString(request.result_key, "result_key", RESULT_KEY); + const {request_id: id, result_key: key} = returnAddress(request); const now = requireNonEmptyString(request.now, "now"); if (now.length > 64 || parseTodoTimestampMicros(now) === null) throw new EffectRuntimeRequestError("invalid return clock"); requireStringLiteral(request.phase, ["prepare", "attempt", "settle"], "return delivery phase"); @@ -224,11 +229,19 @@ async function scanReturns(request: JsonObject): Promise { if (!Array.isArray(request.candidates) || request.candidates.length > 128) { throw new EffectRuntimeRequestError("return scan requires at most 128 candidates"); } - const commands = request.candidates.map(candidate => returnCommand({ ...request, - ...requireJsonObject(candidate, "return candidate"), phase: "prepare", facts: {} })); const candidates = []; let unavailable = 0; - for (const command of commands) { + for (const candidate of request.candidates) { + let address: ReturnType; + try { address = returnAddress(candidate); } + catch (error) { + if (!(error instanceof EffectRuntimeRequestError)) throw error; + unavailable += 1; + continue; + } + // A damaged address is local to that item; an invalid transaction envelope + // still rejects the scan. Candidate fields cannot replace its root or scope. + const command = returnCommand({...request, ...address, phase: "prepare", facts: {}}); try { const state = deliveryState(await readReturnRecord(command.path, {})); if (pendingDelivery(command, state)) candidates.push({ request_id: command.request.request_id, result_key: command.key }); @@ -348,6 +361,9 @@ async function sourceFailure(command: ReturnCommand, state: JsonObject, blocker: } function returnAuthorityBlocker(command: ReturnCommand): string | null { + // Null is the Host's completed lookup of a missing original record. IO errors + // arrive separately in facts.error and retain their retryable classification. + if (command.facts.session === null || command.facts.turn === null) return "original_route_unavailable"; const session = requireJsonObject(command.facts.session, "source session"); requireJsonObject(command.facts.turn, "source turn"); const route = requireJsonObject(command.facts.route, "source route"); @@ -463,7 +479,8 @@ function deliveryOutcome(command: ReturnCommand, state: JsonObject): JsonObject return { ...retryState(state, command, "provider_verification_unavailable"), status: "verification_required" }; } return retryState(command.exact ? state : { attempts: state.attempts ?? 0 }, command, outcome.kind === "sender_unavailable" - ? "return_transport_unavailable" : "original_route_or_return_delivery_unavailable"); + ? "return_transport_unavailable" : outcome.kind === "verification_error" + ? "provider_verification_unavailable" : "original_route_or_return_delivery_unavailable"); } async function finishReturn(command: ReturnCommand, state: JsonObject): Promise { diff --git a/tests/control_plane_ts/manager_return_transaction.test.ts b/tests/control_plane_ts/manager_return_transaction.test.ts index 9188d406be..7c16551856 100644 --- a/tests/control_plane_ts/manager_return_transaction.test.ts +++ b/tests/control_plane_ts/manager_return_transaction.test.ts @@ -188,6 +188,52 @@ test("provider exception after a locatorless checkpoint cannot become a retry", } finally { await f.close(); } }); +test("scan isolates malformed addresses while retaining transaction envelope validation", async () => { + const f = await fixture(false); + try { + const result = await f.run("scan", {candidates: [ + {request_id: "!orphan", result_key: "conclusion"}, null, + {request_id: id, result_key: "../escape"}, {request_id: id, result_key: "conclusion"}, + ]}); + assert.deepEqual(result.candidates, [{request_id: id, result_key: "conclusion"}]); + assert.equal(result.unavailable_count, 3); + await assert.rejects(f.run("scan", {runtime_root: "relative", candidates: [ + {request_id: id, result_key: "conclusion"}, + ]}), /absolute/); + await assert.rejects(readFile(f.statePath)); + } finally { await f.close(); } +}); + +for (const missing of ["session", "turn"]) { + test(`accepted legacy attempt terminates when the original ${missing} is absent`, async () => { + const f = await fixture(false, true); + try { + await f.put(f.statePath, {status: "verification_required", attempt}); + const facts = {...f.facts, [missing]: null}; + assert.equal((await f.run("prepare", {facts})).action, "skip"); + assert.equal((await f.state()).status, "explicit_unverified"); + assert.equal((await f.state()).error, "original_route_unavailable"); + assert.deepEqual((await f.state()).attempt, attempt); + assert.equal((await f.run("prepare", {facts})).processed, 0); + } finally { await f.close(); } + }); +} + +test("exact verifier failure retains its provider diagnostic and accepted attempt", async () => { + const f = await fixture(true, true); + try { + const plan = await f.run("prepare"); + await f.run("attempt", {...plan, attempt}); + const resumed = await f.run("prepare", {now: "2026-01-02T03:06:05.123456+00:00"}); + await f.run("settle", {...resumed, now: "2026-01-02T03:06:05.123456+00:00", + outcome: {kind: "verification_error", blocker: null}}); + assert.equal((await f.state()).status, "retry_pending"); + assert.equal((await f.state()).error, "provider_verification_unavailable"); + assert.deepEqual((await f.state()).attempt, attempt); + assert.equal((await f.run("prepare", {now: "2026-01-02T03:08:05.123456+00:00"})).action, "verify"); + } finally { await f.close(); } +}); + test("invalid UTF-8 and unexpected BOMs cannot become a repaired receipt or delivery instruction", async () => { const f = await fixture(false); try { diff --git a/tests/test_manager_context_roundtrip.py b/tests/test_manager_context_roundtrip.py index c4b6328852..350c51b246 100644 --- a/tests/test_manager_context_roundtrip.py +++ b/tests/test_manager_context_roundtrip.py @@ -464,6 +464,69 @@ def ambiguous(*args): } +def test_return_scan_isolates_invalid_address_from_valid_result(flow): + root, registry, store, create = flow + session, turn, receipt = create() + rid = receipt["request_id"] + acknowledge(root, "research", "worker", rid, "adopt", "Checked") + report(root, "research", "worker", rid, "conclusion", "The original request is complete.") + orphan = _root(root) / "replies" / "!orphan" / "conclusion.json" + _write(orphan, {"text": "An unrelated damaged address."}) + original = orphan.read_bytes() + + assert drain(root, registry, store, None) == 1 + assert drain(root, registry, ChatSessionStore(root), None) == 0 + assert reply_status(root, receipt)[0]["status"] == "delivered" + returned = [row for row in store.messages(session["session_id"]) + if row.get("origin") == "manager_followup"] + assert len(returned) == 1 and returned[0]["turn_id"] == turn["turn_id"] + assert orphan.read_bytes() == original + assert not orphan.with_name("conclusion.delivery.json").exists() + + +@pytest.mark.parametrize("missing", ["session", "turn"]) +def test_accepted_return_with_missing_original_route_terminates_without_resend(flow, missing): + root, registry, store, create = flow + session, turn, receipt = create(True) + rid = receipt["request_id"] + acknowledge(root, "research", "worker", rid, "adopt", "Checked") + report(root, "research", "worker", rid, "conclusion", "The original request is complete.") + attempt = { + "schema_version": "manager_return_delivery_attempt_v0", "provider": "lark", + "message_ref": "om_recorded_result", "intent_digest": "sha256:" + "a" * 64, + "provider_receipt": "sha256:" + "b" * 64, + } + + class Transport: + sends = 0 + + def send_with_attempt(self, route, session, turn, text, record_attempt): + self.sends += 1 + record_attempt(attempt) + return {"external_write_performed": True, "reply_verified": False} + + def verify(self, *_): + pytest.fail("a missing original route cannot authorize verification") + + sender = Transport() + drain(root, registry, store, sender) + path = _root(root) / "replies" / rid / "conclusion.delivery.json" + assert json.loads(path.read_text())["status"] == "verification_required" + directory = store.root / "sessions" / session["session_id"] + (directory / "session.json" if missing == "session" else + directory / "turns" / (turn["turn_id"] + ".json")).unlink() + + assert drain(root, registry, ChatSessionStore(root), sender) == 1 + state = json.loads(path.read_text()) + assert state["status"] == "explicit_unverified" + assert state["error"] == "original_route_unavailable" + assert state["attempt"] == attempt + saved = path.read_bytes() + assert drain(root, registry, ChatSessionStore(root), sender, + now=datetime.now(timezone.utc) + timedelta(days=1)) == 0 + assert sender.sends == 1 and path.read_bytes() == saved + + def test_known_provider_locator_is_verified_after_restart_without_resend(flow): root, registry, store, create = flow session, _, receipt = create(True) From 8af44c5853f8b330118a2acdc9030b316767392a Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 19 Oct 2026 20:31:19 +0800 Subject: [PATCH 4/6] perf(collaboration): reuse the executable revision for one inbox command Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/cli_commands/manager_inbox.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/loopx/cli_commands/manager_inbox.py b/loopx/cli_commands/manager_inbox.py index 53ae1cbf4d..c14106c677 100644 --- a/loopx/cli_commands/manager_inbox.py +++ b/loopx/cli_commands/manager_inbox.py @@ -10,6 +10,7 @@ configure_evidence_scope, ) from ..control_plane.projects.registry_codec import load_project_registry +from ..control_plane.effect_runtime import effect_runtime_request_scope from ..presentation.answer_instruction import collaboration_answer_instruction @@ -88,6 +89,7 @@ def register_manager_inbox(subparsers, add_format): parser.add_argument("--reason") +@effect_runtime_request_scope() def handle_manager_inbox(args, registry_path, runtime_root): try: if getattr(args, "read_goal_scope", None) is not None and args.manager_inbox_action != "configure-read-scope": From 6bf9e5a96e1967b3848c9bc0ba79fabce569c83a Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Thu, 22 Oct 2026 23:03:51 +0800 Subject: [PATCH 5/6] refactor(collaboration): reuse the shared inbox dispatcher scope Signed-off-by: aihop <4408344+hhyykk@users.noreply.github.com> --- loopx/cli_commands/manager_inbox.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/loopx/cli_commands/manager_inbox.py b/loopx/cli_commands/manager_inbox.py index c14106c677..53ae1cbf4d 100644 --- a/loopx/cli_commands/manager_inbox.py +++ b/loopx/cli_commands/manager_inbox.py @@ -10,7 +10,6 @@ configure_evidence_scope, ) from ..control_plane.projects.registry_codec import load_project_registry -from ..control_plane.effect_runtime import effect_runtime_request_scope from ..presentation.answer_instruction import collaboration_answer_instruction @@ -89,7 +88,6 @@ def register_manager_inbox(subparsers, add_format): parser.add_argument("--reason") -@effect_runtime_request_scope() def handle_manager_inbox(args, registry_path, runtime_root): try: if getattr(args, "read_goal_scope", None) is not None and args.manager_inbox_action != "configure-read-scope": From 80bbbccf4129293dc2c3fef6137f51c167b1c30f Mon Sep 17 00:00:00 2001 From: aihop <4408344+hhyykk@users.noreply.github.com> Date: Thu, 22 Oct 2026 23:47:50 +0800 Subject: [PATCH 6/6] perf(collaboration): defer quota projections on inbox startup Signed-off-by: aihop <4408344+hhyykk@users.noreply.github.com> --- loopx/history.py | 2 +- loopx/registry.py | 3 ++- tests/test_cli_entrypoint.py | 3 ++- 3 files changed, 5 insertions(+), 3 deletions(-) diff --git a/loopx/history.py b/loopx/history.py index 20729b9efb..f9ba77885e 100644 --- a/loopx/history.py +++ b/loopx/history.py @@ -68,7 +68,6 @@ markdown_table_row, markdown_table_separator, ) -from .quota import goal_quota_with_spend_ledger from .registry import registry_goals STATUS_NEUTRAL_CLASSIFICATIONS = { @@ -407,6 +406,7 @@ def collect_history( project_goal_with_builtin_machine_configuration, ) from .capabilities.machine_configuration.store import read_machine_configuration + from .quota import goal_quota_with_spend_ledger if registry is None: registry = load_registry(registry_path) diff --git a/loopx/registry.py b/loopx/registry.py index bc1108a112..640bb08f12 100644 --- a/loopx/registry.py +++ b/loopx/registry.py @@ -17,7 +17,6 @@ from .explore_graph import compact_explore_graph_policy from .orchestration import compact_orchestration_policy, orchestration_policy_summary from .presentation.markdown import markdown_code, markdown_table_row, markdown_table_separator -from .quota import goal_quota_config def read_json(path: Path) -> dict[str, Any]: @@ -329,6 +328,8 @@ def inspect_registry_boundary(path: Path) -> dict[str, Any]: def inspect_registry(path: Path) -> dict[str, Any]: + from .quota import goal_quota_config + if not path.exists(): return { "ok": False, diff --git a/tests/test_cli_entrypoint.py b/tests/test_cli_entrypoint.py index ac476bf1b7..35229b5e11 100644 --- a/tests/test_cli_entrypoint.py +++ b/tests/test_cli_entrypoint.py @@ -499,7 +499,8 @@ def test_selected_manager_result_matches_full_replay_without_loading_full_cli(tm argv = ["--registry", str(registry), "--runtime-root", str(tmp_path), "--format", "json", "manager-inbox", "report", "--goal-id", "delivery", "--agent-id", "builder", "--request-id", rid, "--reply-text", "The independent evidence check passed."] - selected = run_cli_main("loopx.entrypoint", argv, forbidden_modules=("loopx.cli",)) + selected = run_cli_main("loopx.entrypoint", argv, + forbidden_modules=("loopx.cli", "loopx.quota")) path = _root(tmp_path) / "replies" / rid / "conclusion.json" assert selected.returncode == 0, selected.stderr committed = path.read_bytes()