fix(control-plane): require visions for long-lived peer lanes (#2343)

Author: huangruiteng

examples/control_plane/agent-identity-readmodel-smoke.py

@@ -13,6 +13,7 @@
     build_quota_agent_identity,
     quota_registered_agents,
 )
+from loopx.control_plane.agents.profile import normalize_agent_profile  # noqa: E402
 from loopx.control_plane.agents.runtime_model import (  # noqa: E402
     AgentRuntimeModel,
     agent_runtime_model_for_goal,
@@ -87,6 +88,40 @@ def assert_peer_identity_projects_only_valid_advisory_profile() -> None:
     assert "agent_profile" not in invalid_identity, invalid_identity
 
 
+def assert_long_lived_profile_requires_vision_by_default() -> None:
+    profile = normalize_agent_profile(
+        {
+            "agent_id": AGENTS[1],
+            "default_task_classes": ["advancement_task", "continuous_monitor"],
+        },
+        registered_agents=AGENTS,
+    )
+    assert profile["vision_requirement"] == "required", profile
+
+    optional = normalize_agent_profile(
+        {
+            "agent_id": AGENTS[1],
+            "default_task_classes": ["advancement_task", "continuous_monitor"],
+            "vision_requirement": "optional",
+        },
+        registered_agents=AGENTS,
+    )
+    assert optional["vision_requirement"] == "optional", optional
+
+    try:
+        normalize_agent_profile(
+            {
+                "agent_id": AGENTS[1],
+                "vision_requirement": "sometimes",
+            },
+            registered_agents=AGENTS,
+        )
+    except ValueError as exc:
+        assert "vision_requirement" in str(exc), exc
+    else:
+        raise AssertionError("invalid vision requirement should fail closed")
+
+
 def assert_legacy_state_only_projects_migration() -> None:
     goal = legacy_goal()
     identity = build_quota_agent_identity(goal, agent_id=AGENTS[1])
@@ -136,6 +171,7 @@ def assert_errors_are_actionable() -> None:
 def main() -> None:
     assert_peer_identity_has_no_rank()
     assert_peer_identity_projects_only_valid_advisory_profile()
+    assert_long_lived_profile_requires_vision_by_default()
     assert_legacy_state_only_projects_migration()
     assert_assignment_is_deterministic()
     assert_errors_are_actionable()

examples/control_plane/quota-replan-decision-plane-smoke.py

@@ -1065,6 +1065,58 @@ def assert_repeat_advancement_vision_replans_past_peer_only_work() -> None:
     assert "next runnable" in action["text"], guard
 
 
+def assert_required_profile_without_vision_replans_past_peer_only_work() -> None:
+    payload = status_payload(
+        [monitor_item(), primary_claimed_advancement()],
+        replan_obligation=None,
+    )
+    profile = {
+        "schema_version": "agent_profile_v1",
+        "agent_id": SIDE_AGENT,
+        "profile_role": "quality-qualification",
+        "scope_summary": "Continuous qualification and maintainability work.",
+        "default_task_classes": ["advancement_task", "continuous_monitor"],
+    }
+    for coordination in (
+        payload["attention_queue"]["items"][0]["coordination"],
+        payload["run_history"]["goals"][0]["coordination"],
+    ):
+        coordination["agent_profiles"] = {SIDE_AGENT: profile}
+
+    guard = build_quota_should_run(
+        payload,
+        goal_id=GOAL_ID,
+        agent_id=SIDE_AGENT,
+    )
+
+    assert guard["decision"] == "autonomous_replan_required", guard
+    assert guard["effective_action"] == "autonomous_replan_required", guard
+    assert guard["agent_identity"]["agent_profile"]["vision_requirement"] == (
+        "required"
+    ), guard
+    gaps = guard["goal_frontier_projection"]["acceptance_gaps"]
+    assert gaps[0]["kind"] == "required_agent_vision_missing", guard
+    assert gaps[0]["source"] == "agent_profile", guard
+    frontier = guard["goal_frontier_projection"]["remaining_advancement_frontier"]
+    assert frontier == {
+        "current_agent_claimed_advancement_count": 0,
+        "unclaimed_advancement_count": 0,
+        "other_agent_claimed_advancement_count": 1,
+    }, guard
+    assert "agent_scope_frontier" not in guard, guard
+
+    profile["vision_requirement"] = "optional"
+    optional_guard = build_quota_should_run(
+        payload,
+        goal_id=GOAL_ID,
+        agent_id=SIDE_AGENT,
+    )
+    assert optional_guard["decision"] != "autonomous_replan_required", optional_guard
+    assert optional_guard["goal_frontier_projection"]["acceptance_gaps"] == [], (
+        optional_guard
+    )
+
+
 def assert_repeat_advancement_vision_accepts_runnable_successor() -> None:
     guard = build_quota_should_run(
         status_payload(
@@ -1584,6 +1636,7 @@ def main() -> None:
     assert_due_monitor_runs_under_watched_open_agent_vision()
     assert_repeat_advancement_vision_beats_watch_lane_continuation_ack()
     assert_repeat_advancement_vision_replans_past_peer_only_work()
+    assert_required_profile_without_vision_replans_past_peer_only_work()
     assert_repeat_advancement_vision_accepts_runnable_successor()
     assert_non_watch_replan_ack_does_not_suppress_open_agent_vision()
     assert_open_agent_vision_with_runnable_frontier_uses_neutral_gap_trigger()

loopx/cli_commands/registry_admin.py

@@ -349,7 +349,8 @@ def register_registry_admin_commands(subparsers: argparse._SubParsersAction) ->
         default=None,
         help=(
             "Validated agent_profile_v1 JSON object for a registered peer. "
-            "Repeatable; writes advisory task routing hints only."
+            "Repeatable; writes advisory task routing hints plus the peer lane's "
+            "vision requirement."
         ),
     )
     configure_goal_parser.add_argument(

loopx/cli_commands/status.py

@@ -567,6 +567,12 @@ def _build_agent_member_projection(
     scope_summary = _profile_scope_summary(profile)
     if scope_summary:
         member["scope_summary"] = scope_summary
+    vision_requirement = _compact_member_text(
+        profile.get("vision_requirement"),
+        limit=16,
+    )
+    if vision_requirement:
+        member["vision_requirement"] = vision_requirement
     member["handoff_assignment_status"] = "task_policy_selected"
     return member
 

loopx/control_plane/agents/profile.py

@@ -18,6 +18,7 @@
     "profile_role",
     "scope_summary",
     "default_task_classes",
+    "vision_requirement",
     "preferred_action_kinds",
     "avoid_action_kinds",
 }
@@ -32,6 +33,7 @@
 AGENT_PROFILE_HIERARCHY_AGENT_ROLE = re.compile(
     r"(?:^|-)(?:main|primary|side)-agent(?:-|$)"
 )
+AGENT_PROFILE_VISION_REQUIREMENTS = {"optional", "required"}
 
 
 def _bounded_text(value: Any, *, field: str, limit: int) -> str | None:
@@ -103,6 +105,29 @@ def _action_patterns(value: Any, *, field: str) -> list[str]:
     return patterns
 
 
+def _vision_requirement(value: Any, *, task_classes: list[str]) -> str | None:
+    requirement = str(value or "").strip().lower()
+    if requirement and requirement not in AGENT_PROFILE_VISION_REQUIREMENTS:
+        raise ValueError(
+            "agent profile vision_requirement must be optional or required"
+        )
+    if requirement:
+        return requirement
+    if {"advancement_task", "continuous_monitor"}.issubset(task_classes):
+        return "required"
+    return None
+
+
+def agent_profile_requires_vision(profile: Mapping[str, Any] | None) -> bool:
+    if not isinstance(profile, Mapping):
+        return False
+    requirement = str(profile.get("vision_requirement") or "").strip().lower()
+    if requirement:
+        return requirement == "required"
+    task_classes = _task_classes(profile.get("default_task_classes"))
+    return {"advancement_task", "continuous_monitor"}.issubset(task_classes)
+
+
 def normalize_agent_profile(
     raw_profile: Mapping[str, Any],
     *,
@@ -140,6 +165,7 @@ def normalize_agent_profile(
             "agent profile action kind globs cannot be both preferred and avoided: "
             + ", ".join(overlap)
         )
+    task_classes = _task_classes(raw_profile.get("default_task_classes"))
     profile = {
         "schema_version": PEER_AGENT_PROFILE_SCHEMA_VERSION,
         "agent_id": agent_id,
@@ -149,8 +175,10 @@ def normalize_agent_profile(
             field="scope_summary",
             limit=320,
         ),
-        "default_task_classes": _task_classes(
-            raw_profile.get("default_task_classes")
+        "default_task_classes": task_classes,
+        "vision_requirement": _vision_requirement(
+            raw_profile.get("vision_requirement"),
+            task_classes=task_classes,
         ),
         "preferred_action_kinds": preferred,
         "avoid_action_kinds": avoided,

loopx/control_plane/goals/goal_frontier.py

@@ -8,6 +8,7 @@
     agent_scope_item_claimed_by,
     agent_scope_item_claimed_by_agent_or_unclaimed,
 )
+from ..agents.profile import agent_profile_requires_vision
 from ..agents.runtime_model import peer_work_key, select_peer_for_work
 from ..work_items.autonomous_replan_ack import (
     autonomous_replan_ack_matches_frontier,
@@ -45,6 +46,7 @@
 VISION_ACCEPTANCE_GAP_TRIGGER = "vision_acceptance_gap"
 VISION_SUCCESSOR_GAP_TRIGGER = "vision_successor_required"
 VISION_CHECKPOINT_MISSING_TRIGGER = "vision_checkpoint_missing"
+VISION_PROFILE_MISSING_TRIGGER = "required_agent_vision_missing"
 TODO_SUCCESSION_GAP_TRIGGER = "completed_advancement_without_successor"
 TODO_TASK_CLASS_ADVANCEMENT = "advancement_task"
 TODO_TASK_CLASS_MONITOR = "continuous_monitor"
@@ -634,6 +636,41 @@ def acceptance_gaps_from_agent_vision(
     return [gap]
 
 
+def acceptance_gaps_from_agent_profile_requirement(
+    agent_profile: dict[str, Any] | None,
+    *,
+    agent_id: str | None,
+    agent_vision: dict[str, Any] | None,
+    missing_checkpoint: dict[str, Any] | None,
+) -> list[dict[str, Any]]:
+    """Require a vision baseline for registered long-lived peer lanes."""
+
+    if (
+        not agent_id
+        or not agent_profile_requires_vision(agent_profile)
+        or isinstance(agent_vision, dict)
+        or isinstance(missing_checkpoint, dict)
+    ):
+        return []
+    return [
+        {
+            "kind": VISION_PROFILE_MISSING_TRIGGER,
+            "source": "agent_profile",
+            "agent_id": agent_id,
+            "replan_trigger_summary": (
+                "the registered long-lived agent lane requires a persisted vision "
+                "baseline before ordinary delivery or monitor-only quiet wait"
+            ),
+            "acceptance_summary": (
+                "Write a bounded agent vision with objective, acceptance evidence, "
+                "advancement policy, and replan trigger; or explicitly mark the "
+                "profile vision requirement optional."
+            ),
+            "advancement_policy": "repeat_until_closed",
+        }
+    ]
+
+
 def acceptance_gaps_from_vision_checkpoint(
     checkpoint: dict[str, Any] | None,
 ) -> list[dict[str, Any]]:
@@ -1187,7 +1224,8 @@ def derive_goal_frontier_replan_obligation_from_summaries(
         item for item in (acceptance_gaps or []) if isinstance(item, dict)
     ]
     successor_vision_required = any(
-        item.get("kind") == VISION_SUCCESSOR_GAP_TRIGGER
+        item.get("kind")
+        in {VISION_SUCCESSOR_GAP_TRIGGER, VISION_PROFILE_MISSING_TRIGGER}
         for item in compact_acceptance_gaps
     )
     acceptance_allows_watch_lane_continuation = bool(
@@ -1514,6 +1552,7 @@ def build_goal_frontier_projection_context_from_status(
     neutral_replan_ack_classifications: set[str],
     registered_agent_ids: list[str] | None = None,
     goal_status: str | None = None,
+    agent_profile: dict[str, Any] | None = None,
 ) -> dict[str, Any]:
     """Build the quota-facing goal-frontier read model.
 
@@ -1548,7 +1587,13 @@ def build_goal_frontier_projection_context_from_status(
         agent_id=agent_id,
     )
     source_acceptance_gaps = (
-        acceptance_gaps_from_agent_vision(
+        acceptance_gaps_from_agent_profile_requirement(
+            agent_profile,
+            agent_id=agent_id,
+            agent_vision=latest_agent_vision,
+            missing_checkpoint=latest_missing_vision_checkpoint,
+        )
+        + acceptance_gaps_from_agent_vision(
             latest_agent_vision,
             goal_status=goal_status,
         )

loopx/heartbeat_prompt.py

@@ -136,6 +136,7 @@ def agent_profile_prompt_projection(profile: dict[str, Any] | None) -> dict[str,
         "default_scopes",
         "scopes",
         "default_task_classes",
+        "vision_requirement",
         "preferred_action_kinds",
         "avoid_action_kinds",
     }

loopx/presentation/renderers/status_markdown.py

@@ -1363,6 +1363,10 @@ def append_attention_queue_project_asset_markdown(
             member_fields.append(f"profile_role={markdown_scalar(agent_member['profile_role'])}")
         if agent_member.get("scope_summary"):
             member_fields.append(f"scope={markdown_scalar(agent_member['scope_summary'])}")
+        if agent_member.get("vision_requirement"):
+            member_fields.append(
+                f"vision_requirement={markdown_scalar(agent_member['vision_requirement'])}"
+            )
         member_fields.extend(
             [
                 f"claims={markdown_scalar(current_claims)}",

loopx/quota.py

@@ -1198,6 +1198,13 @@ def _effective_action(
     return "quota_skip"
 
 
+def _quota_agent_profile(agent_identity: dict[str, Any] | None) -> dict[str, Any] | None:
+    if not isinstance(agent_identity, dict):
+        return None
+    profile = agent_identity.get("agent_profile")
+    return profile if isinstance(profile, dict) else None
+
+
 def build_quota_should_run(
     status_payload: dict[str, Any],
     *,
@@ -1414,6 +1421,7 @@ def build_quota_should_run(
             neutral_replan_ack_classifications=AUTONOMOUS_REPLAN_ACK_NEUTRAL_CLASSIFICATIONS,
             registered_agent_ids=registered_agent_ids,
             goal_status=str(registry_goal.get("status") or ""),
+            agent_profile=_quota_agent_profile(agent_identity),
         )
         replan_obligation = goal_frontier_context.get("replan_obligation")
         replan_scope = goal_frontier_context.get("replan_scope") or {}