fix: repair pre-existing watcher_pane tests

.console/log.md

@@ -1,5 +1,32 @@
 # Log
 
+## 2026-05-26 — Repair pre-existing watcher_pane tests
+
+Fixed 9 stale/regressed failures in tests/test_watcher_pane.py.
+
+- Banner-message tests (healthy/switchboard/gate/queue/info): STALE tests. The
+  banner copy was intentionally title-cased and reworded ("All Systems Nominal",
+  "SwitchBoard Offline", "Global Gate at Cap", "Queue Depth", "Stabilizing").
+  Updated assertions to the current strings.
+- test_exec_budget_reads_usage: STALE test — it never isolated _resource_gate(),
+  which reads the real on-disk OC config; gate values shadow the env caps the
+  test sets, so daily_cap came back as the config's value. Added a monkeypatch
+  stubbing _resource_gate -> {} so the env-override path is the one exercised.
+- test_critical_sorts_before_warning_sorts_before_info: STALE premise — the
+  "just started" INFO banner is intentionally PINNED to the front of the cycle,
+  so a linear crit<warn<info ordering can't hold within 30s of launch. Split
+  into test_critical_sorts_before_warning (pure severity order, started_at=0)
+  and test_just_started_info_is_pinned_to_front (documents the pinning).
+- test_overflow_proportional / test_collapsed_during_overflow: REAL code bug.
+  _allocate_section_rows did no overflow scaling and over-allocated far past the
+  available rows (42/31 vs cap 10). Added proportional down-scaling on overflow
+  with a min(3, natural) floor; natural-fit, size_mult, collapsed, and empty
+  paths preserved.
+
+tests/test_watcher_pane.py: 27 passed. Full suite: 132 passed, 3 failed —
+the 3 are pre-existing cxrp schema_version 0.2-vs-0.3 mismatches in
+tests/test_cxrp_capture.py, unrelated to this change. Custodian clean.
+
 ## 2026-05-22 — Rename ContextLifecycleProtocol → ContextLifecycle
 
 Hard cutover. Renamed profile file from `contextlifecycleprotocol.yaml` to `contextlifecycle.yaml`. Updated all references in config, git_watcher, and platform.yaml.

src/operator_console/watcher_status_pane.py

@@ -1169,28 +1169,98 @@ def _allocate_section_rows(
 ) -> list[int]:
     """Decide how many on-screen rows each section gets.
 
-    Each section gets its full natural height (collapsed = 1 row,
-    expanded = ``len(lines) * size_mult.get(id, 1.0)``, rounded up).
-    No proportional scaling on overflow — the render loop truncates
-    at ``middle_bottom``, so sections later in the list lose visibility
-    when earlier expanded sections use up the available space. The
-    operator manages overflow by collapsing sections, not by resizing.
+    Each section's *natural* height is collapsed = 1 row, expanded =
+    ``len(lines) * size_mult.get(id, 1.0)`` (rounded up), and empty = 0.
+
+    When the natural heights fit within ``available_rows`` they are
+    returned unchanged. On overflow the expanded (non-collapsed,
+    non-empty) sections are scaled down proportionally to their natural
+    height so the total fits, with a per-section floor of ``min(3,
+    natural)`` rows so every visible section keeps a usable slice.
+    Collapsed sections always keep their single row; empty sections
+    always get zero.
     """
     collapsed = collapsed or {}
     size_mult = size_mult or {}
     if available_rows <= 0 or not sections:
         return [0] * len(sections)
-    from math import ceil
-    out: list[int] = []
+    from math import ceil, floor
+
+    natural: list[int] = []
     for s in sections:
         if not s["lines"]:
