fix(manager): project all peer facts through typed global reads

docs/reference/protocols/global-manager-command-v0.md

@@ -114,17 +114,79 @@ Implementations may read only compact LoopX control-plane surfaces:
 - rollout event log summaries;
 - review packets for explicit goal drilldown.
 
-`loopx global-gates` reads the compact status projection. Status internally
-consumes compact run-history state to build its current attention queue, but
-the gates builder does not issue a second history read or expose history items
-in its response.
-
-`loopx global-todos` reads status exactly once, inspects a bounded set of unique
-attention-queue goals, and builds at most one authoritative quota packet for
-each inspected goal. Status may consume the compact run-history projection,
-but the todos builder does not issue a second history or store read. It derives
-todo candidates only from the selected lane and structured quota/todo
-projections; it does not parse active-state prose.
+`loopx global-todos`, `loopx global-summary`, and `loopx global-gates` each read
+status exactly once and reuse that observation when collecting quota facts.
+Status internally consumes compact run-history state to build its attention
+queue. These manager builders do not fetch a second history projection or read
+the authority store directly; source reads inside the quota owner retain their
+existing semantics. Manager queries invoke the read-only quota builder, not a
+Turn admission, claim, scheduler mutation or quota spend.
+
+The per-goal quota-evaluation budget is:
+
+- With an explicit `--agent-id`, at most one packet for that exact identity.
+  The successful packet must confirm the requested identity; an unscoped or
+  default-lane packet cannot satisfy the filter.
+- Without an agent filter and without discovered agent candidates, one
+  unscoped packet.
+- Without an agent filter and with `A` distinct agent candidates, at most one
+  unscoped packet plus one scoped packet per candidate. With a discoverable
+  roster, collect its complete identities and existing lane recommendations;
+  repeated identities or queue entries must not multiply reads.
+
+This replaces the former one-packet-per-goal budget for unscoped goals with
+agent candidates. The current implementation evaluates the scoped packets
+sequentially; its quota-builder call count is `O(G + sum(A))` for `G` inspected
+unique goals. Individual evaluation cost also depends on goal/history size.
+The existing goal scan and result limits remain command-specific: a display
+limit does not cap the roster or the complete counts within an inspected goal.
+For `global-todos`, `goal_scan_limit` and `goal_scan_truncated` still describe
+the earlier goal scan. Projection-envelope goal counts describe coverage, not
+physical store reads or the number of scoped quota evaluations.
+
+Keep each agent's eligibility with the quota owner. The manager may aggregate
+its structured facts but must not infer eligibility from a title, a registry
+entry or the first healthy lane. A failed scoped read cannot establish complete
+coverage: `global-todos` retains healthy lanes with its redacted source warnings
+and degraded-goal count, or omits the goal if no lane can be verified. Summary
+and gate reads retain their own source-failure rules. Warning-list truncation
+must not erase the full warning count. No result authorizes the suggested work.
+
+The cost tradeoff was checked on the same synthetic nine-agent goal: runnable
+work belongs only to agent nine, while agents one and two each have a distinct
+user gate. Immutable base `f941b5939345c6d1c3d770a982182624e7ef393f`
+used one quota evaluation per command; the complete collection at
+`afb3b5c1787dad1293da1150d93da8c22afe729a` used ten, with one status call
+in both. The base omitted the ninth runnable entry from `global-todos` and
+showed only one gate; the complete collection retained the work and both gates.
+An isolated one-quota shortcut using the current gate consumer again lost one
+gate on File and SQLite. A future complete batch may reduce the read count;
+truncating candidates or returning the first healthy lane is not an equivalent
+optimization.
+
+The following are instrumented CLI-entrypoint wall-time medians in seconds over
+three invocations per cell on macOS, Python 3.14.8 and Node 24.21.0. The first todos invocation in each
+isolated runtime was cold; later calls reused that runtime. These small,
+contended samples explain the additional cost, not a latency limit or a speedup.
+
+| Provider | Command | One-packet base | Complete collection |
+| --- | --- | ---: | ---: |
+| File | global-todos | 1.154 | 1.611 |
+| File | global-summary | 0.344 | 1.513 |
+| File | global-gates | 0.319 | 1.563 |
+| SQLite | global-todos | 0.478 | 1.246 |
+| SQLite | global-summary | 0.412 | 1.172 |
+| SQLite | global-gates | 0.375 | 0.892 |
+
+Across those cells, Python caller CPU medians were 0.170–0.187 seconds for the
+base and 0.352–0.372 for complete collection; caller peak-RSS medians were
+46.4–46.8 MiB and 47.1–49.2 MiB. Wall time includes service waits, but Node
+CPU/RSS and fixture setup are outside those resource figures. Canonical Todo
+records/provider revisions and source/registry bytes remained unchanged.
+These five-Todo fixtures do not qualify large histories, hundred-agent scale,
+installed-host adoption or long-horizon efficiency. The original budget failure
+remains recorded; this revision changes the declared source budget, not a frozen
+experiment, runtime deadline, transport size limit or execution allowance.
 
 `loopx global-risks` reads status exactly once. It accepts structured contract
 diagnostics, global registry findings, attention-queue stale-run warnings, and,
