diff --git a/docs/reference/protocols/global-manager-command-v0.md b/docs/reference/protocols/global-manager-command-v0.md index bbb492cfc3..5c88889837 100644 --- a/docs/reference/protocols/global-manager-command-v0.md +++ b/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 diff --git a/loopx/control_plane/presentation/global_manager_read.ts b/loopx/control_plane/presentation/global_manager_read.ts index d05ce877d2..e89943c5b0 100644 --- a/loopx/control_plane/presentation/global_manager_read.ts +++ b/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 = {}; - 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(); + 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(), seenTodos = new Set(); + 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,12 +227,14 @@ 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(); 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); @@ -206,7 +242,9 @@ export function projectGlobalGates(input: JsonObject): JsonObject { 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}, diff --git a/loopx/global_todos.py b/loopx/global_todos.py index 5f6bbb6792..29e577f60f 100644 --- a/loopx/global_todos.py +++ b/loopx/global_todos.py @@ -9,6 +9,10 @@ seal_projection_envelope, source_fact, ) +from .control_plane.quota.error_codes import ( + QuotaIdentityPrecondition, + QuotaIdentityPreconditionError, +) from .control_plane.runtime.time import now_utc_iso from .control_plane.todos.decision_scope import todo_gate_relations from .control_plane.todos.user_gate import open_user_gate_todo_items @@ -114,13 +118,113 @@ def _warning( return warning -def _quota_for_goal( +def is_unregistered_agent_scope_miss( + payload: dict[str, Any] | BaseException, + *, + agent_id: str | None, +) -> bool: + if not agent_id: + return False + if isinstance(payload, QuotaIdentityPreconditionError): + return payload.precondition in { + QuotaIdentityPrecondition.REQUESTED_AGENT_REGISTERED, + QuotaIdentityPrecondition.REGISTERED_AGENT_ROSTER_PRESENT, + } + if not isinstance(payload, dict) or payload.get("ok") is not False: + return False + if payload.get("error_code") in { + "quota_agent_not_registered", + "quota_agent_registry_roster_missing", + }: + return True + detail = str(payload.get("error") or payload.get("reason") or "") + expected_messages = { + str( + QuotaIdentityPreconditionError( + QuotaIdentityPrecondition.REQUESTED_AGENT_REGISTERED, + agent_id=agent_id, + ) + ), + str( + QuotaIdentityPreconditionError( + QuotaIdentityPrecondition.REGISTERED_AGENT_ROSTER_PRESENT, + agent_id=agent_id, + ) + ), + } + return detail in expected_messages + + +def registered_goal_agents( + status_payload: dict[str, Any], + goal_id: str, + quota_payload: dict[str, Any] | None = None, + *, + prefer_queue_recommendation: bool = False, +) -> list[str]: + registered: list[str] = [] + seen_registered: set[str] = set() + + def add_registered(value: object) -> None: + candidate = str(value or "").strip() + if candidate and candidate not in seen_registered: + seen_registered.add(candidate) + registered.append(candidate) + + run_history = as_dict(status_payload.get("run_history")) + for goal in as_list(run_history.get("goals")): + if not isinstance(goal, dict) or str(goal.get("id") or "").strip() != goal_id: + continue + coordination = as_dict(goal.get("coordination")) + for candidate in as_list(coordination.get("registered_agents")): + add_registered(candidate) + + if isinstance(quota_payload, dict): + for container_key in ("automation_prompt_upgrade", "agent_identity"): + container = as_dict(quota_payload.get(container_key)) + for candidate in as_list(container.get("registered_agents")): + add_registered(candidate) + + if not registered: + return [] + + agents: list[str] = [] + seen: set[str] = set() + + def add_agent(value: object) -> None: + candidate = str(value or "").strip() + if candidate and candidate not in seen: + seen.add(candidate) + agents.append(candidate) + + queue = as_dict(status_payload.get("attention_queue")) + if prefer_queue_recommendation: + for item in as_list(queue.get("items")): + if not isinstance(item, dict) or str(item.get("goal_id") or "").strip() != goal_id: + continue + add_agent(as_dict(item.get("agent_lane_recommendation")).get("agent_id")) + add_agent(as_dict(as_dict(item.get("project_asset")).get("owner")).get("agent_id")) + + for candidate in registered: + add_agent(candidate) + + if not prefer_queue_recommendation: + for item in as_list(queue.get("items")): + if not isinstance(item, dict) or str(item.get("goal_id") or "").strip() != goal_id: + continue + add_agent(as_dict(item.get("agent_lane_recommendation")).get("agent_id")) + add_agent(as_dict(as_dict(item.get("project_asset")).get("owner")).get("agent_id")) + + return agents + + +def _quotas_for_goal( status_payload: dict[str, Any], *, goal_id: str, agent_id: str | None, warnings: list[dict[str, Any]], -) -> dict[str, Any] | None: +) -> tuple[list[dict[str, Any]], bool]: try: payload = build_quota_should_run( status_payload, @@ -128,6 +232,8 @@ def _quota_for_goal( agent_id=agent_id, ) except Exception as exc: + if is_unregistered_agent_scope_miss(exc, agent_id=agent_id): + return [], False warnings.append( _warning( "quota_unavailable", @@ -135,8 +241,10 @@ def _quota_for_goal( detail=exc, ) ) - return None + return [], True if payload.get("ok") is False: + if is_unregistered_agent_scope_miss(payload, agent_id=agent_id): + return [], False warnings.append( _warning( "quota_unavailable", @@ -144,8 +252,51 @@ def _quota_for_goal( detail=payload.get("error") or payload.get("reason"), ) ) - return None - return payload + return [], True + if agent_id is None: + registered_agents = registered_goal_agents( + status_payload, + goal_id, + payload, + ) + if registered_agents: + scoped_payloads: list[dict[str, Any]] = [] + scoped_failed = False + for candidate_agent_id in registered_agents: + try: + scoped = build_quota_should_run( + status_payload, + goal_id=goal_id, + agent_id=candidate_agent_id, + ) + except Exception as exc: + scoped_failed = True + warnings.append( + _warning( + "quota_unavailable", + goal_id=goal_id, + detail=exc, + ) + ) + continue + if not isinstance(scoped, dict) or scoped.get("ok") is not True: + scoped_failed = True + warnings.append( + _warning( + "quota_unavailable", + goal_id=goal_id, + detail=(scoped.get("error") or scoped.get("reason")) + if isinstance(scoped, dict) + else "scoped quota payload unavailable", + ) + ) + continue + scoped_payloads.append(scoped) + if scoped_payloads: + return scoped_payloads, scoped_failed + if scoped_failed: + return [], True + return [payload], False def _quota_matches_agent(payload: dict[str, Any], *, agent_id: str | None) -> bool: @@ -560,25 +711,47 @@ def build_global_todos( quota_read_at = now_utc_iso() quota_unavailable = 0 for goal_id in goal_ids: - quota_payload = _quota_for_goal( + quota_payloads, was_unavailable = _quotas_for_goal( status_payload, goal_id=goal_id, agent_id=agent_id, warnings=warnings, ) - quota_unavailable += int(quota_payload is None) - if quota_payload is None or not _quota_matches_agent( - quota_payload, - agent_id=agent_id, - ): + quota_unavailable += int(was_unavailable) + if not quota_payloads: continue - classified.extend( - _classify_goal_todos( + recommended_action_by_agent = { + str(as_dict(q.get("agent_identity")).get("agent_id") or "").strip(): q.get( + "recommended_action" + ) + for q in quota_payloads + if str(as_dict(q.get("agent_identity")).get("agent_id") or "").strip() + } + goal_classified: list[dict[str, Any]] = [] + for quota_payload in quota_payloads: + if not _quota_matches_agent( quota_payload, - goal_id=goal_id, - warnings=warnings, + agent_id=agent_id, + ): + continue + goal_classified.extend( + _classify_goal_todos( + quota_payload, + goal_id=goal_id, + warnings=warnings, + ) ) - ) + if len(quota_payloads) > 1 and recommended_action_by_agent: + for todo in goal_classified: + owner = str(todo.get("claimed_by") or "").strip() + if ( + todo.get("readiness") != "runnable" + and owner in recommended_action_by_agent + ): + todo["next_safe_action"] = _redact_text( + recommended_action_by_agent[owner] + ) + classified.extend(goal_classified) matched = _deduplicate_todos(classified, warnings=warnings) matched.sort(key=_todo_sort_key) diff --git a/loopx/summary_all.py b/loopx/summary_all.py index 1b2aad80f5..5564cc5648 100644 --- a/loopx/summary_all.py +++ b/loopx/summary_all.py @@ -10,6 +10,7 @@ ) from .control_plane.runtime.time import now_utc, now_utc_iso from .control_plane.todos.user_gate import open_user_gate_todo_items +from .global_todos import is_unregistered_agent_scope_miss, registered_goal_agents from .presentation.markdown import as_dict as _as_dict from .presentation.markdown import as_list as _as_list from .presentation.public_safety import public_safe_boundary, redact_public_text @@ -100,6 +101,24 @@ def _quota_matches_agent_scope( return str(identity.get("agent_id") or "") == agent_id +def _build_goal_quotas( + status_payload: dict[str, Any], *, goal_id: str, agent_id: str | None, +) -> tuple[list[dict[str, Any]], bool]: + """Collect complete scoped facts; the TS projection selects the visible lane.""" + payload = _build_goal_quota(status_payload, goal_id=goal_id, agent_id=agent_id) + if is_unregistered_agent_scope_miss(payload, agent_id=agent_id): + return [], False + if agent_id is not None or payload.get("ok") is False: + return [payload], False + agents = registered_goal_agents( + status_payload, goal_id, payload, prefer_queue_recommendation=True, + ) + if not agents: + return [payload], False + return [_build_goal_quota(status_payload, goal_id=goal_id, agent_id=agent) + for agent in agents], True + + def _global_gates_request() -> dict[str, Any]: return { "schema_version": "global_manager_command_request_v0", @@ -259,18 +278,22 @@ def _read_samples(status: dict[str, Any], *, agent_id: str | None, if not goal_id or goal_id in seen: continue seen.add(goal_id) - quota = _build_goal_quota(status, goal_id=goal_id, agent_id=agent_id) - sample = {"item": _read_facts(item), "quota": _read_facts(quota)} - facts = _gate_source_facts(item, quota, agent_id) if gates_only else None - if facts is not None: - sample["gate_source_facts"] = _read_facts(facts) - samples.append(sample) - if not gates_only or not _quota_matches_agent_scope(quota, agent_id=agent_id): - continue - if facts is not None and facts["eligible"]: - found += 1 - if found >= limit: - break + quotas, scoped_reads = _build_goal_quotas(status, goal_id=goal_id, agent_id=agent_id) + for quota in quotas: + sample = {"item": _read_facts(item), "quota": _read_facts(quota)} + scoped_agent = (_as_dict(quota.get("agent_identity")).get("agent_id") + if scoped_reads else agent_id) + facts = (_gate_source_facts(item, quota, scoped_agent) + if gates_only or scoped_reads else None) + if facts is not None: + sample["gate_source_facts"] = _read_facts(facts) + samples.append(sample) + if not gates_only or not _quota_matches_agent_scope(quota, agent_id=agent_id): + continue + if facts is not None and facts["eligible"]: + found += 1 + if gates_only and found >= limit: + break return samples diff --git a/tests/control_plane_ts/global_manager_read.test.ts b/tests/control_plane_ts/global_manager_read.test.ts index f8a592350f..2e5cdff9fb 100644 --- a/tests/control_plane_ts/global_manager_read.test.ts +++ b/tests/control_plane_ts/global_manager_read.test.ts @@ -57,3 +57,99 @@ test("gate projection consumes source eligibility without reclassifying quota", samples[0].gate_source_facts.eligible = true; assert.equal((projectGlobalGates({...request(), samples}).gates as JsonObject[])[0].owner, "controller"); }); + +test("one goal retains its runnable peer, late gate and unique source counts in either order", () => { + const samples = Array.from({length: 9}, (_, index) => { + const agent = `agent-${index + 1}`, ready = index === 0, gated = index === 8; + const gate = {todo_id: "todo_gate", task_class: "user_gate", text: "Approve synthetic output?", + unblocks_todo_id: "todo_held", _display: {text: "Approve synthetic output?"}}; + const quota: JsonObject = {ok: true, goal_id: "synthetic", state: gated ? "operator_gate" : "eligible", + normal_delivery_allowed: ready, agent_identity: {agent_id: agent}, + ...(ready ? {selected_todo: {todo_id: "todo_ready"}, agent_lane_next_action: {todo_id: "todo_ready", agent_id: agent}} : {}), + ...(gated ? {gate_id: "todo_gate", selected_todo: {todo_id: "todo_held"}, + agent_lane_next_action: {todo_id: "todo_held", agent_id: agent}, + user_todo_summary: {open_count: 1, first_open_items: [gate], gate_open_items: [gate]}} : {})}; + return {item: {goal_id: "synthetic"}, quota, gate_source_facts: {formal_gate: gated ? gate : null, eligible: gated}}; + }); + for (const ordered of [samples, [...samples].reverse()]) { + const input = {...request(), samples: [...ordered, samples[8]]}, before = structuredClone(input); + const summary = projectGlobalSummary(input), gates = projectGlobalGates(input); + assert.deepEqual((summary.lanes as JsonObject[]).map(row => [row.agent_id, row.top_todo_id]), [["agent-1", "todo_ready"]]); + assert.equal((summary.summary as JsonObject).open_gate_count, 1); + assert.equal((gates.summary as JsonObject).open_gate_count, 1); + assert.equal((summary.todos as JsonObject[]).length, 2); + assert.deepEqual((gates.gates as JsonObject[])[0].blocks, ["todo_held"]); + for (const payload of [summary, gates]) { + const sources = (payload.projection_envelope as JsonObject).sources as JsonObject[]; + assert.equal(sources.find(row => row.source_id === "goal_quota")!.item_count, 1); + } + assert.deepEqual(input, before); + } +}); + +test("an unavailable peer stays visible as source degradation without hiding healthy work", () => { + const input = {...request(), samples: [ + {item: {goal_id: "synthetic"}, quota: {ok: false, state: "quota_unavailable"}}, + {item: {goal_id: "synthetic"}, quota: {ok: true, state: "normal_run", normal_delivery_allowed: true, + agent_identity: {agent_id: "healthy-peer"}, agent_lane_next_action: {todo_id: "todo_ready"}}}, + ]}; + const payload = projectGlobalSummary(input); + assert.equal((payload.lanes as JsonObject[])[0].agent_id, "healthy-peer"); + const envelope = payload.projection_envelope as JsonObject; + assert.equal(envelope.alert, true); + assert.equal((envelope.sources as JsonObject[]).find(row => row.source_id === "goal_quota")!.unreadable_count, 1); +}); + +test("foreign-only Todo notifications do not duplicate gates; applicable and unknown scope survive", () => { + for (const [scope, expected] of [ + [{agent_id: "peer-a", current_agent_blocking_open_count: 0, other_agent_scoped_open_count: 1}, 0], + [{agent_id: "peer-a", current_agent_blocking_open_count: 1, other_agent_scoped_open_count: 1}, 1], + [{agent_id: "peer-a", other_agent_scoped_open_count: 1}, 1], + ] as const) { + const quota = {source: "active_state", status: "active_state_user_gate", + interaction_contract: {user_channel: {action_required: true}}, + user_todo_summary: {agent_scope_filter: scope}}; + const sample = {item: {goal_id: "synthetic"}, quota, gate_source_facts: {formal_gate: null, eligible: true}}; + const input = {...request(), samples: [sample]}; + assert.equal((projectGlobalSummary(input).gates as JsonObject[]).length, expected); + assert.equal((projectGlobalGates(input).gates as JsonObject[]).length, expected); + } +}); + +for (const [name, origin] of [ + ["unknown source", {}], + ["ordinary user Todo", {source: "active_state", status: "active_state_user_todo"}], + ["another source", {source: "latest_run", status: "active_state_user_gate"}], + ["explicit decision", {source: "active_state", status: "active_state_user_gate", gate_id: "distinct-decision"}], +] as const) { + test(`foreign Todo counts preserve a required decision from ${name}`, () => { + const question = "Confirm independent recovery?"; + const quota = {...origin, interaction_contract: {user_channel: { + action_required: true, question, _display: {question}, + }}, user_todo_summary: {agent_scope_filter: { + agent_id: "peer-a", current_agent_blocking_open_count: 0, other_agent_scoped_open_count: 1, + }}}; + const sample = {item: {goal_id: "synthetic"}, quota}, before = structuredClone(sample); + const input = {...request(), samples: [{...sample, gate_source_facts: {formal_gate: null, eligible: true}}]}; + for (const result of [projectGlobalSummary(input), projectGlobalGates(input), + projectGlobalSummary({...request(), samples: [sample]})]) { + const gates = result.gates as JsonObject[]; + assert.equal(gates.length, 1); + assert.equal(gates[0].question, question); + } + assert.deepEqual(sample, before); + }); +} + +test("a formal gate survives foreign Todo scope counts", () => { + const gate = {todo_id: "todo_current_gate", task_class: "user_gate", unblocks_todo_id: "todo_current_work"}; + const sample = {item: {goal_id: "synthetic"}, gate_source_facts: {formal_gate: gate, eligible: true}, + quota: {source: "active_state", status: "active_state_user_gate", user_todo_summary: {agent_scope_filter: { + agent_id: "peer-a", current_agent_blocking_open_count: 0, other_agent_scoped_open_count: 1, + }}}}; + for (const project of [projectGlobalSummary, projectGlobalGates]) { + const gates = project({...request(), samples: [sample]}).gates as JsonObject[]; + assert.equal(gates.length, 1); + assert.deepEqual(gates[0].blocks, ["todo_current_work"]); + } +}); diff --git a/tests/test_global_todos.py b/tests/test_global_todos.py index fe4555c2f0..1f2b578a63 100644 --- a/tests/test_global_todos.py +++ b/tests/test_global_todos.py @@ -620,3 +620,238 @@ def test_global_todos_error_markdown_stays_public_safe(tmp_path) -> None: assert "Global Todos Unavailable" in markdown assert str(tmp_path) not in markdown assert "" in markdown + + +def test_unscoped_peer_v1_goal_fans_out_across_registered_agents_and_matches_owner_actions( + monkeypatch, + tmp_path, +) -> None: + status = { + "ok": True, + "attention_queue": { + "items": [{"goal_id": "goal-peer"}], + }, + "run_history": { + "goals": [ + { + "id": "goal-peer", + "coordination": { + "registered_agents": ["review-a", "product-engineer"], + }, + } + ] + }, + } + deferred_item = { + "todo_id": "todo-deferred-product", + "status": "deferred", + "resume_ready": True, + "priority": "P1", + "text": "Deferred work for product-engineer", + "claimed_by": "product-engineer", + "action_kind": "implement", + } + unscoped_quota = { + "ok": True, + "goal_id": "goal-peer", + "should_run": False, + "normal_delivery_allowed": False, + "reason": ( + "coordination.registered_agents is configured for peer_v1, " + "but quota should-run was called without --agent-id" + ), + "recommended_action": ( + "Regenerate each installed heartbeat with its registered --agent-id, " + "then rerun quota should-run with the same identity." + ), + "heartbeat_recommendation": { + "recommended_mode": "automation_prompt_upgrade", + }, + "agent_todo_summary": { + "deferred_resume_candidates": [deferred_item], + }, + } + review_quota = runnable_quota( + "goal-peer", + "todo-runnable-review", + priority="P1", + action_kind="review_pr", + agent_id="review-a", + ) + review_quota["recommended_action"] = "Review exact PR heads." + review_quota["agent_todo_summary"] = { + "deferred_resume_candidates": [deferred_item], + } + product_quota = { + "ok": True, + "goal_id": "goal-peer", + "normal_delivery_allowed": False, + "agent_identity": {"agent_id": "product-engineer", "registered": True}, + "recommended_action": "Resume product-engineer deferred slice.", + "agent_todo_summary": { + "deferred_resume_candidates": [deferred_item], + }, + } + + def build_quota(status_payload, *, goal_id, agent_id=None): + assert status_payload is status + assert goal_id == "goal-peer" + if agent_id is None: + return unscoped_quota + if agent_id == "review-a": + return review_quota + if agent_id == "product-engineer": + return product_quota + raise AssertionError(f"unexpected agent_id {agent_id}") + + monkeypatch.setattr(global_todos, "collect_status", lambda **_: status) + monkeypatch.setattr(global_todos, "build_quota_should_run", build_quota) + + payload = build_payload(tmp_path) + + assert payload["summary"]["runnable_count"] == 1 + assert payload["summary"]["deferred_ready_count"] == 1 + by_id = {todo["todo_id"]: todo for todo in payload["todos"]} + assert by_id["todo-runnable-review"]["readiness"] == "runnable" + assert ( + by_id["todo-runnable-review"]["next_safe_action"] + == "Review exact PR heads." + ) + assert by_id["todo-deferred-product"]["readiness"] == "deferred_ready" + assert ( + by_id["todo-deferred-product"]["next_safe_action"] + == "Resume product-engineer deferred slice." + ) + + +def test_agent_scope_skips_goals_where_agent_is_not_registered_without_warning( + monkeypatch, + tmp_path, +) -> None: + patch_manager_reads( + monkeypatch, + status=status_for("goal-unregistered", "goal-registered"), + quotas={ + "goal-unregistered": { + "ok": False, + "error": "agent_id='product-engineer' is not registered in the selected registry goal", + }, + "goal-registered": runnable_quota( + "goal-registered", + "todo-registered", + agent_id="product-engineer", + ), + }, + ) + + payload = build_payload(tmp_path, agent_id="product-engineer") + + assert [todo["todo_id"] for todo in payload["todos"]] == ["todo-registered"] + assert payload["source_warnings"] == [] + + +def test_unscoped_peer_v1_goal_fans_out_past_eighth_registered_agent_and_records_lane_failures( + monkeypatch, + tmp_path, +) -> None: + agents = [f"agent-{idx}" for idx in range(1, 10)] + status = { + "ok": True, + "attention_queue": { + "items": [{"goal_id": "goal-nine-agents"}], + }, + "run_history": { + "goals": [ + { + "id": "goal-nine-agents", + "coordination": { + "registered_agents": agents, + }, + } + ] + }, + } + unscoped_quota = { + "ok": True, + "goal_id": "goal-nine-agents", + "should_run": False, + "normal_delivery_allowed": False, + } + ninth_quota = runnable_quota( + "goal-nine-agents", + "todo-agent-9", + priority="P0", + action_kind="implement", + agent_id="agent-9", + ) + + def build_quota(_status_payload, *, goal_id, agent_id=None): + assert goal_id == "goal-nine-agents" + if agent_id is None: + return unscoped_quota + if agent_id == "agent-9": + return ninth_quota + if agent_id == "agent-2": + return {"ok": False, "error": "lane-2 state corrupted"} + return { + "ok": True, + "goal_id": "goal-nine-agents", + "normal_delivery_allowed": False, + "agent_identity": {"agent_id": agent_id, "registered": True}, + } + + monkeypatch.setattr(global_todos, "collect_status", lambda **_: status) + monkeypatch.setattr(global_todos, "build_quota_should_run", build_quota) + + payload = build_payload(tmp_path) + + assert [todo["todo_id"] for todo in payload["todos"]] == ["todo-agent-9"] + assert payload["summary"]["runnable_count"] == 1 + assert len(payload["source_warnings"]) == 1 + assert payload["source_warnings"][0]["reason_code"] == "quota_unavailable" + assert ( + payload["projection_envelope"]["sources"][1]["unreadable_count"] == 1 + ) + + +def test_peer_source_budget_preserves_complete_counts_before_display_limit( + monkeypatch, tmp_path, +) -> None: + agents = [f"agent-{index}" for index in range(1, 10)] + goal = "source-budget-goal" + status = {"ok": True, "attention_queue": {"items": [{"goal_id": goal}] * 2}, + "run_history": {"goals": [{"id": goal, "coordination": { + "registered_agents": [*agents, agents[0]], + }}]}} + reads = {"status": 0, "quota": []} + + def collect_status(**_kwargs): + reads["status"] += 1 + return status + + def build_quota(source, *, goal_id, agent_id=None): + assert source is status and goal_id == goal + reads["quota"].append(agent_id) + if agent_id is None: + return {"ok": True, "goal_id": goal, "normal_delivery_allowed": False} + return runnable_quota(goal, f"todo_{agent_id}", agent_id=agent_id) + + monkeypatch.setattr(global_todos, "collect_status", collect_status) + monkeypatch.setattr(global_todos, "build_quota_should_run", build_quota) + for limit in (1, 100): + reads["status"], reads["quota"] = 0, [] + result = build_payload(tmp_path, limit=limit) + assert result["summary"]["matched_todo_count"] == len(agents) + assert result["summary"]["runnable_count"] == len(agents) + assert len(result["todos"]) == min(limit, len(agents)) + assert reads["status"] == 1 + # The public budget is a ceiling: a future complete batch may use + # fewer packets. Duplicate queue/roster entries cannot multiply reads. + assert len(reads["quota"]) <= 1 + len(agents) + scoped = [agent for agent in reads["quota"] if agent is not None] + assert len(scoped) == len(set(scoped)) + + reads["status"], reads["quota"] = 0, [] + result = build_payload(tmp_path, agent_id=agents[-1], limit=1) + assert [row["todo_id"] for row in result["todos"]] == [f"todo_{agents[-1]}"] + assert reads == {"status": 1, "quota": [agents[-1]]} diff --git a/tests/test_summary_all.py b/tests/test_summary_all.py index 85b84a9f38..fbfd73e7e7 100644 --- a/tests/test_summary_all.py +++ b/tests/test_summary_all.py @@ -1,6 +1,19 @@ from __future__ import annotations +import json +import os +from pathlib import Path +import subprocess +import sys + +import pytest +from control_plane.canonical_authority_fixture import ( + initialize_canonical_source_fixture, + isolate_sqlite_runtime, +) + import loopx.summary_all as summary_all +from loopx.paths import global_registry_path def patch_manager_reads( @@ -616,3 +629,356 @@ def test_global_gates_error_redacts_paths_and_renderer_stays_focused(tmp_path) - assert "Recent Progress" not in markdown assert "Risks" not in markdown assert "## Lanes" not in markdown + + +def test_summary_all_resolves_registered_agent_for_unscoped_peer_v1_goal( + monkeypatch, + tmp_path, +) -> None: + status_payload = { + "ok": True, + "attention_queue": { + "items": [ + { + "goal_id": "peer-goal", + "status": "eligible", + "waiting_on": "codex", + "agent_lane_recommendation": { + "agent_id": "review-b", + "recommended_action": "Continue review-b queue.", + }, + } + ] + }, + "global_registry": {"findings": []}, + "run_history": { + "goals": [ + { + "id": "peer-goal", + "coordination": { + "registered_agents": ["review-a", "review-b"], + }, + } + ], + "recent_runs": [], + }, + } + unscoped_quota = { + "ok": True, + "goal_id": "peer-goal", + "state": "eligible", + "should_run": False, + "normal_delivery_allowed": False, + "reason": ( + "coordination.registered_agents is configured for peer_v1, " + "but quota should-run was called without --agent-id" + ), + "recommended_action": ( + "Regenerate each installed heartbeat with its registered --agent-id, " + "then rerun quota should-run with the same identity." + ), + "heartbeat_recommendation": { + "recommended_mode": "automation_prompt_upgrade", + }, + } + scoped_quota = { + "ok": True, + "goal_id": "peer-goal", + "state": "eligible", + "should_run": True, + "normal_delivery_allowed": True, + "agent_identity": {"agent_id": "review-b", "registered": True}, + "selected_todo": { + "todo_id": "todo-review-b", + "status": "open", + "priority": "P1", + "text": "Review exact PR queue.", + }, + "agent_lane_next_action": { + "todo_id": "todo-review-b", + "status": "open", + "agent_id": "review-b", + "text": "Review exact PR queue.", + }, + "recommended_action": "Review exact PR queue.", + } + + def build_quota(_status, *, goal_id, agent_id=None): + assert goal_id == "peer-goal" + return unscoped_quota if agent_id is None else scoped_quota + + monkeypatch.setattr(summary_all, "collect_status", lambda **_: status_payload) + monkeypatch.setattr(summary_all, "build_quota_should_run", build_quota) + + payload = summary_all.build_summary_all( + registry_path=tmp_path / "registry.json", + runtime_root_override=None, + agent_id=None, + time_range="24h", + limit=5, + ) + + assert len(payload["lanes"]) == 1 + assert payload["lanes"][0]["agent_id"] == "review-b" + assert payload["lanes"][0]["top_todo_id"] == "todo-review-b" + assert payload["lanes"][0]["next_safe_action"] == "Review exact PR queue." + + +def test_summary_all_and_global_gates_preserve_scoped_user_gate_alongside_runnable_peer_lane( + monkeypatch, + tmp_path, +) -> None: + agents = [f"agent-{idx}" for idx in range(1, 10)] + status_payload = { + "ok": True, + "attention_queue": { + "items": [ + { + "goal_id": "peer-mixed-goal", + "status": "eligible", + "waiting_on": "user", + "agent_lane_recommendation": { + "agent_id": "agent-1", + "recommended_action": "Run agent-1 delivery.", + }, + } + ] + }, + "global_registry": {"findings": []}, + "run_history": { + "goals": [ + { + "id": "peer-mixed-goal", + "coordination": { + "registered_agents": agents, + }, + } + ], + "recent_runs": [], + }, + } + unscoped_quota = { + "ok": True, + "goal_id": "peer-mixed-goal", + "state": "eligible", + "should_run": False, + "normal_delivery_allowed": False, + } + runnable_lane_quota = { + "ok": True, + "goal_id": "peer-mixed-goal", + "state": "eligible", + "should_run": True, + "normal_delivery_allowed": True, + "agent_identity": {"agent_id": "agent-1", "registered": True}, + "selected_todo": { + "todo_id": "todo-agent-1", + "status": "open", + "priority": "P1", + "text": "Run agent-1 delivery.", + }, + "agent_lane_next_action": { + "todo_id": "todo-agent-1", + "status": "open", + "agent_id": "agent-1", + "text": "Run agent-1 delivery.", + }, + "recommended_action": "Run agent-1 delivery.", + } + gated_ninth_quota = { + "ok": True, + "goal_id": "peer-mixed-goal", + "state": "operator_gate", + "should_run": False, + "normal_delivery_allowed": False, + "agent_identity": {"agent_id": "agent-9", "registered": True}, + "gate_id": "gate-agent-9", + "user_todo_summary": { + "open_count": 1, + "first_open_items": [ + { + "todo_id": "user-gate-9", + "task_class": "user_gate", + "text": "Approve agent-9 schema migration?", + } + ], + "gate_open_items": [ + { + "todo_id": "user-gate-9", + "task_class": "user_gate", + "text": "Approve agent-9 schema migration?", + } + ], + }, + "interaction_contract": { + "user_channel": { + "action_required": True, + "question": "Approve agent-9 schema migration?", + } + }, + "recommended_action": "Wait for user approval on agent-9.", + } + + def build_quota(_status, *, goal_id, agent_id=None): + assert goal_id == "peer-mixed-goal" + if agent_id is None: + return unscoped_quota + if agent_id == "agent-1": + return runnable_lane_quota + if agent_id == "agent-9": + return gated_ninth_quota + return { + "ok": True, + "goal_id": "peer-mixed-goal", + "state": "eligible", + "should_run": False, + "normal_delivery_allowed": False, + "agent_identity": {"agent_id": agent_id, "registered": True}, + } + + monkeypatch.setattr(summary_all, "collect_status", lambda **_: status_payload) + monkeypatch.setattr(summary_all, "build_quota_should_run", build_quota) + + summary_payload = summary_all.build_summary_all( + registry_path=tmp_path / "registry.json", + runtime_root_override=None, + agent_id=None, + time_range="24h", + limit=5, + ) + gates_payload = build_global_gates(tmp_path) + + assert len(summary_payload["lanes"]) == 1 + assert summary_payload["lanes"][0]["agent_id"] == "agent-1" + assert summary_payload["lanes"][0]["top_todo_id"] == "todo-agent-1" + assert len(summary_payload["gates"]) == 1 + assert summary_payload["gates"][0]["gate_id"] == "gate-agent-9" + assert ( + summary_payload["gates"][0]["question"] + == "Approve agent-9 schema migration?" + ) + assert len(gates_payload["gates"]) == 1 + assert gates_payload["gates"][0]["gate_id"] == "gate-agent-9" + + +@pytest.mark.parametrize("origin", [ + {}, + {"source": "active_state", "status": "active_state_user_gate", "gate_id": "distinct-decision"}, +]) +@pytest.mark.parametrize("command", ["global-summary", "global-gates"]) +def test_foreign_todo_counts_preserve_an_independent_decision_through_native_projection( + monkeypatch, tmp_path, origin, command, +): + isolate_sqlite_runtime(tmp_path, monkeypatch) + question = "Confirm independent recovery?" + status = {"ok": True, "attention_queue": {"items": [ + {"goal_id": "synthetic", "waiting_on": "controller"}, + ]}, "global_registry": {"findings": []}, "run_history": {"recent_runs": []}} + quota = {**origin, "ok": True, "agent_identity": {"agent_id": "peer-a"}, + "interaction_contract": {"user_channel": {"action_required": True, "question": question}}, + "user_todo_summary": {"agent_scope_filter": { + "agent_id": "peer-a", "current_agent_blocking_open_count": 0, + "other_agent_scoped_open_count": 1, + }}} + before = json.dumps([status, quota], sort_keys=True) + monkeypatch.setattr(summary_all, "collect_status", lambda **_: status) + monkeypatch.setattr(summary_all, "build_quota_should_run", lambda *_, **__: quota) + kwargs = {"registry_path": tmp_path / "registry.json", "runtime_root_override": None, + "agent_id": "peer-a", "limit": 8} + result = (summary_all.build_summary_all(**kwargs, time_range="24h") + if command == "global-summary" else summary_all.build_global_gates(**kwargs)) + assert result["summary"]["open_gate_count"] == 1 + assert result["gates"][0]["question"] == question + assert json.dumps([status, quota], sort_keys=True) == before + + +@pytest.mark.parametrize("provider", ["file", "sqlite"]) +@pytest.mark.parametrize("reverse", [False, True]) +@pytest.mark.parametrize("mixed", [False, True], ids=["ninth-worker", "worker-and-peer-gate"]) +@pytest.mark.parametrize("extra_gate", [False, True], ids=["original-gates", "another-peer-gate"]) +def test_real_peer_manager_reads_preserve_work_gates_and_authority( + tmp_path, monkeypatch, provider, reverse, mixed, extra_gate, +): + isolate_sqlite_runtime(tmp_path, monkeypatch) + project = tmp_path / "synthetic project" + project.mkdir() + state, runtime = project / "state.md", tmp_path / "runtime" + runtime.mkdir() + registry = global_registry_path(runtime) + goal = "synthetic-peer-manager" + worker = "agent-1" if mixed else "agent-9" + gate = ( + "- [ ] Approve synthetic peer output?\n" + " \n" + ) if mixed else "" + held = ( + "- [ ] Verify held peer output.\n" + " \n" + ) if mixed else "" + if extra_gate: + gate += ( + "- [ ] Approve another peer output?\n" + " \n" + ) + held += ( + "- [ ] Verify another held peer output.\n" + " \n" + ) + state.write_text( + "---\nstatus: active\nobjective: Read synthetic peer work without executing it.\n---\n" + "# Synthetic Goal\n\n## Objective\n\nRead synthetic peer work without executing it.\n" + "\n## Operating Contract\n\n- Read-only synthetic qualification.\n" + "\n## User Todo\n\n" + gate + "\n## Agent Todo\n\n" + "- [ ] Verify runnable synthetic output.\n" + f" \n" + held + + "\n## Completed Work Archive\n" + ) + agents = [f"agent-{i}" for i in range(1, 10)] + registry.write_text(json.dumps({"common_runtime_root": str(runtime), "goals": [{ + "id": goal, "repo": str(project), "state_file": state.name, "status": "active", + "objective": "Read synthetic peer work without executing it.", "domain": "test", + "adapter": {"kind": "read_only_project_map_v0", "status": "connected-read-only"}, + "coordination": {"agent_model": "peer_v1", "registered_agents": agents[::-1] if reverse else agents}, + }]})) + initialize_canonical_source_fixture(runtime, goal, state_path=state, provider=provider) + repo = Path(__file__).resolve().parents[1] + prefix = [sys.executable, "-m", "loopx.cli", "--registry", str(registry), + "--runtime-root", str(runtime), "--format", "json"] + + def read(*args): + result = subprocess.run([*prefix, *args], cwd=repo, text=True, capture_output=True, + env={**os.environ, "LOOPX_USAGE_PING": "0"}, timeout=45) + assert result.returncode == 0, result.stdout + result.stderr + payload = json.loads(result.stdout) + assert payload["ok"] is True + return payload + + before = read("todo", "list", "--goal-id", goal) + source_bytes, registry_bytes = state.read_bytes(), registry.read_bytes() + todos = read("global-todos") + summary = read("global-summary") + gates = read("global-gates") + assert "todo_ready" in {row["todo_id"] for row in todos["groups"]["runnable"]} + assert len(summary["lanes"]) == 1 + assert summary["lanes"][0]["agent_id"] == worker + assert summary["lanes"][0]["top_todo_id"] == "todo_ready" + assert summary["summary"]["open_gate_count"] == gates["summary"]["open_gate_count"] == int(mixed) + int(extra_gate) + if mixed: + original_gate = next(row for row in gates["gates"] if row["blocks"] == ["todo_held"]) + assert "Approve synthetic peer output?" in original_gate["question"] + explicit = read("global-gates", "--agent-id", "agent-1") + assert explicit["gates"] == [] + if extra_gate: + other_gate = next(row for row in gates["gates"] if row["blocks"] == ["todo_other_held"]) + assert "Approve another peer output?" in other_gate["question"] + after = read("todo", "list", "--goal-id", goal) + assert before["todos"] == after["todos"] + assert before["authority_read"] == after["authority_read"] + assert state.read_bytes() == source_bytes and registry.read_bytes() == registry_bytes + for payload in (todos, summary, gates): + assert str(tmp_path) not in json.dumps(payload)