-            out.append(0)
-            continue
-        if collapsed.get(s["id"], False):
-            out.append(1)
-            continue
-        mult = size_mult.get(s["id"], 1.0)
-        out.append(max(1, ceil(len(s["lines"]) * mult)))
+            natural.append(0)
+        elif collapsed.get(s["id"], False):
+            natural.append(1)
+        else:
+            mult = size_mult.get(s["id"], 1.0)
+            natural.append(max(1, ceil(len(s["lines"]) * mult)))
+
+    if sum(natural) <= available_rows:
+        return natural
+
+    # Overflow: fixed rows = collapsed (1 each) + empty (0). Distribute the
+    # remaining budget across the expandable sections proportionally.
+    expand_idx = [
+        i for i, s in enumerate(sections)
+        if s["lines"] and not collapsed.get(s["id"], False)
+    ]
+    out = list(natural)
+    fixed = sum(out[i] for i in range(len(sections)) if i not in expand_idx)
+    budget = available_rows - fixed
+    if budget <= 0 or not expand_idx:
+        # No room left for expandable content; give each at most 1 row,
+        # trimming from the end until we fit.
+        for i in expand_idx:
+            out[i] = 1
+        while sum(out) > available_rows and expand_idx:
+            out[expand_idx.pop()] = 0
+        return out
+
+    total_natural = sum(natural[i] for i in expand_idx)
+    floors = {i: min(3, natural[i]) for i in expand_idx}
+    if sum(floors.values()) >= budget:
+        # Even the floors don't fit — hand out the budget one row at a
+        # time, largest-natural first, so no section is starved unfairly.
+        for i in expand_idx:
+            out[i] = 0
+        order = sorted(expand_idx, key=lambda i: natural[i], reverse=True)
+        remaining = budget
+        while remaining > 0:
+            progressed = False
+            for i in order:
+                if remaining <= 0:
+                    break
+                if out[i] < natural[i]:
+                    out[i] += 1
+                    remaining -= 1
+                    progressed = True
+            if not progressed:
+                break
+        return out
+
+    # Proportional share above the floor.
+    extra = budget - sum(floors.values())
+    weights = {i: natural[i] / total_natural for i in expand_idx}
+    alloc = {i: floors[i] + floor(extra * weights[i]) for i in expand_idx}
+    for i in expand_idx:
+        alloc[i] = min(alloc[i], natural[i])
+    # Distribute any leftover rows (from flooring) to largest natural first.
+    leftover = budget - sum(alloc.values())
+    order = sorted(expand_idx, key=lambda i: natural[i], reverse=True)
+    while leftover > 0:
+        progressed = False
+        for i in order:
+            if leftover <= 0:
+                break
+            if alloc[i] < natural[i]:
+                alloc[i] += 1
+                leftover -= 1
+                progressed = True
+        if not progressed:
+            break
+    for i in expand_idx:
+        out[i] = alloc[i]
     return out
 
 

tests/test_watcher_pane.py

@@ -16,6 +16,10 @@ def test_exec_budget_reads_usage(self, tmp_path, monkeypatch):
             json.dumps({"hourly_exec_count": 7, "daily_exec_count": 33})
         )
         monkeypatch.setattr(wsp, "_USAGE_PATH", target / "usage.json")
+        # Isolate from any on-disk resource_gate so the env-override path
+        # (the behaviour this test verifies) is the one actually exercised;
+        # gate values take precedence over env when present.
+        monkeypatch.setattr(wsp, "_resource_gate", lambda: {})
         monkeypatch.setenv("OPERATIONS_CENTER_MAX_EXEC_PER_HOUR", "12")
         monkeypatch.setenv("OPERATIONS_CENTER_MAX_EXEC_PER_DAY", "60")
         b = wsp._exec_budget()
@@ -238,7 +242,7 @@ def test_healthy_when_no_conditions(self, monkeypatch):
         result = wsp._banner_conditions(self._data(), started_at=0)
         assert len(result) == 1
         assert result[0][0] == wsp.BANNER_HEALTHY
-        assert "nominal" in result[0][1]
+        assert "Nominal" in result[0][1]
 
     def test_critical_stall_takes_precedence(self, monkeypatch):
         from operator_console import watcher_status_pane as wsp