@@ -263,10 +325,14 @@ the global limit.
 
 Global and per-goal failures have different envelopes. If the global status
 source is unhealthy, the command returns `ok=false` with a public-safe error
-and exits non-zero; it must not look like a successful empty inbox. If one
-goal's quota read raises or returns `ok=false`, the command skips that
-unverifiable goal, records one bounded redacted warning, and retains results
-from healthy goals.
+and exits non-zero; it must not look like a successful empty inbox. A known
+out-of-scope agent identity remains a filter miss. If a goal's other initial
+quota read raises or returns `ok=false`, the command skips that
+unverifiable goal and records a bounded redacted warning. During scoped
+collection, a failed agent read warns without hiding verified peers; if every
+scoped read fails, the goal is omitted. The full source-warning count and
+degraded-goal coverage remain visible, including when displayed warnings are
+capped. Healthy goals remain available.
 
 ### Focused Global Risks Response
 

loopx/control_plane/presentation/global_manager_read.ts

@@ -48,7 +48,15 @@ function laneFrom(item: JsonObject, quota: JsonObject): JsonObject {
   return result;
 }
 const waitAction = "Wait for the projected user/controller decision before running the blocked path.";
+function foreignTodoGateOnly(quota: JsonObject): boolean {
+  const scope = object(object(quota.user_todo_summary).agent_scope_filter);
+  // Scope counts describe Todo gates, not decisions from another or unknown source.
+  return quota.source === "active_state" && quota.status === "active_state_user_gate" && !truth(quota.gate_id)
+    && typeof scope.agent_id === "string" && scope.current_agent_blocking_open_count === 0
+    && typeof scope.other_agent_scoped_open_count === "number" && scope.other_agent_scoped_open_count > 0;
+}
 function summaryGate(item: JsonObject, quota: JsonObject): JsonObject | null {
+  if (foreignTodoGateOnly(quota)) return null;
   const channel = object(object(quota.interaction_contract).user_channel), summary = object(quota.user_todo_summary);
   if (!truth(channel.action_required) && Number(or(summary.open_count, summary.count, 0)) <= 0) return null;
   let first: JsonObject = {};
@@ -89,6 +97,9 @@ function blockedRefs(gate: JsonObject | null, quota: JsonObject, goal: string):
 }
 function globalGate(item: JsonObject, quota: JsonObject, facts: GateSourceFacts): JsonObject | null {
   if (!facts.eligible) return null;
+  // A peer notification can mention another peer's gate. Preserve the explicit
+  // scope counts rather than presenting that notification as an additional gate.
+  if (!facts.formal_gate && foreignTodoGateOnly(quota)) return null;
   const channel = object(object(quota.interaction_contract).user_channel), formal = facts.formal_gate;
   const action = channel.action_required === true, waiting = field(item, "waiting_on").trim();
   const owner = formal || action ? "user" : "controller", goal = str(or(item.goal_id, quota.goal_id, "")).trim();
@@ -116,6 +127,7 @@ function envelope(input: JsonObject, projection: string, evaluated: number, unav
     upstream: Object.keys(upstream).length ? [upstream] : []});
 }
 const queueCount = (status: JsonObject): number => new Set(rows(object(status.attention_queue).items).map(row => field(row, "goal_id").trim()).filter(Boolean)).size;
