fix(quota): resolve an explicit --todo-id from the Goal's own rows

loopx/control_plane/quota/should_run_prepare.py

@@ -30,7 +30,6 @@
 from ..goals.goal_frontier import (
     build_goal_frontier_projection_context_from_status,
 )
-from ..quota.blocked_transition_notice import build_blocked_transition_notice
 from ..quota.error_codes import HeartbeatReceiptIdentityConflictError
 from ..agents.capability_memory import resolve_agent_capabilities
 from ..quota.goal_boundary import (
@@ -93,6 +92,7 @@
     select_task_orchestration_authority_items,
 )
 from ..todos.summary_item import compact_todo_summary_item
+from ..todos.goal_todo_projection import goal_todo_summaries
 from ..todos.user_gate import (
     open_todo_count as _open_todo_count,
 )
@@ -229,7 +229,6 @@ def _blocked_priority_fallback(
         return None
 
     blocked_items: list[dict[str, Any]] = []
-    transition_notices: list[dict[str, Any]] = []
     owner_visible_blocker = False
     for item in first_open:
         if not isinstance(item, dict):
@@ -271,13 +270,6 @@ def _blocked_priority_fallback(
             status == TODO_STATUS_BLOCKED or resume_condition_pending
         ):
             owner_visible_blocker = True
-            # The owner notice is a typed contract, not only a boolean: it
-            # carries the cause, evidence, impact, responsible party, recovery
-            # condition and next action that #4381 asks for, and it keeps
-            # "must know" separate from "must act". See blocked_transition_notice.
-            notice = build_blocked_transition_notice(item, selected_executable=selected)
-            if notice is not None:
-                transition_notices.append(notice)
 
     if not blocked_items:
         return None
@@ -302,7 +294,6 @@ def _blocked_priority_fallback(
             )
         ),
         "blocked_items": blocked_items[:3],
