perf(chat): batch original delegation lifecycle checks

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:

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",
+    })),
+  };
+}

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

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)],

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

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/);
+  }
+});