+const gateKey = (gate: JsonObject): string => JSON.stringify([gate.goal_id, gate.gate_id, gate.question, gate.blocks]);
 type GateSourceFacts = {formal_gate: JsonObject | null; eligible: boolean};
 function decodeGateSourceFacts(value: unknown): GateSourceFacts {
   const facts = requireJsonObject(value, "gate_source_facts");
@@ -145,15 +157,37 @@ export function projectGlobalSummary(input: JsonObject): JsonObject {
   const surfaces = requireStringArray(input.source_surfaces, "source_surfaces");
   if (input.since !== null) requireInteger(input.since, "since");
   const lanes: JsonObject[] = [], gates: JsonObject[] = [], todos: JsonObject[] = [], states: Record<string, number> = {};
-  for (const {item, quota} of samples) {
-    const goal = field(item, "goal_id").trim(), lane = laneFrom(item, quota), gate = summaryGate(item, quota), todo = firstTodo(quota);
+  const goals = new Map<string, typeof samples>();
+  for (const sample of samples) {
+    const goal = field(sample.item, "goal_id").trim(), group = goals.get(goal) ?? [];
+    group.push(sample); goals.set(goal, group);
+  }
+  for (const [goal, group] of goals) {
+    // Quota owns eligibility. Retain one goal lane while reading every peer's
+    // gates and work, including peers after a runnable recommendation.
+    const preferred = group.find(({quota}) => quota.normal_delivery_allowed === true)
+      ?? group.find(({quota}) => quota.ok !== false && Object.keys(object(quota.selected_todo)).length > 0)
+      ?? group.find(({quota}) => quota.ok !== false) ?? group[0]!;
+    const lane = laneFrom(preferred.item, preferred.quota);
     lanes.push(lane); states[String(lane.status ?? "unknown")] = (states[String(lane.status ?? "unknown")] ?? 0) + 1;
-    if (gate) gates.push(gate);
-    if (todo) todos.push({todo_id: todo.todo_id ?? null, goal_id: goal, role: or(todo.role, "agent"), claimed_by: todo.claimed_by ?? null,
-      state: or(todo.status, "open"), priority: todo.priority ?? null, title: display([todo, "title"], [todo, "text"]), next_safe_action: display([quota, "recommended_action"])});
+    const seenGates = new Set<string>(), seenTodos = new Set<string>();
+    for (const {item, quota, gate_source_facts} of group) {
+      const gate = gate_source_facts === undefined ? summaryGate(item, quota)
+        : globalGate(item, quota, decodeGateSourceFacts(gate_source_facts));
+      const todo = firstTodo(quota);
+      if (gate && !seenGates.has(gateKey(gate))) {seenGates.add(gateKey(gate)); gates.push(gate);}
+      if (todo) {
+        const key = str(or(todo.todo_id, todo.title, ""));
+        if (seenTodos.has(key)) continue;
+        seenTodos.add(key);
+        todos.push({todo_id: todo.todo_id ?? null, goal_id: goal, role: or(todo.role, "agent"), claimed_by: todo.claimed_by ?? null,
+          state: or(todo.status, "open"), priority: todo.priority ?? null, title: display([todo, "title"], [todo, "text"]), next_safe_action: display([quota, "recommended_action"])});
+      }
+    }
   }
   lanes.sort((a, b) => (LANE_PRIORITY[String(a.status)] ?? 4) - (LANE_PRIORITY[String(b.status)] ?? 4) || authorityUnicodeCompare(field(a, "goal_id"), field(b, "goal_id")));
-  const sealed = envelope(input, "global_summary", samples.length, samples.filter(({quota}) => quota.state === "quota_unavailable").length,
+  const unavailable = [...goals.values()].filter(group => group.some(({quota}) => quota.state === "quota_unavailable")).length;
+  const sealed = envelope(input, "global_summary", goals.size, unavailable,
     Math.min(lanes.length, limit), queueCount(status));
   const risks = rows(object(status.global_registry).findings).map(row => ({kind: str(or(row.kind, "status_warning")), severity: str(or(row.severity, "info")),
     goal_id: row.goal_id ?? null, evidence_refs: ["status_contract"], next_safe_action: display([row, "recommended_action"], [row, "message"])}));
@@ -193,20 +227,24 @@ export function projectGlobalGates(input: JsonObject): JsonObject {
   if (status.ok !== true) return {ok: false, schema_version: SCHEMA_VERSION, request, generated_at: input.generated_at,
     error: "Global status source unavailable.", omissions: ["Raw/private failure details and local paths were intentionally omitted."], boundary: input.boundary,
     projection_envelope: envelope(input, "global_gates", 0, 0, 0, 0)};
-  const gates: JsonObject[] = [], lanes: JsonObject[] = [];
+  const gates: JsonObject[] = [], lanes: JsonObject[] = [], seenGates = new Set<string>();
   const agent = input.agent_id == null ? null : requireNonEmptyString(input.agent_id, "agent_id");
   for (const {item, quota, facts} of gateSamples) {
     if (agent && (quota.ok === false || field(object(quota.agent_identity), "agent_id") !== agent)) continue;
     const gate = globalGate(item, quota, facts);
     if (!gate) continue;
+    if (seenGates.has(gateKey(gate))) continue;
+    seenGates.add(gateKey(gate));
     gates.push(gate);
     const lane = laneFrom(item, quota), goal = field(item, "goal_id").trim();
     const verified = list(gate.blocks).map(str).filter(ref => ref !== goal);
     if (!verified.includes(field(lane, "top_todo_id"))) delete lane.top_todo_id;
     lanes.push(lane);
     if (gates.length >= limit) break;
   }
-  const sealed = envelope(input, "global_gates", samples.length, samples.filter(({quota}) => quota.state === "quota_unavailable").length, samples.length, queueCount(status));
+  const evaluated = new Set(samples.map(({item}) => field(item, "goal_id").trim())).size;
+  const unavailable = new Set(samples.filter(({quota}) => quota.state === "quota_unavailable").map(({item}) => field(item, "goal_id").trim())).size;
+  const sealed = envelope(input, "global_gates", evaluated, unavailable, evaluated, queueCount(status));
   return {ok: true, schema_version: SCHEMA_VERSION, request, generated_at: input.generated_at,
     summary: {headline: `${gates.length} open user/controller ${gates.length === 1 ? "gate requires" : "gates require"} a decision.`,
       open_gate_count: gates.length, blocked_lane_count: lanes.length, source_surfaces: GLOBAL_GATES_SOURCE_SURFACES},