-        "blocked_transition_notices": transition_notices[:3],
         "selected_executable": selected_item,
         "recommended_action": (
             "Keep the blocked core todo visible in status while selecting fallback; "
@@ -452,6 +443,48 @@ def _build_agent_work_lane(
     return monitor_only, work_lane, task_orchestration
 
 
+def _authoritative_requested_agent_rows(
+    *,
+    registry_goal: Mapping[str, Any] | None,
+    todo_id: str | None,
+) -> list[dict[str, Any]]:
+    """Resolve one explicitly requested Todo from the Goal's own state rows.
+
+    The presented planning lanes are a display budget. Before this lookup an
+    owned, open, typed advancement Todo that sat outside them was unreachable by
+    an explicit ``--todo-id``, so the caller had no legal way to bind its own
+    quota guard to the row it had just worked on. Eligibility predicates still
+    run in the builder, so this widens reachability of the lookup, not what may
+    be selected.
+    """
+
+    normalized_todo_id = normalize_todo_id(todo_id)
+    if not normalized_todo_id or not isinstance(registry_goal, Mapping):
+        return []
+    from ...materials import goal_state_path
+
+    state_path = goal_state_path(dict(registry_goal))
+    if state_path is None or not state_path.is_file():
+        return []
+    try:
+        projected = goal_todo_summaries(
+            dict(registry_goal),
+            state_text=state_path.read_text(encoding="utf-8"),
+            state_path=state_path,
+            rollout_events=[],
+            roles=["agent"],
+            status="open",
+            todo_id=normalized_todo_id,
+            agent_id=None,
+            limit=None,
+        )
+    except (OSError, ValueError):
+        # A malformed or unreadable state file must not fail the guard; the
+        # presented lanes still decide, and the caller keeps the same refusal.
+        return []
+    return [row for row in projected.todos if isinstance(row, dict)]
+
+
 def _prepare_quota_should_run_item(
     status_payload: dict[str, Any],
     *,
@@ -827,6 +860,10 @@ def _prepare_quota_should_run_item(
                     capability_gate=capability_gate,
                 ),
                 *agent_todo_planning_source_items,
+                *_authoritative_requested_agent_rows(
+                    registry_goal=registry_goal,
+                    todo_id=requested_action_todo_id,
+                ),
             ],
             available_capabilities=effective_available_capabilities,
             todo_id=requested_action_todo_id,

tests/control_plane/test_explicit_todo_id_beyond_presented_lanes.py

@@ -0,0 +1,118 @@
+"""An explicit --todo-id must reach the Goal's own rows, not only displayed lanes."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+from loopx.control_plane.agents.agent_lane_recommendation import (
+    build_explicit_advancement_next_action,
+)
+from loopx.control_plane.quota.should_run_prepare import (
+    _authoritative_requested_agent_rows,
+)
+from loopx.control_plane.todos.quota_summary import (
+    select_planning_inventory_source_items,
+)
+
+GOAL_ID = "goal-beyond-lanes"
+AGENT_ID = "agent-beyond-lanes"
+WANTED_TODO_ID = "todo_beyond_lanes"
+OTHER_AGENT_ID = "agent-somebody-else"
+
+
+def _row(todo_id: str, *, claimed_by: str = AGENT_ID, status: str = "open") -> str:
+    return (
+        f"- [ ] [P1] row {todo_id}\n"
+        "  <!-- loopx:todo "
+        f"status={status} task_class=advancement_task claimed_by={claimed_by} "
+        f"todo_id={todo_id} -->"
+    )
+
+
+def _fixture(tmp_path: Path, rows: list[str]) -> tuple[dict, Path]:
+    project = tmp_path / "project"
+    state = project / ".codex" / "goals" / GOAL_ID / "ACTIVE_GOAL_STATE.md"
+    state.parent.mkdir(parents=True)
+    state.write_text(
+        "# Active Goal State\n\n## Agent Todo\n\n" + "\n".join(rows) + "\n",
+        encoding="utf-8",
+    )
+    goal = {
+        "id": GOAL_ID,
+        "status": "active",
+        "repo": str(project),
+        "state_file": str(state.relative_to(project)),
+        "coordination": {"agent_model": "peer_v1", "registered_agents": [AGENT_ID]},
+    }
+    return goal, state
+
+
+def _candidate(rows: list[dict], todo_id: str):
+    return build_explicit_advancement_next_action(
+        agent_identity={"agent_id": AGENT_ID},
+        agent_todo_items=rows,
+        available_capabilities=["network", "filesystem_write"],
+        todo_id=todo_id,
+        selection_binding="pending_action_selection",
+    )
+
+
+def test_a_row_outside_the_presented_lanes_is_reachable_by_id(tmp_path: Path) -> None:
+    rows = [_row(f"todo_presented_{index}") for index in range(1, 6)]
+    rows.append(_row(WANTED_TODO_ID))
+    goal, _state = _fixture(tmp_path, rows)
+    # The presented lanes carry a bounded selection, so the row the caller
+    # worked on is absent from them.
+    presented_summary = {
+        "items": [
+            {"todo_id": "todo_presented_1", "text": "[P1] row", "status": "open", "task_class": "advancement_task", "claimed_by": AGENT_ID},
+            {"todo_id": "todo_presented_2", "text": "[P1] row", "status": "open", "task_class": "advancement_task", "claimed_by": AGENT_ID},
+        ]
+    }
+    presented = select_planning_inventory_source_items(presented_summary, None)
+    assert WANTED_TODO_ID not in {item.get("todo_id") for item in presented}
+    assert _candidate(presented, WANTED_TODO_ID) is None
+
+    authoritative = _authoritative_requested_agent_rows(
+        registry_goal=goal,
+        todo_id=WANTED_TODO_ID,
+    )
+
+    assert [row["todo_id"] for row in authoritative] == [WANTED_TODO_ID]
+    assert _candidate(authoritative, WANTED_TODO_ID) is not None
+
+
+def test_unknown_id_or_unreadable_state_resolves_nothing(tmp_path: Path) -> None:
+    goal, _state = _fixture(tmp_path, [_row(WANTED_TODO_ID)])
+
+    assert _authoritative_requested_agent_rows(registry_goal=goal, todo_id="todo_unknown") == []
+    assert _authoritative_requested_agent_rows(registry_goal=goal, todo_id=None) == []
+    assert (
+        _authoritative_requested_agent_rows(
+            registry_goal={"id": GOAL_ID, "repo": str(tmp_path / "missing"), "state_file": "none.md"},
+            todo_id=WANTED_TODO_ID,
+        )
+        == []
+    )
+
+
+def test_predicates_still_refuse_another_agents_or_blocked_rows(tmp_path: Path) -> None:
+    goal, _state = _fixture(
+        tmp_path,
+        [
+            _row("todo_other_agent", claimed_by=OTHER_AGENT_ID),
+            _row("todo_blocked_row", status="blocked"),
+        ],
+    )
+
+    other_agent_rows = _authoritative_requested_agent_rows(
+        registry_goal=goal,
+        todo_id="todo_other_agent",
+    )
+    blocked_rows = _authoritative_requested_agent_rows(
+        registry_goal=goal,
+        todo_id="todo_blocked_row",
+    )
+
+    assert _candidate(other_agent_rows, "todo_other_agent") is None
+    assert _candidate(blocked_rows, "todo_blocked_row") is None