Fix operator view status and pane role registry mapping

harness/lib/pane_overlay_state.py

@@ -0,0 +1,97 @@
+"""Shared TUI pane overlay classification.
+
+The key rule: tmux capture includes scrollback. A permissions/survey/proceed
+prompt is only actionable when it is still after the latest live Claude prompt.
+If a clean prompt/footer appears after it, the text is stale history and must
+not block scheduling or recovery decisions.
+"""
+
+from __future__ import annotations
+
+import re
+
+
+FOOTER_RE = re.compile(
+    r"⏵.*(auto|accept edits|edit|bypass permissions).*mode on|shift\+tab|esc to interrupt|/effort",
+    re.I,
+)
+SURVEY_RE = re.compile(
+    r"How is Claude doing this session\?|1:\s*Bad\s+2:\s*Fine\s+3:\s*Good\s+0:\s*Dismiss|survey_blocked",
+    re.I,
+)
+PERMISSION_RE = re.compile(
+    r"permissions?_prompt_blocked|pane_permissions_prompt_blocked|Do you want to make this edit|"
+    r"allow all edits during this session|allow this command|approval required",
+    re.I,
+)
+PROCEED_RE = re.compile(r"Do you want to proceed\?|Would you like to proceed\?|Enter to confirm|Esc to cancel", re.I)
+QUEUED_RE = re.compile(r"Press up to edit queued messages|ready_for_builder|ready_for_evaluator|graph_node_idle_assigned", re.I)
+
+OVERLAY_PATTERNS = {
+    "survey": SURVEY_RE,
+    "permission": PERMISSION_RE,
+    "proceed": PROCEED_RE,
+    "queued_input": QUEUED_RE,
+}
+
+
+def live_prompt_index(lines: list[str], footer_at: int | None = None) -> int:
+    footer_at = len(lines) if footer_at is None else footer_at
+    for idx in reversed([i for i, line in enumerate(lines) if "❯" in line]):
+        if idx <= footer_at and footer_at - idx <= 8:
+            return idx
+    return -1
+
+
+def tail_has_idle_prompt_footer(text: str) -> bool:
+    lines = [line.rstrip() for line in str(text or "").splitlines()]
+    saw_footer = False
+    for line in reversed(lines[-12:]):
+        stripped = line.strip()
+        if not stripped:
+            continue
+        lowered = stripped.lower()
+        if stripped.startswith("────────────────") or stripped.isdigit():
+            continue
+        if stripped.startswith("❯"):
+            remainder = stripped[1:].strip()
+            return remainder.startswith("Try ") or (not remainder and saw_footer)
+        if FOOTER_RE.search(stripped) or lowered.startswith(("esc ", "tab ", "interrupt")) or "tokens" in lowered:
+            saw_footer = True
+            continue
+        return False
+    return False
+
+
+def prompt_match_is_stale(text: str, match: re.Match[str] | None) -> bool:
+    if match is None:
+        return False
+    after = str(text or "")[match.end():]
+    return bool(re.search(r"❯[\s\u00a0]+Try\s+\"", after)) or tail_has_idle_prompt_footer(after)
+
+
+def pane_overlay_detail(tail: str) -> dict:
+    lines = str(tail or "").splitlines()
+    if not lines:
+        return {"state": "none", "type": "", "detail": ""}
+    footer_indexes = [idx for idx, line in enumerate(lines) if FOOTER_RE.search(line)]
+    footer_at = footer_indexes[-1] if footer_indexes else len(lines)
+    live_prompt_at = live_prompt_index(lines, footer_at)
+    newest_match: tuple[int, str, str] | None = None
+    for idx, line in enumerate(lines):
+        for kind, pattern in OVERLAY_PATTERNS.items():
+            if pattern.search(line):
+                newest_match = (idx, kind, line.strip())
+    if newest_match is None:
+        return {"state": "none", "type": "", "detail": ""}
+    match_at, kind, detail = newest_match
+    if live_prompt_at >= 0 and live_prompt_at > match_at:
+        return {"state": "stale_scrollback_ignored", "type": kind, "detail": detail[:240]}
+    return {"state": "pane_overlay_blocked", "type": kind, "detail": detail[:240]}
+
+
+def pane_overlay_blocked(tail: str, *kinds: str) -> bool:
+    detail = pane_overlay_detail(tail)
+    if detail.get("state") != "pane_overlay_blocked":
+        return False
+    return not kinds or str(detail.get("type") or "") in set(kinds)

harness/lib/pane_role_pool.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import argparse
+import fnmatch
 import json
 import os
 import subprocess
@@ -23,6 +24,9 @@ def harness_dir() -> Path:
 SESSION = os.environ.get("SOLAR_HARNESS_SESSION", "solar-harness")
 HARNESS_DIR = harness_dir()
 REGISTRY_PATH = HARNESS_DIR / "run" / "pane-hygiene.json"
+PHYSICAL_OPERATORS_PATH = Path(
+    os.environ.get("SOLAR_MULTI_TASK_OPERATORS", HARNESS_DIR / "config" / "physical-operators.json")
+)
 
 
 def list_tmux_panes() -> list[dict[str, str]]:
@@ -50,10 +54,64 @@ def _allowed_session(pane: str) -> bool:
     return pane.startswith(f"{SESSION}:") or pane.startswith("solar-harness-lab:") or pane.startswith("solar-harness-multi-task:")
 
 
