From 33011f4e4edc26100dcf931e56cf64e66c224455 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Sun, 1 Nov 2026 23:38:34 +0530 Subject: [PATCH 1/5] manager: fan out unscoped global-todos and global-summary across peer_v1 agents Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/global_todos.py | 121 ++++++++++++++++++++++++++++++----- loopx/summary_all.py | 60 ++++++++++++++++- tests/test_global_todos.py | 128 +++++++++++++++++++++++++++++++++++++ tests/test_summary_all.py | 93 +++++++++++++++++++++++++++ 4 files changed, 385 insertions(+), 17 deletions(-) diff --git a/loopx/global_todos.py b/loopx/global_todos.py index 5f6bbb6792..55f6a8699f 100644 --- a/loopx/global_todos.py +++ b/loopx/global_todos.py @@ -114,13 +114,62 @@ def _warning( return warning -def _quota_for_goal( +def _is_unscoped_peer_v1_quota(payload: dict[str, Any]) -> bool: + heartbeat = as_dict(payload.get("heartbeat_recommendation")) + if heartbeat.get("recommended_mode") == "automation_prompt_upgrade": + return True + reason = str(payload.get("reason") or "") + return "configured for peer_v1" in reason and "without --agent-id" in reason + + +def _is_unregistered_agent_scope_miss( + payload: dict[str, Any], + *, + agent_id: str | None, +) -> bool: + if not agent_id or payload.get("ok") is not False: + return False + detail = str(payload.get("error") or payload.get("reason") or "") + return "is not registered in the selected registry goal" in detail + + +def _registered_goal_agents( + status_payload: dict[str, Any], + goal_id: str, +) -> list[str]: + 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) + + 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 registered in as_list(coordination.get("registered_agents")): + add_agent(registered) + + queue = as_dict(status_payload.get("attention_queue")) + 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[:8] + + +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, @@ -135,8 +184,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 +195,24 @@ def _quota_for_goal( detail=payload.get("error") or payload.get("reason"), ) ) - return None - return payload + return [], True + if agent_id is None and _is_unscoped_peer_v1_quota(payload): + registered_agents = _registered_goal_agents(status_payload, goal_id) + scoped_payloads: list[dict[str, Any]] = [] + 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: + continue + if isinstance(scoped, dict) and scoped.get("ok") is True: + scoped_payloads.append(scoped) + if scoped_payloads: + return scoped_payloads, False + return [payload], False def _quota_matches_agent(payload: dict[str, Any], *, agent_id: str | None) -> bool: @@ -560,25 +627,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 c4261e9321..7e059f4740 100644 --- a/loopx/summary_all.py +++ b/loopx/summary_all.py @@ -115,6 +115,44 @@ def _goal_agent_id(item: dict[str, Any], quota_payload: dict[str, Any]) -> str | return None +def _candidate_goal_agents( + status_payload: dict[str, Any], + goal_id: str, +) -> list[str]: + 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")) + 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")) + + 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 registered in _as_list(coordination.get("registered_agents")): + add_agent(registered) + return agents[:8] + + +def _is_unscoped_peer_v1_quota(payload: dict[str, Any]) -> bool: + heartbeat = _as_dict(payload.get("heartbeat_recommendation")) + if heartbeat.get("recommended_mode") == "automation_prompt_upgrade": + return True + reason = str(payload.get("reason") or "") + return "configured for peer_v1" in reason and "without --agent-id" in reason + + def _build_goal_quota( status_payload: dict[str, Any], *, @@ -122,7 +160,7 @@ def _build_goal_quota( agent_id: str | None, ) -> dict[str, Any]: try: - return build_quota_should_run(status_payload, goal_id=goal_id, agent_id=agent_id) + payload = build_quota_should_run(status_payload, goal_id=goal_id, agent_id=agent_id) except Exception as exc: return { "ok": False, @@ -131,6 +169,26 @@ def _build_goal_quota( "decision": "quota_unavailable", "reason": _redact_text(exc), } + if agent_id is None and _is_unscoped_peer_v1_quota(payload): + first_ok: dict[str, Any] | None = None + for candidate_agent_id in _candidate_goal_agents(status_payload, goal_id): + try: + scoped = build_quota_should_run( + status_payload, + goal_id=goal_id, + agent_id=candidate_agent_id, + ) + except Exception: + continue + if not isinstance(scoped, dict) or scoped.get("ok") is False: + continue + if first_ok is None: + first_ok = scoped + if scoped.get("normal_delivery_allowed") is True or scoped.get("selected_todo"): + return scoped + if first_ok is not None: + return first_ok + return payload def _quota_unavailable(quota_payload: dict[str, Any]) -> bool: diff --git a/tests/test_global_todos.py b/tests/test_global_todos.py index fe4555c2f0..9e5df4e7a3 100644 --- a/tests/test_global_todos.py +++ b/tests/test_global_todos.py @@ -620,3 +620,131 @@ 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"] == [] diff --git a/tests/test_summary_all.py b/tests/test_summary_all.py index 85b84a9f38..4e8149021b 100644 --- a/tests/test_summary_all.py +++ b/tests/test_summary_all.py @@ -616,3 +616,96 @@ 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." From ee7441e7770f139a2f15342b4184944ef781c11d Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Mon, 2 Nov 2026 08:47:26 +0530 Subject: [PATCH 2/5] fix(manager): remove 8-agent cap, preserve peer user gates, and use typed identity checks Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- loopx/global_todos.py | 170 ++++++++++++++++++++++-------- loopx/summary_all.py | 211 ++++++++++++++++++++----------------- tests/test_global_todos.py | 64 +++++++++++ tests/test_summary_all.py | 137 ++++++++++++++++++++++++ 4 files changed, 445 insertions(+), 137 deletions(-) diff --git a/loopx/global_todos.py b/loopx/global_todos.py index 55f6a8699f..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,53 +118,104 @@ def _warning( return warning -def _is_unscoped_peer_v1_quota(payload: dict[str, Any]) -> bool: - heartbeat = as_dict(payload.get("heartbeat_recommendation")) - if heartbeat.get("recommended_mode") == "automation_prompt_upgrade": - return True - reason = str(payload.get("reason") or "") - return "configured for peer_v1" in reason and "without --agent-id" in reason - - -def _is_unregistered_agent_scope_miss( - payload: dict[str, Any], +def is_unregistered_agent_scope_miss( + payload: dict[str, Any] | BaseException, *, agent_id: str | None, ) -> bool: - if not agent_id or payload.get("ok") is not False: + 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 "") - return "is not registered in the selected registry goal" in detail + 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( +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]: - agents: list[str] = [] - seen: set[str] = set() + registered: list[str] = [] + seen_registered: set[str] = set() - def add_agent(value: object) -> None: + def add_registered(value: object) -> None: candidate = str(value or "").strip() - if candidate and candidate not in seen: - seen.add(candidate) - agents.append(candidate) + 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 registered in as_list(coordination.get("registered_agents")): - add_agent(registered) + 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")) - 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[:8] + 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( @@ -177,6 +232,8 @@ def _quotas_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", @@ -186,7 +243,7 @@ def _quotas_for_goal( ) return [], True if payload.get("ok") is False: - if _is_unregistered_agent_scope_miss(payload, agent_id=agent_id): + if is_unregistered_agent_scope_miss(payload, agent_id=agent_id): return [], False warnings.append( _warning( @@ -196,22 +253,49 @@ def _quotas_for_goal( ) ) return [], True - if agent_id is None and _is_unscoped_peer_v1_quota(payload): - registered_agents = _registered_goal_agents(status_payload, goal_id) - scoped_payloads: list[dict[str, Any]] = [] - 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: - continue - if isinstance(scoped, dict) and scoped.get("ok") is 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, False + if scoped_payloads: + return scoped_payloads, scoped_failed + if scoped_failed: + return [], True return [payload], False diff --git a/loopx/summary_all.py b/loopx/summary_all.py index 7e059f4740..9176580d71 100644 --- a/loopx/summary_all.py +++ b/loopx/summary_all.py @@ -12,6 +12,7 @@ from .control_plane.runtime.time import now_utc, 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 +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 @@ -115,80 +116,71 @@ def _goal_agent_id(item: dict[str, Any], quota_payload: dict[str, Any]) -> str | return None -def _candidate_goal_agents( - status_payload: dict[str, Any], - goal_id: str, -) -> list[str]: - 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")) - 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")) - - 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 registered in _as_list(coordination.get("registered_agents")): - add_agent(registered) - return agents[:8] - - -def _is_unscoped_peer_v1_quota(payload: dict[str, Any]) -> bool: - heartbeat = _as_dict(payload.get("heartbeat_recommendation")) - if heartbeat.get("recommended_mode") == "automation_prompt_upgrade": - return True - reason = str(payload.get("reason") or "") - return "configured for peer_v1" in reason and "without --agent-id" in reason +def _unavailable_goal_quota(goal_id: str, exc: object) -> dict[str, Any]: + return { + "ok": False, + "goal_id": goal_id, + "state": "quota_unavailable", + "decision": "quota_unavailable", + "reason": _redact_text(exc), + } -def _build_goal_quota( +def _build_goal_quotas( status_payload: dict[str, Any], *, goal_id: str, agent_id: str | None, -) -> dict[str, Any]: +) -> list[dict[str, Any]]: try: payload = build_quota_should_run(status_payload, goal_id=goal_id, agent_id=agent_id) except Exception as exc: - return { - "ok": False, - "goal_id": goal_id, - "state": "quota_unavailable", - "decision": "quota_unavailable", - "reason": _redact_text(exc), - } - if agent_id is None and _is_unscoped_peer_v1_quota(payload): - first_ok: dict[str, Any] | None = None - for candidate_agent_id in _candidate_goal_agents(status_payload, goal_id): - try: - scoped = build_quota_should_run( - status_payload, - goal_id=goal_id, - agent_id=candidate_agent_id, - ) - except Exception: - continue - if not isinstance(scoped, dict) or scoped.get("ok") is False: - continue - if first_ok is None: - first_ok = scoped - if scoped.get("normal_delivery_allowed") is True or scoped.get("selected_todo"): - return scoped - if first_ok is not None: - return first_ok - return payload + if is_unregistered_agent_scope_miss(exc, agent_id=agent_id): + return [] + return [_unavailable_goal_quota(goal_id, exc)] + if isinstance(payload, dict) and is_unregistered_agent_scope_miss(payload, agent_id=agent_id): + return [] + if agent_id is None: + candidate_agents = registered_goal_agents( + status_payload, + goal_id, + payload if isinstance(payload, dict) else None, + prefer_queue_recommendation=True, + ) + if candidate_agents: + runnable_quotas: list[dict[str, Any]] = [] + other_ok_quotas: list[dict[str, Any]] = [] + failed_quotas: list[dict[str, Any]] = [] + for candidate_agent_id in candidate_agents: + try: + scoped = build_quota_should_run( + status_payload, + goal_id=goal_id, + agent_id=candidate_agent_id, + ) + except Exception as exc: + failed_quotas.append(_unavailable_goal_quota(goal_id, exc)) + continue + if not isinstance(scoped, dict) or scoped.get("ok") is False: + failed_quotas.append( + scoped + if isinstance(scoped, dict) and _quota_unavailable(scoped) + else _unavailable_goal_quota( + goal_id, + (scoped.get("error") or scoped.get("reason")) + if isinstance(scoped, dict) + else "scoped quota payload unavailable", + ) + ) + continue + if scoped.get("normal_delivery_allowed") is True or scoped.get("selected_todo"): + runnable_quotas.append(scoped) + else: + other_ok_quotas.append(scoped) + ordered = [*runnable_quotas, *other_ok_quotas, *failed_quotas] + if ordered: + return ordered + return [payload] def _quota_unavailable(quota_payload: dict[str, Any]) -> bool: @@ -463,32 +455,46 @@ def _collect_global_gate_state( if not goal_id or goal_id in seen_goal_ids: continue seen_goal_ids.add(goal_id) - quota_payload = _build_goal_quota( + goal_quotas = _build_goal_quotas( status_payload, goal_id=goal_id, agent_id=agent_id, ) - quota_evaluated += 1 - quota_unavailable += int(_quota_unavailable(quota_payload)) - if not _quota_matches_agent_scope(quota_payload, agent_id=agent_id): - continue - gate = _global_gate_from_item( - item, - quota_payload=quota_payload, - agent_id=agent_id, - ) - if not gate: + if not goal_quotas: continue - gates.append(gate) - lane = _lane_from_item(item, quota_payload=quota_payload) - verified_todo_ids = { - str(blocked_ref) - for blocked_ref in _as_list(gate.get("blocks")) - if str(blocked_ref) != goal_id - } - if str(lane.get("top_todo_id") or "") not in verified_todo_ids: - lane.pop("top_todo_id", None) - lanes.append(lane) + quota_evaluated += 1 + quota_unavailable += int(any(_quota_unavailable(q) for q in goal_quotas)) + seen_goal_gates: set[tuple[str, str, tuple[str, ...]]] = set() + for quota_payload in goal_quotas: + if not _quota_matches_agent_scope(quota_payload, agent_id=agent_id): + continue + gate = _global_gate_from_item( + item, + quota_payload=quota_payload, + agent_id=agent_id, + ) + if not gate: + continue + gate_key = ( + str(gate.get("gate_id") or ""), + str(gate.get("question") or ""), + tuple(str(b) for b in _as_list(gate.get("blocks"))), + ) + if gate_key in seen_goal_gates: + continue + seen_goal_gates.add(gate_key) + gates.append(gate) + lane = _lane_from_item(item, quota_payload=quota_payload) + verified_todo_ids = { + str(blocked_ref) + for blocked_ref in _as_list(gate.get("blocks")) + if str(blocked_ref) != goal_id + } + if str(lane.get("top_todo_id") or "") not in verified_todo_ids: + lane.pop("top_todo_id", None) + lanes.append(lane) + if len(gates) >= limit: + break if len(gates) >= limit: break unique_goal_count = len({str(item.get("goal_id") or "").strip() for item in queue_items} - {""}) @@ -744,18 +750,35 @@ def build_summary_all( if not goal_id or goal_id in seen_goal_ids: continue seen_goal_ids.add(goal_id) - quota_payload = _build_goal_quota(status_payload, goal_id=goal_id, agent_id=agent_id) - quota_unavailable += int(_quota_unavailable(quota_payload)) + goal_quotas = _build_goal_quotas(status_payload, goal_id=goal_id, agent_id=agent_id) + if not goal_quotas: + continue + quota_payload = goal_quotas[0] + quota_unavailable += int(any(_quota_unavailable(q) for q in goal_quotas)) lane = _lane_from_item(item, quota_payload=quota_payload) if lane: lanes.append(lane) quota_states[str(lane.get("status") or "unknown")] = quota_states.get(str(lane.get("status") or "unknown"), 0) + 1 - gate = _gate_from_item(item, quota_payload=quota_payload) - if gate: - gates.append(gate) - todo = _todo_from_quota(goal_id, quota_payload) - if todo: - todos.append(todo) + seen_gate_keys: set[tuple[str, str, tuple[str, ...]]] = set() + for candidate_quota in goal_quotas: + gate = _gate_from_item(item, quota_payload=candidate_quota) + if gate: + gate_key = ( + str(gate.get("gate_id") or ""), + str(gate.get("question") or ""), + tuple(str(b) for b in _as_list(gate.get("blocks"))), + ) + if gate_key not in seen_gate_keys: + seen_gate_keys.add(gate_key) + gates.append(gate) + seen_todo_keys: set[str] = set() + for candidate_quota in goal_quotas: + todo = _todo_from_quota(goal_id, candidate_quota) + if todo: + todo_key = str(todo.get("todo_id") or todo.get("title") or "") + if todo_key not in seen_todo_keys: + seen_todo_keys.add(todo_key) + todos.append(todo) lanes.sort(key=_lane_sort_key) envelope = _global_projection_envelope( "global_summary", diff --git a/tests/test_global_todos.py b/tests/test_global_todos.py index 9e5df4e7a3..5b4fa10a3a 100644 --- a/tests/test_global_todos.py +++ b/tests/test_global_todos.py @@ -748,3 +748,67 @@ def test_agent_scope_skips_goals_where_agent_is_not_registered_without_warning( 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 + ) diff --git a/tests/test_summary_all.py b/tests/test_summary_all.py index 4e8149021b..b118d2480a 100644 --- a/tests/test_summary_all.py +++ b/tests/test_summary_all.py @@ -709,3 +709,140 @@ def build_quota(_status, *, goal_id, agent_id=None): 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" From d84a433942c3b6d3d71c93b8f9f701bddaa07c37 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 3 Nov 2026 04:12:27 +0800 Subject: [PATCH 3/5] fix(manager): preserve decisions outside scoped Todo gates Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../presentation/global_manager_read.ts | 4 +- .../global_manager_read.test.ts | 41 ++++++++++++++++++- tests/test_summary_all.py | 30 ++++++++++++++ 3 files changed, 73 insertions(+), 2 deletions(-) diff --git a/loopx/control_plane/presentation/global_manager_read.ts b/loopx/control_plane/presentation/global_manager_read.ts index 0ec63c41af..e89943c5b0 100644 --- a/loopx/control_plane/presentation/global_manager_read.ts +++ b/loopx/control_plane/presentation/global_manager_read.ts @@ -50,7 +50,9 @@ function laneFrom(item: JsonObject, quota: JsonObject): JsonObject { 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); - return typeof scope.agent_id === "string" && scope.current_agent_blocking_open_count === 0 + // 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 { diff --git a/tests/control_plane_ts/global_manager_read.test.ts b/tests/control_plane_ts/global_manager_read.test.ts index c59b48b89c..2e5cdff9fb 100644 --- a/tests/control_plane_ts/global_manager_read.test.ts +++ b/tests/control_plane_ts/global_manager_read.test.ts @@ -106,7 +106,8 @@ test("foreign-only Todo notifications do not duplicate gates; applicable and unk [{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 = {interaction_contract: {user_channel: {action_required: true}}, + 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]}; @@ -114,3 +115,41 @@ test("foreign-only Todo notifications do not duplicate gates; applicable and unk 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_summary_all.py b/tests/test_summary_all.py index a065f5f3ae..b796bf8082 100644 --- a/tests/test_summary_all.py +++ b/tests/test_summary_all.py @@ -861,6 +861,36 @@ def build_quota(_status, *, goal_id, agent_id=None): 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, +): + 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"]) From afb3b5c1787dad1293da1150d93da8c22afe729a Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 3 Nov 2026 04:21:15 +0800 Subject: [PATCH 4/5] test(manager): isolate native projection recovery fixtures Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/test_summary_all.py | 1 + 1 file changed, 1 insertion(+) diff --git a/tests/test_summary_all.py b/tests/test_summary_all.py index b796bf8082..198226a2be 100644 --- a/tests/test_summary_all.py +++ b/tests/test_summary_all.py @@ -869,6 +869,7 @@ def build_quota(_status, *, goal_id, agent_id=None): 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"}, From 150d31abc03aab3487ed5d2a346df616e70760f0 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Tue, 3 Nov 2026 06:28:11 +0800 Subject: [PATCH 5/5] docs(manager): qualify complete peer read budget and gate coverage Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../protocols/global-manager-command-v0.md | 96 ++++++++++++++++--- tests/test_global_todos.py | 43 +++++++++ tests/test_summary_all.py | 23 ++++- 3 files changed, 143 insertions(+), 19 deletions(-) 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/tests/test_global_todos.py b/tests/test_global_todos.py index 5b4fa10a3a..1f2b578a63 100644 --- a/tests/test_global_todos.py +++ b/tests/test_global_todos.py @@ -812,3 +812,46 @@ def build_quota(_status_payload, *, goal_id, agent_id=None): 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 198226a2be..fbfd73e7e7 100644 --- a/tests/test_summary_all.py +++ b/tests/test_summary_all.py @@ -895,8 +895,9 @@ def test_foreign_todo_counts_preserve_an_independent_decision_through_native_pro @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, + tmp_path, monkeypatch, provider, reverse, mixed, extra_gate, ): isolate_sqlite_runtime(tmp_path, monkeypatch) project = tmp_path / "synthetic project" @@ -916,6 +917,17 @@ def test_real_peer_manager_reads_preserve_work_gates_and_authority( " \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" @@ -955,12 +967,15 @@ def read(*args): 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) + assert summary["summary"]["open_gate_count"] == gates["summary"]["open_gate_count"] == int(mixed) + int(extra_gate) if mixed: - assert gates["gates"][0]["blocks"] == ["todo_held"] - assert "Approve synthetic peer output?" in gates["gates"][0]["question"] + 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"]