@@ -251,7 +255,7 @@ def test_switchboard_down_is_critical(self, monkeypatch):
         from operator_console import watcher_status_pane as wsp
         monkeypatch.setattr(wsp, "_stale_heartbeat_roles", lambda: [])
         result = wsp._banner_conditions(self._data(sb=False), started_at=0)
-        assert any("SwitchBoard offline" in m for _, m in result)
+        assert any("SwitchBoard Offline" in m for _, m in result)
 
     def test_resource_gate_at_cap_is_critical(self, monkeypatch):
         from operator_console import watcher_status_pane as wsp
@@ -261,7 +265,7 @@ def test_resource_gate_at_cap_is_critical(self, monkeypatch):
             backend_usage={"team_executor": {"in_flight": 2}},
         )
         result = wsp._banner_conditions(d, started_at=0)
-        assert any(s == wsp.BANNER_CRIT and "Global gate" in m
+        assert any(s == wsp.BANNER_CRIT and "Global Gate" in m
                    for s, m in result)
 
     def test_backend_saturation_is_warning(self, monkeypatch):
@@ -279,30 +283,45 @@ def test_queue_overflow_is_warning(self, monkeypatch):
         monkeypatch.setattr(wsp, "_stale_heartbeat_roles", lambda: [])
         d = self._data(queue=[{"task_type": "goal"}] * 12)
         result = wsp._banner_conditions(d, started_at=0)
-        assert any(s == wsp.BANNER_WARN and "Queue depth" in m
+        assert any(s == wsp.BANNER_WARN and "Queue Depth" in m
                    for s, m in result)
 
     def test_info_banner_during_first_30_seconds(self, monkeypatch):
         import time
         from operator_console import watcher_status_pane as wsp
         monkeypatch.setattr(wsp, "_stale_heartbeat_roles", lambda: [])
         result = wsp._banner_conditions(self._data(), started_at=time.time())
-        assert any(s == wsp.BANNER_INFO and "stabilizing" in m
+        assert any(s == wsp.BANNER_INFO and "Stabilizing" in m
                    for s, m in result)
 
-    def test_critical_sorts_before_warning_sorts_before_info(
-        self, monkeypatch,
-    ):
-        import time
+    def test_critical_sorts_before_warning(self, monkeypatch):
         from operator_console import watcher_status_pane as wsp
         monkeypatch.setattr(wsp, "_stale_heartbeat_roles", lambda: ["goal"])
         d = self._data(
             queue=[{"task_type": "goal"}] * 12,
         )
-        result = wsp._banner_conditions(d, started_at=time.time())
+        # started_at far in the past: no "just started" INFO is injected,
+        # so we observe the pure severity-sorted order.
+        result = wsp._banner_conditions(d, started_at=0)
         levels = [s for s, _ in result]
-        # CRIT (stall) should come before WARN (queue depth) before INFO.
+        # CRIT (stall) should come before WARN (queue depth).
         crit_idx = levels.index(wsp.BANNER_CRIT)
         warn_idx = levels.index(wsp.BANNER_WARN)
-        info_idx = levels.index(wsp.BANNER_INFO)
-        assert crit_idx < warn_idx < info_idx
+        assert crit_idx < warn_idx
+
+    def test_just_started_info_is_pinned_to_front(self, monkeypatch):
+        import time
+        from operator_console import watcher_status_pane as wsp
+        monkeypatch.setattr(wsp, "_stale_heartbeat_roles", lambda: ["goal"])
+        d = self._data(
+            queue=[{"task_type": "goal"}] * 12,
+        )
+        # Within the first 30s the "stabilizing" INFO is pinned to the
+        # front of the cycle so operators see it first on launch, even
+        # when CRITICAL conditions are also active.
+        result = wsp._banner_conditions(d, started_at=time.time())
+        assert result[0][0] == wsp.BANNER_INFO
+        assert "Stabilizing" in result[0][1]
+        # The severity ordering still holds for the remaining conditions.
+        rest = [s for s, _ in result[1:]]
+        assert rest.index(wsp.BANNER_CRIT) < rest.index(wsp.BANNER_WARN)