diff --git a/docs/product/core-control-plane/rule-seam-map.md b/docs/product/core-control-plane/rule-seam-map.md index 2f672b483..ec325d5da 100644 --- a/docs/product/core-control-plane/rule-seam-map.md +++ b/docs/product/core-control-plane/rule-seam-map.md @@ -79,6 +79,9 @@ private evidence migration, production actions, or public first-screen copy. responsibilities. - `agent_lane_next_action` is per-agent runnable routing; `## Next Action` is durable goal-level guidance. +- `goal_route_hint` may summarize current, other-agent, and unclaimed lane + signals for hosts, but it is read-only advisory projection and must preserve + shared `## Next Action`. - Monitor routing is attribute based, not name based: `task_class=continuous_monitor` plus due metadata defines due monitor work, while `waiting_on=external_evidence` or diff --git a/docs/quota-allocation.md b/docs/quota-allocation.md index 598d0c495..596687d40 100644 --- a/docs/quota-allocation.md +++ b/docs/quota-allocation.md @@ -633,7 +633,13 @@ When available, `quota should-run` also keeps next-action signals separate: `active_state_next_action` is the durable `## Next Action`, `latest_run_recommended_action` is the latest non-agent-lane run's recommendation, and `agent_lane_next_action` is the current `--agent-id` -slice. If the active-state and latest-run actions differ, +slice. Agent-scoped payloads may also include +`goal_route_hint.schema_version=goal_route_hint_v0`, a compact read-only +synthesis that says whether the current lane should run, claim, wait, or +reassign while preserving `## Next Action` as durable goal-level guidance. It +is an advisory routing hint, not a writeback instruction and not a replacement +for `agent_todo_summary`. +If the active-state and latest-run actions differ, `next_action_projection_warning` asks the executor to explicitly write back the intended durable route with a primary goal-scope `refresh-state --next-action` or keep treating the signals as distinct. diff --git a/docs/status-data-contract.md b/docs/status-data-contract.md index a80165eac..43e73b183 100644 --- a/docs/status-data-contract.md +++ b/docs/status-data-contract.md @@ -1125,6 +1125,13 @@ render it as an agent-lane pointer, not as `recommended_action` replacement. Human markdown should label this pointer as the current agent's todo and mark co-displayed global agent todo rows as goal-wide, so `--agent-id` is not mistaken for a filter that replaces the primary/global queue. +The same scoped guard may include +`goal_route_hint.schema_version=goal_route_hint_v0`. This is a goal-level +read-path synthesis over the current `agent_lane_next_action`, +`agent_scope_frontier`, and compact per-agent todo lanes. It carries +`preserves_goal_next_action=true` and `goal_next_action_mutation=none` so hosts +can explain the lane decision without mutating shared `## Next Action` or +collapsing other agents' queues into the current agent's route. Within a candidate source, selection is ordered by current-agent claim first, then `capability_repair_mode=true`, then priority/index. A repair-mode item therefore stays visible as the suggested agent-lane slice even when an older diff --git a/examples/work-lane-contract-smoke.py b/examples/work-lane-contract-smoke.py index 937b470fe..0cfb841e3 100644 --- a/examples/work-lane-contract-smoke.py +++ b/examples/work-lane-contract-smoke.py @@ -1258,9 +1258,25 @@ def assert_side_agent_next_action_projects_without_stealing_goal_next_action() - assert next_action["confidence"] == "selected", next_action assert next_action["source"] == "capability_gate.runnable_candidates", next_action assert next_action["preserves_goal_next_action"] is True, next_action + route_hint = guard["goal_route_hint"] + assert route_hint["schema_version"] == "goal_route_hint_v0", route_hint + assert route_hint["route_decision"] == "run_current_agent_lane", route_hint + assert route_hint["agent_id"] == "codex-side-bypass", route_hint + assert route_hint["primary_agent"] == "codex-main-control", route_hint + assert route_hint["preserves_goal_next_action"] is True, route_hint + assert route_hint["goal_next_action_mutation"] == "none", route_hint + assert route_hint["has_durable_next_action"] is True, route_hint + assert route_hint["has_latest_run_recommended_action"] is False, route_hint + assert route_hint["selected_action_differs_from_durable"] is True, route_hint + assert route_hint["current_agent_next_action"]["todo_id"] == "todo_side_tui", route_hint + assert route_hint["other_agent_next_actions"][0]["todo_id"] == "todo_primary", route_hint + assert route_hint["counts"]["current_agent_claimed_advancement_count"] == 2, route_hint + assert route_hint["counts"]["other_agent_claimed_advancement_count"] == 1, route_hint assert guard["capability_gate"]["runnable_candidates"][0]["todo_id"] == "todo_side_tui", guard markdown = render_quota_should_run_markdown(guard) assert "agent_lane_next_action: todo_id=todo_side_tui" in markdown, markdown + assert "goal_route_hint: decision=run_current_agent_lane" in markdown, markdown + assert "goal_route_hint_current_action: todo_id=todo_side_tui" in markdown, markdown assert side_action in markdown, markdown assert primary_action not in guard["agent_lane_next_action"]["text"], guard @@ -1314,6 +1330,14 @@ def assert_side_agent_waits_when_only_other_agent_has_claimed_work() -> None: assert hint["reason_code"] == "blocked_by_other_agent_frontier", hint assert hint["target_todo_id"] == "todo_primary_suite", hint assert hint["quiet_noop_allowed"] is True, hint + route_hint = guard["goal_route_hint"] + assert route_hint["schema_version"] == "goal_route_hint_v0", route_hint + assert route_hint["route_decision"] == "agent_scope_wait", route_hint + assert route_hint["preserves_goal_next_action"] is True, route_hint + assert route_hint["has_durable_next_action"] is True, route_hint + assert route_hint["frontier_hint"]["decision"] == "quiet_noop_blocker", route_hint + assert route_hint["other_agent_next_actions"][0]["todo_id"] == "todo_primary_suite", route_hint + assert "current_agent_next_action" not in route_hint, route_hint assert frontier["frontier_hint"] == hint, frontier assert "codex-side-bypass" in guard["recommended_action"], guard assert "codex-main-control" in guard["recommended_action"], guard diff --git a/loopx/policies/goal_route_hint.py b/loopx/policies/goal_route_hint.py new file mode 100644 index 000000000..59525d731 --- /dev/null +++ b/loopx/policies/goal_route_hint.py @@ -0,0 +1,309 @@ +from __future__ import annotations + +from typing import Any + +from ..state_projection import actions_are_projection_aligned +from ..todo_contract import ( + TODO_STATUS_OPEN, + TODO_TASK_CLASS_ADVANCEMENT, + normalize_todo_blocks_agent, + normalize_todo_claimed_by, + normalize_todo_id, + normalize_todo_status, + normalize_todo_task_class, +) +from ..todo_handoff_gate import HandoffGateState + + +GOAL_ROUTE_HINT_SCHEMA_VERSION = "goal_route_hint_v0" +GOAL_ROUTE_HINT_ITEM_LIMIT = 3 + + +def _compact_text(value: Any, *, limit: int = 220) -> str: + text = " ".join(str(value or "").strip().split()) + if len(text) <= limit: + return text + return text[: limit - 1].rstrip() + "…" + + +def _int_count(value: Any) -> int: + try: + return max(0, int(value or 0)) + except (TypeError, ValueError): + return 0 + + +def _is_actionable_advancement(item: dict[str, Any]) -> bool: + if not isinstance(item, dict): + return False + if item.get("done") is True: + return False + status = normalize_todo_status(item.get("status")) + if status and status != TODO_STATUS_OPEN: + return False + return ( + normalize_todo_task_class( + item.get("task_class"), + text=str(item.get("text") or ""), + action_kind=item.get("action_kind"), + ) + == TODO_TASK_CLASS_ADVANCEMENT + ) + + +def _compact_todo(item: dict[str, Any]) -> dict[str, Any] | None: + if not _is_actionable_advancement(item): + return None + text = _compact_text(item.get("text"), limit=220) + if not text: + return None + payload: dict[str, Any] = {} + for key, normalizer in ( + ("todo_id", normalize_todo_id), + ("claimed_by", normalize_todo_claimed_by), + ("blocks_agent", normalize_todo_blocks_agent), + ("unblocks_todo_id", normalize_todo_id), + ): + normalized = normalizer(item.get(key)) + if normalized: + payload[key] = normalized + for key in ("priority", "action_kind"): + if item.get(key) is not None: + payload[key] = item.get(key) + if not payload.get("todo_id"): + payload["text"] = _compact_text(text, limit=120) + return payload + + +def _todo_items( + items: Any, + *, + agent_id: str, + claim: str, + limit: int = GOAL_ROUTE_HINT_ITEM_LIMIT, +) -> list[dict[str, Any]]: + if not isinstance(items, list): + return [] + selected: list[dict[str, Any]] = [] + seen: set[str] = set() + for item in items: + if not isinstance(item, dict): + continue + claimed_by = normalize_todo_claimed_by(item.get("claimed_by")) + if claim == "current" and claimed_by != agent_id: + continue + if claim == "other" and (not claimed_by or claimed_by == agent_id): + continue + if claim == "unclaimed" and claimed_by: + continue + compact = _compact_todo(item) + if not compact: + continue + identity = str(compact.get("todo_id") or compact.get("text") or "") + if identity in seen: + continue + seen.add(identity) + selected.append(compact) + if len(selected) >= limit: + break + return selected + + +def _count_advancement_items(items: Any, *, agent_id: str | None = None, claim: str | None = None) -> int: + if not isinstance(items, list): + return 0 + count = 0 + for item in items: + if not isinstance(item, dict) or not _is_actionable_advancement(item): + continue + claimed_by = normalize_todo_claimed_by(item.get("claimed_by")) + if claim == "unclaimed" and claimed_by: + continue + if claim == "other" and (not claimed_by or claimed_by == agent_id): + continue + if claim == "current" and claimed_by != agent_id: + continue + count += 1 + return count + + +def _current_action(agent_lane_next_action: dict[str, Any] | None) -> dict[str, Any] | None: + if not isinstance(agent_lane_next_action, dict): + return None + action: dict[str, Any] = {} + todo_id = normalize_todo_id(agent_lane_next_action.get("todo_id")) + if todo_id: + action["todo_id"] = todo_id + for key in ("selected_by", "confidence", "source"): + if agent_lane_next_action.get(key) is not None: + action[key] = agent_lane_next_action.get(key) + if not action.get("todo_id"): + text = _compact_text(agent_lane_next_action.get("text"), limit=120) + if not text: + return None + action["text"] = text + if agent_lane_next_action.get("claim_required_before_work") is True: + action["claim_required_before_work"] = True + return action + + +def _blocking_handoff_gates(agent_todo_summary: dict[str, Any], *, agent_id: str) -> list[dict[str, Any]]: + gates = agent_todo_summary.get("current_agent_handoff_gates") + if not isinstance(gates, list): + gates = agent_todo_summary.get("handoff_gates") + if not isinstance(gates, list): + return [] + selected: list[dict[str, Any]] = [] + seen: set[str] = set() + for item in gates: + if not isinstance(item, dict): + continue + if normalize_todo_blocks_agent(item.get("blocks_agent")) != agent_id: + continue + if item.get("gate_state") != HandoffGateState.BLOCKING.value: + continue + identity = str(item.get("todo_id") or item.get("index") or item.get("text") or "") + if identity in seen: + continue + seen.add(identity) + compact: dict[str, Any] = {} + for key, normalizer in ( + ("todo_id", normalize_todo_id), + ("claimed_by", normalize_todo_claimed_by), + ("blocks_agent", normalize_todo_blocks_agent), + ("unblocks_todo_id", normalize_todo_id), + ): + normalized = normalizer(item.get(key)) + if normalized: + compact[key] = normalized + if item.get("gate_state") is not None: + compact["gate_state"] = item.get("gate_state") + selected.append(compact) + if len(selected) >= GOAL_ROUTE_HINT_ITEM_LIMIT: + break + return selected + + +def build_goal_route_hint( + *, + agent_identity: dict[str, Any] | None, + agent_todo_summary: dict[str, Any] | None, + agent_lane_next_action: dict[str, Any] | None, + agent_scope_frontier: dict[str, Any] | None, + agent_lane_frontier_hint: dict[str, Any] | None, + active_state_next_action: Any, + latest_run_recommended_action: Any, + selected_recommended_action: Any, +) -> dict[str, Any] | None: + if not isinstance(agent_identity, dict): + return None + agent_id = normalize_todo_claimed_by(agent_identity.get("agent_id")) + if not agent_id: + return None + agent_summary = agent_todo_summary if isinstance(agent_todo_summary, dict) else {} + current_action = _current_action(agent_lane_next_action) + other_agent_actions = _todo_items( + agent_summary.get("claimed_advancement_open_items"), + agent_id=agent_id, + claim="other", + ) + if not other_agent_actions: + other_agent_actions = _todo_items( + agent_summary.get("claimed_by_others_items"), + agent_id=agent_id, + claim="other", + ) + unclaimed_actions = _todo_items( + agent_summary.get("unclaimed_priority_open_items"), + agent_id=agent_id, + claim="unclaimed", + ) + blocking_handoff_gates = _blocking_handoff_gates(agent_summary, agent_id=agent_id) + current_count = _int_count(agent_summary.get("current_agent_claimed_advancement_count")) + unclaimed_count = _count_advancement_items( + agent_summary.get("unclaimed_priority_open_items"), + agent_id=agent_id, + claim="unclaimed", + ) + other_count = _int_count(agent_summary.get("claimed_advancement_open_count")) + if not other_count: + other_count = _count_advancement_items( + agent_summary.get("claimed_advancement_open_items"), + agent_id=agent_id, + claim="other", + ) + else: + other_count = max(0, other_count - current_count) + other_count = max(other_count, len(other_agent_actions)) + + route_decision = "durable_goal_route" + reason = "no agent-lane override is needed; preserve durable goal Next Action" + if current_action: + if current_action.get("claim_required_before_work") is True: + route_decision = "claim_then_run_current_agent" + reason = "quota selected an unclaimed in-scope todo for this agent lane" + else: + route_decision = "run_current_agent_lane" + reason = "quota selected a runnable current-agent lane todo" + elif isinstance(agent_scope_frontier, dict): + route_decision = str( + agent_scope_frontier.get("effective_action") + or agent_scope_frontier.get("action") + or "agent_scope_frontier" + ) + reason = str(agent_scope_frontier.get("reason") or "agent-scope frontier blocks delivery") + elif blocking_handoff_gates or other_agent_actions: + route_decision = "wait_or_reassign_other_agent_lane" + reason = "visible advancement work is owned by another agent lane" + elif unclaimed_actions: + route_decision = "claim_unowned_in_scope" + reason = "unclaimed advancement work is available to the current agent lane" + + active_text = _compact_text(active_state_next_action, limit=260) + selected_text = _compact_text(selected_recommended_action, limit=260) + latest_text = _compact_text(latest_run_recommended_action, limit=260) + hint: dict[str, Any] = { + "schema_version": GOAL_ROUTE_HINT_SCHEMA_VERSION, + "kind": "agent_lane_synthesis", + "agent_id": agent_id, + "primary_agent": normalize_todo_claimed_by(agent_identity.get("primary_agent")), + "role": agent_identity.get("role"), + "route_decision": route_decision, + "reason": reason, + "preserves_goal_next_action": True, + "goal_next_action_mutation": "none", + "counts": { + "current_agent_claimed_advancement_count": current_count, + "unclaimed_advancement_count": unclaimed_count, + "other_agent_claimed_advancement_count": other_count, + "blocking_handoff_gate_count": len(blocking_handoff_gates), + }, + "selected_action_differs_from_durable": bool( + active_text + and selected_text + and not actions_are_projection_aligned(active_text, selected_text) + ), + "has_durable_next_action": bool(active_text), + "has_latest_run_recommended_action": bool(latest_text), + } + if current_action: + hint["current_agent_next_action"] = current_action + if other_agent_actions: + hint["other_agent_next_actions"] = other_agent_actions + if unclaimed_actions: + hint["unclaimed_next_actions"] = unclaimed_actions + if blocking_handoff_gates: + hint["blocking_handoff_gates"] = blocking_handoff_gates + if isinstance(agent_lane_frontier_hint, dict): + hint["frontier_hint"] = { + key: agent_lane_frontier_hint[key] + for key in ( + "decision", + "source", + "reason_code", + "target_todo_id", + "quiet_noop_allowed", + ) + if agent_lane_frontier_hint.get(key) is not None + } + return hint diff --git a/loopx/quota.py b/loopx/quota.py index 1dc107e31..b63ddc520 100644 --- a/loopx/quota.py +++ b/loopx/quota.py @@ -32,6 +32,7 @@ from .long_task_cadence import long_task_cadence_hint_summary from .orchestration import compact_orchestration_policy, orchestration_policy_summary from .policies.execution_obligation import build_execution_obligation +from .policies.goal_route_hint import build_goal_route_hint from .policies.monitor_todo import ( monitor_todo_expires_at, monitor_todo_is_actionable_open, @@ -6475,6 +6476,16 @@ def build_quota_should_run( latest_run_recommended_action=latest_run_recommended_action_text, agent_lane_next_action=agent_lane_next_action_text, ) + goal_route_hint = build_goal_route_hint( + agent_identity=agent_identity, + agent_todo_summary=agent_todo_summary, + agent_lane_next_action=agent_lane_next_action, + agent_scope_frontier=agent_scope_frontier, + agent_lane_frontier_hint=agent_lane_frontier_hint, + active_state_next_action=active_state_next_action_text, + latest_run_recommended_action=latest_run_recommended_action_text, + selected_recommended_action=selected_recommended_action, + ) agent_scope_action = _agent_scope_frontier_action(effective_action) payload = { "ok": bool(plan.get("ok")) or self_repair_allowed or capability_repair_allowed or workspace_repair_allowed, @@ -6571,6 +6582,8 @@ def build_quota_should_run( payload["agent_lane_next_action"] = agent_lane_next_action if agent_lane_frontier_hint: payload["agent_lane_frontier_hint"] = agent_lane_frontier_hint + if goal_route_hint: + payload["goal_route_hint"] = goal_route_hint if agent_scope_frontier: payload["agent_scope_frontier"] = agent_scope_frontier if workspace_guard: @@ -8342,6 +8355,35 @@ def render_quota_should_run_markdown(payload: dict[str, Any]) -> str: f"reason_code={agent_lane_frontier_hint.get('reason_code')} " f"target_todo_id={agent_lane_frontier_hint.get('target_todo_id')}" ) + goal_route_hint = ( + payload.get("goal_route_hint") + if isinstance(payload.get("goal_route_hint"), dict) + else {} + ) + if goal_route_hint: + counts = ( + goal_route_hint.get("counts") + if isinstance(goal_route_hint.get("counts"), dict) + else {} + ) + lines.append( + "- goal_route_hint: " + f"decision={goal_route_hint.get('route_decision')} " + f"agent_id={goal_route_hint.get('agent_id')} " + f"preserves_goal_next_action={goal_route_hint.get('preserves_goal_next_action')} " + f"other_agent_claimed_advancement_count={counts.get('other_agent_claimed_advancement_count')}" + ) + current_action = ( + goal_route_hint.get("current_agent_next_action") + if isinstance(goal_route_hint.get("current_agent_next_action"), dict) + else {} + ) + if current_action: + lines.append( + "- goal_route_hint_current_action: " + f"todo_id={current_action.get('todo_id')} " + f"selected_by={current_action.get('selected_by')}" + ) agent_scope_frontier = ( payload.get("agent_scope_frontier") if isinstance(payload.get("agent_scope_frontier"), dict)