From 9030951015f4db4df43d75d52a8586465b638c0b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sat, 7 Nov 2026 14:02:36 +0800 Subject: [PATCH] perf(chat): batch original delegation lifecycle checks Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../collaboration/conversation_inventory.py | 33 +++++++------ .../collaboration/goal_instance_lifecycle.ts | 19 ++++++- .../collaboration/goal_instance_scope.py | 49 +++++++++++++++++-- .../control_plane/effect_runtime_handlers.ts | 1 + .../test_chat_request_inventory.py | 39 ++++++++++++--- ...laboration_goal_instance_lifecycle.test.ts | 24 ++++++++- 6 files changed, 136 insertions(+), 29 deletions(-) diff --git a/loopx/control_plane/collaboration/conversation_inventory.py b/loopx/control_plane/collaboration/conversation_inventory.py index c399670fac..97fd5e28c5 100644 --- a/loopx/control_plane/collaboration/conversation_inventory.py +++ b/loopx/control_plane/collaboration/conversation_inventory.py @@ -12,7 +12,7 @@ import json from ..content_digest import BARE_SHA256_PATTERN -from .goal_instance_scope import CollaborationLifecycleRejected, collaboration_goal_scope, decide_collaboration_lifecycle +from .goal_instance_scope import collaboration_goal_scope, inspect_original_request_batch from .inbox import _entry, _now, _read, _receipt, _request_lock, _root, read_inbox_transaction, reply_status from .links import read_linked_work from .source_chat_observation import is_source_chat_host, original_delivery_facts, original_turn_reader @@ -175,7 +175,7 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core caller_goal_ref=requests[0][1].get("goal_ref")) as scope, _locked_request_batch(root, scope, requests) as available, ): - prepared = [] + candidates = [] for path, route, request_id in requests: if path not in available: failed += 1 @@ -196,23 +196,24 @@ def read_conversation_requests(root, registry, store, session, *, goal_ids, core continue grant = source_context_target_authority(root, source_session, turn, {"goal_id": gid, "agent_id": route["agent_id"]}) - try: - decide_collaboration_lifecycle(scope, operation="original_request_inspect", record=row, route=route, - initial_delivery=original_delivery_facts(row, route, turn, source_id=grant["source_id"])) - except CollaborationLifecycleRejected as exc: - # Only this typed refusal proves the route and source - # matched before the conflicting receipt was observed. - # Withhold the brief, work and completion state as before. - if exc.decision.get("delivery_proof_reason") == "handoff_receipt_conflict" and len(diagnostics) < 12: - diagnostics.append({"request_id": row["request_id"], "goal_id": gid, "agent_id": row["agent_id"], - "code": exc.decision["code"], "reason": exc.decision["delivery_proof_reason"]}) - raise - # Do not retain full model responses while waiting for - # linked-work IO; only the displayed question is used. + delivery = original_delivery_facts(row, route, turn, source_id=grant["source_id"]) + # Keep the displayed question and original handoff proof. source_turn = {"turn_id": turn["turn_id"], "message": turn.get("message", "")} - prepared.append((row, source_session["session_id"], source_turn)) + candidates.append((row, source_session["session_id"], source_turn, delivery)) except (OSError, ValueError, KeyError, TypeError): failed += 1 + decisions = inspect_original_request_batch(scope, [item[3] for item in candidates]) + prepared = [] + for (row, source_session_id, source_turn, _), decision in zip(candidates, decisions, strict=True): + if decision["kind"] == "reject": + failed += 1 + # The typed owner only reports a conflict after matching + # the original route and source. + if decision.get("delivery_proof_reason") == "handoff_receipt_conflict" and len(diagnostics) < 12: + diagnostics.append({"request_id": row["request_id"], "goal_id": gid, "agent_id": row["agent_id"], + "code": decision["code"], "reason": decision["delivery_proof_reason"]}) + continue + prepared.append((row, source_session_id, source_turn)) observations = _read_receipt_batch(root, [row for row, _, _ in prepared]) for (row, source_session_id, source_turn), observed in zip(prepared, observations, strict=True): if observed is None: diff --git a/loopx/control_plane/collaboration/goal_instance_lifecycle.ts b/loopx/control_plane/collaboration/goal_instance_lifecycle.ts index b56cdb4858..970b2baa2f 100644 --- a/loopx/control_plane/collaboration/goal_instance_lifecycle.ts +++ b/loopx/control_plane/collaboration/goal_instance_lifecycle.ts @@ -1,6 +1,6 @@ import type { JsonObject } from "../effect_program.ts"; import { EffectRuntimeRequestError } from "../effect_runtime_errors.ts"; -import { assertNever, jsonObject } from "../runtime_decode.ts"; +import { assertNever, jsonObject, requireJsonObject } from "../runtime_decode.ts"; import { BARE_SHA256_PATTERN } from "../content_digest.ts"; import { parseExactGoalRef, @@ -353,3 +353,20 @@ export function decideCollaborationLifecycle( "operation" in facts ? decideExactLifecycle(facts) : facts, raw, selectedOperation, ); } + +/** Decide one existing inventory batch under the caller's retained lifetime guard. */ +export function inspectOriginalRequestBatch(value: JsonObject): JsonObject { + const { profile_id, caller_goal_ref, current_goal_ref, items } = value; + if (!Array.isArray(items) || items.length > 12) { + throw new EffectRuntimeRequestError("original request inspection accepts at most 12 items"); + } + return { + items: items.map((item) => decideCollaborationLifecycle({ + ...requireJsonObject(item, "original request inspection item"), + profile_id, + caller_goal_ref, + current_goal_ref, + operation: "original_request_inspect", + })), + }; +} diff --git a/loopx/control_plane/collaboration/goal_instance_scope.py b/loopx/control_plane/collaboration/goal_instance_scope.py index 7932328520..e36efabb15 100644 --- a/loopx/control_plane/collaboration/goal_instance_scope.py +++ b/loopx/control_plane/collaboration/goal_instance_scope.py @@ -8,7 +8,7 @@ from ...agent_registry import registered_agent_ids_for_goal from ...file_lock import exclusive_cross_runtime_file_lock -from ..effect_runtime import effect_runtime_result +from ..effect_runtime import EffectRuntimeRejected, effect_runtime_result from ..goals.activation import goal_is_stopped from ..goals.source_session_registry_state import exact_goal_ref, guard_path from ..projects.registry_codec import ( @@ -185,6 +185,14 @@ def __init__(self, decision: dict[str, Any]): super().__init__(f"collaboration lifecycle rejected: {decision.get('code')}") +def _lifecycle_scope_facts(scope: CollaborationGoalScope) -> dict[str, Any]: + return { + "profile_id": scope.profile_id, + "caller_goal_ref": scope.caller_goal_ref, + "current_goal_ref": scope.current_goal_ref, + } + + def decide_collaboration_lifecycle( scope: CollaborationGoalScope, *, @@ -198,10 +206,8 @@ def decide_collaboration_lifecycle( result = effect_runtime_result( "collaboration.goal_instance.decide", { - "profile_id": scope.profile_id, + **_lifecycle_scope_facts(scope), "operation": operation, - "caller_goal_ref": scope.caller_goal_ref, - "current_goal_ref": scope.current_goal_ref, "record_goal_ref": ( record.get("goal_ref") if isinstance(record, dict) else None ), @@ -216,3 +222,38 @@ def decide_collaboration_lifecycle( if result.get("kind") == "reject": raise CollaborationLifecycleRejected(result) return result + + +def inspect_original_request_batch( + scope: CollaborationGoalScope, deliveries: list[dict[str, Any]], +) -> list[dict[str, Any]]: + """Read independent decisions while the caller holds the existing batch guards.""" + if not deliveries: + return [] + scope_facts = _lifecycle_scope_facts(scope) + items = [{ + "record_goal_ref": delivery["request"].get("goal_ref"), + "route_goal_ref": delivery["route"].get("goal_ref"), + "initial_delivery": delivery, + } for delivery in deliveries] + try: + result = effect_runtime_result( + "collaboration.goal_instance.inspect_original_requests", + {**scope_facts, "items": items}, + ) + decisions = result.get("items") if isinstance(result, dict) else None + except EffectRuntimeRejected as error: + if error.diagnostic_code != "request_too_large": + raise + # Large historical proofs that fit individually keep the ordinary wire. + decisions = [ + effect_runtime_result("collaboration.goal_instance.decide", { + **scope_facts, **item, "operation": "original_request_inspect", + }) for item in items + ] + if not isinstance(decisions, list) or len(decisions) != len(items) or any( + not isinstance(decision, dict) or decision.get("kind") not in {"allow", "legacy", "reject"} + for decision in decisions + ): + raise RuntimeError("original request inspection count or decision is invalid") + return decisions diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index f9d7223893..e2517c4950 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -662,6 +662,7 @@ export function createEffectRuntimeHandlers( "collaboration.goal_instance.decide", lazyHandler(() => import("./collaboration/goal_instance_lifecycle.ts"), ({decideCollaborationLifecycle}) => (params) => decideCollaborationLifecycle(params)), ], + ["collaboration.goal_instance.inspect_original_requests", lazyHandler(() => import("./collaboration/goal_instance_lifecycle.ts"), ({inspectOriginalRequestBatch}) => inspectOriginalRequestBatch)], ["external_evidence.discover", lazyHandler(() => import("./capabilities/external_evidence.ts"), ({projectExternalEvidenceDiscovery}) => projectExternalEvidenceDiscovery)], ["external_evidence.plan", lazyHandler(() => import("./capabilities/external_evidence.ts"), ({planExternalEvidenceRequest}) => planExternalEvidenceRequest)], ["external_evidence.receipt", lazyHandler(() => import("./capabilities/external_evidence.ts"), ({recordExternalEvidenceReceiptObservation}) => recordExternalEvidenceReceiptObservation)], diff --git a/tests/control_plane/test_chat_request_inventory.py b/tests/control_plane/test_chat_request_inventory.py index d714b7cd9d..6ac9ffa038 100644 --- a/tests/control_plane/test_chat_request_inventory.py +++ b/tests/control_plane/test_chat_request_inventory.py @@ -512,8 +512,8 @@ def test_inventory_rereads_goal_lifetime_between_batches(api, monkeypatch, limit response={"context_handoff_receipt": delivered}) server.chat_store.finalize_managed_turn_completion(session["session_id"], turn["turn_id"]) original_scope = inventory.collaboration_goal_scope - original_decide = inventory.decide_collaboration_lifecycle - calls, modes = [], [] + original_inspect = inventory.inspect_original_request_batch + calls, modes, batch_sizes = [], [], [] @contextmanager def scope(*args, **kwargs): @@ -525,21 +525,46 @@ def scope(*args, **kwargs): calls.append(current.current_goal_ref) yield current - def decide(*args, **kwargs): - result = original_decide(*args, **kwargs) - modes.append(result.get("mode")) - return result + def inspect(current, deliveries): + results = original_inspect(current, deliveries) + batch_sizes.append(len(deliveries)) + modes.extend(result.get("mode") for result in results) + return results monkeypatch.setattr(inventory, "collaboration_goal_scope", scope) - monkeypatch.setattr(inventory, "decide_collaboration_lifecycle", decide) + monkeypatch.setattr(inventory, "inspect_original_request_batch", inspect) code, value = _inventory(api, route["session_id"], view="history", limit=limit) assert code == 200 and len(calls) == 2 assert calls[0] == f["ref"] and calls[1] != f["ref"] # The existing typed owner permits proven historical inspection after a # replacement; it must observe the new lifetime rather than reuse the old. + assert batch_sizes == [12, 1] assert modes == ["current_instance"] * 12 + ["historical_read"] assert value["matched"] == 13 and value["coverage"] == {"scan_complete": True, "unreadable": 0, "diagnostics": []} +def test_inventory_falls_back_only_when_inspection_exceeds_the_wire_limit(api, monkeypatch): + from loopx.control_plane.collaboration import goal_instance_scope + from loopx.control_plane.effect_runtime import EffectRuntimeRejected + + _, _, route = _source(api) + original = goal_instance_scope.effect_runtime_result + operations = [] + batch_method = "collaboration.goal_instance.inspect_original_requests" + single_method = "collaboration.goal_instance.decide" + + def result(method, params): + operations.append(method) + if method == batch_method: + raise EffectRuntimeRejected("oversized inspection", diagnostic_code="request_too_large") + return original(method, params) + + monkeypatch.setattr(goal_instance_scope, "effect_runtime_result", result) + code, value = _inventory(api, route["session_id"], view="history") + assert code == 200 and value["matched"] == 1 + assert value["rows"][0]["request_id"] == route["request_id"] + assert operations == [batch_method, single_method] + + def test_conflicting_receipt_is_diagnosable_but_does_not_authorize_content_or_controls(api): control, _, route = _source(api) _, f, server = api diff --git a/tests/control_plane_ts/collaboration_goal_instance_lifecycle.test.ts b/tests/control_plane_ts/collaboration_goal_instance_lifecycle.test.ts index 2ac7ba80ca..3e326c3788 100644 --- a/tests/control_plane_ts/collaboration_goal_instance_lifecycle.test.ts +++ b/tests/control_plane_ts/collaboration_goal_instance_lifecycle.test.ts @@ -1,7 +1,7 @@ import assert from "node:assert/strict"; import test from "node:test"; -import { decideCollaborationLifecycle, proveOriginalRequestDelivery } from "../../loopx/control_plane/collaboration/goal_instance_lifecycle.ts"; +import { decideCollaborationLifecycle, inspectOriginalRequestBatch, proveOriginalRequestDelivery } from "../../loopx/control_plane/collaboration/goal_instance_lifecycle.ts"; const INSTANCE_A = "ginst_aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa"; const INSTANCE_B = "ginst_bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb"; @@ -263,3 +263,25 @@ test("readback observes a committed active request without permitting its result current_goal_ref: GOAL_B, initial_delivery: committedDelivery("failed"), })), { kind: "allow", mode: "historical_read", goal_ref: GOAL_A }); }); + +test("a bounded inspection batch keeps independent request refusals and order", () => { + for (const profile_id of [null, "source_session_v1"]) { + const valid = committedDelivery("running", profile_id !== null); + const conflict = { ...valid, turn: { ...valid.turn, context_handoff_receipt: {} } }; + const wrongSource = { ...conflict, authorized_source_id: "web:other" }; + const scope = { profile_id, caller_goal_ref: GOAL_A, current_goal_ref: GOAL_B }; + const item = (initial_delivery: typeof valid) => ({ + initial_delivery, record_goal_ref: GOAL_A, route_goal_ref: GOAL_A, + }); + assert.deepEqual(inspectOriginalRequestBatch({ + ...scope, items: [item(conflict), item(valid), item(wrongSource)], + }), { + items: [ + { kind: "reject", code: "initial_delivery_unproved", delivery_proof_reason: "handoff_receipt_conflict" }, + profile_id === null ? { kind: "legacy" } : { kind: "allow", mode: "historical_read", goal_ref: GOAL_A }, + { kind: "reject", code: "initial_delivery_unproved" }, + ], + }); + assert.throws(() => inspectOriginalRequestBatch({ ...scope, items: Array(13).fill(item(valid)) }), /at most 12/); + } +});