+def _normalize_role(role: str) -> str:
+    return str(role or "").strip().lower().replace("_", "-")
+
+
+def _load_operator_registry() -> dict[str, Any]:
+    if not PHYSICAL_OPERATORS_PATH.exists():
+        return {"version": 1, "operators": {}}
+    try:
+        payload = json.loads(PHYSICAL_OPERATORS_PATH.read_text(encoding="utf-8"))
+        return payload if isinstance(payload, dict) else {"version": 1, "operators": {}}
+    except Exception:
+        return {"version": 1, "operators": {}}
+
+
+def _operator_roles(spec: dict[str, Any]) -> set[str]:
+    raw_roles = spec.get("roles")
+    values: list[str]
+    if isinstance(raw_roles, str):
+        values = [raw_roles]
+    elif isinstance(raw_roles, list):
+        values = [str(item) for item in raw_roles]
+    else:
+        values = []
+    role = str(spec.get("role") or "").strip()
+    if role:
+        values.append(role)
+    return {_normalize_role(item) for item in values if str(item or "").strip()}
+
+
+def _pane_pattern_matches(pattern: str, pane: str) -> bool:
+    raw = str(pattern or "").strip()
+    if not raw:
+        return False
+    if raw.endswith(":*"):
+        return pane.startswith(raw[:-1])
+    return fnmatch.fnmatch(pane, raw)
+
+
+def _registry_roles_for_pane(pane: str) -> set[str]:
+    registry = _load_operator_registry()
+    operators = registry.get("operators") if isinstance(registry.get("operators"), dict) else {}
+    roles: set[str] = set()
+    for spec in operators.values():
+        if not isinstance(spec, dict):
+            continue
+        if not bool(spec.get("enabled", False)) or not bool(spec.get("available", False)):
+            continue
+        if not _pane_pattern_matches(str(spec.get("pane") or ""), pane):
+            continue
+        roles.update(_operator_roles(spec))
+    return roles
+
+
 def infer_role(pane: str, title: str) -> str:
     normalized = title or ""
     base = normalized.split("|", 1)[0].strip()
     lowered = base.lower()
+    registry_roles = _registry_roles_for_pane(pane)
     if pane.endswith(":0.0") and pane.startswith(f"{SESSION}:") and ("pm" in lowered or "产品经理" in base):
         return "pm"
     if "planner" in lowered or "规划者" in base:
@@ -66,6 +124,9 @@ def infer_role(pane: str, title: str) -> str:
         return "observer"
     if "builder" in lowered or "建设者" in base or "lab-builder" in lowered:
         return "builder"
+    for candidate in ("planner", "evaluator", "architect", "builder", "pm"):
+        if candidate in registry_roles:
+            return candidate
     if pane.startswith("solar-harness-lab:") or pane.startswith("solar-harness-multi-task:"):
         return "builder"
     if pane == f"{SESSION}:0.0":
@@ -81,21 +142,29 @@ def _planner_rank(item: dict[str, str]) -> tuple[int, str]:
     role = item["host_role"]
     pane = item["pane"]
     if role == "planner":
+        if pane.startswith(f"{SESSION}:"):
+            return (0, pane)
+        if pane.startswith("solar-harness-multi-task:"):
+            return (1, pane)
         return (0, pane)
     if role == "architect":
-        return (1, pane)
-    if role == "builder":
         return (2, pane)
+    if role == "builder":
+        return (3, pane)
     return (9, pane)
 
 
 def _evaluator_rank(item: dict[str, str]) -> tuple[int, str]:
     role = item["host_role"]
     pane = item["pane"]
     if role == "evaluator":
+        if pane.startswith(f"{SESSION}:"):
+            return (0, pane)
+        if pane.startswith("solar-harness-multi-task:"):
+            return (1, pane)
         return (0, pane)
     if role == "builder":
-        return (1, pane)
+        return (2, pane)
     return (9, pane)
 
 
@@ -114,19 +183,28 @@ def discover_role_pool(role: str) -> list[dict[str, str]]:
         pane = item["pane"]
         if not _allowed_session(pane):
             continue
+        registry_roles = _registry_roles_for_pane(pane)
         host_role = infer_role(pane, item["title"])
-        item = {"pane": pane, "title": item["title"], "host_role": host_role}
+        effective_host_role = host_role
+        if role in registry_roles:
+            effective_host_role = role
+        item = {
+            "pane": pane,
+            "title": item["title"],
+            "host_role": effective_host_role,
+            "registry_roles": sorted(registry_roles),
+        }
         if role == "pm":
-            if host_role in {"pm", "observer"}:
+            if effective_host_role in {"pm", "observer"} or role in registry_roles:
                 rows.append(item)
         elif role == "planner":
-            if host_role in {"planner", "architect", "builder"}:
+            if effective_host_role in {"planner", "architect", "builder"} or role in registry_roles:
                 rows.append(item)
         elif role == "builder":
-            if host_role == "builder":
+            if effective_host_role == "builder" or role in registry_roles:
                 rows.append(item)
         elif role == "evaluator":
-            if host_role in {"evaluator", "builder"}:
+            if effective_host_role in {"evaluator", "builder"} or role in registry_roles:
                 rows.append(item)
     if role == "planner":
         rows.sort(key=_planner_rank)