From 340cc9e85b76fc5171d3a35483dd63491f2e73e1 Mon Sep 17 00:00:00 2001 From: operator Date: Sat, 10 Oct 2026 11:59:11 -0400 Subject: [PATCH] fix(tui): harden all Box TUI tabs against nulls, boundary slices, and small screens --- bin/box-fleet-tui.py | 155 ++++++++++++++++++++--------------- bin/muse-tui.py | 87 ++++++++++++-------- tests/test_box_tui_render.py | 144 ++++++++++++++++++++++++++++++++ 3 files changed, 287 insertions(+), 99 deletions(-) diff --git a/bin/box-fleet-tui.py b/bin/box-fleet-tui.py index f5da338..90d472a 100755 --- a/bin/box-fleet-tui.py +++ b/bin/box-fleet-tui.py @@ -71,6 +71,12 @@ IterLinesFn = Callable[[Path], Iterable[str]] EPOCH = datetime(1970, 1, 1, tzinfo=timezone.utc) +def _safe_str(val: Any, default: str = "") -> str: + """Safe string converter that turns None into default.""" + return default if val is None else str(val) + + + # ===================================================================== # Default IO primitives (injectable seams for tests) # ===================================================================== @@ -1074,7 +1080,7 @@ class BoxFleetTUI: lines: List[Tuple[str, str]]) -> None: self.safe_addstr(3, 2, title, self._attr("cyan")) self.safe_addstr(4, 2, "-" * (w - 4), self._attr("dim")) - max_rows = h - 8 + max_rows = max(0, h - 8) visible = lines[self.scroll:self.scroll + max_rows] for i, (text, attr_name) in enumerate(visible): self.safe_addstr(5 + i, 2, text, self._attr(attr_name)) @@ -1093,19 +1099,23 @@ class BoxFleetTUI: % ("TIMER", "NEXT", "IN", "LAST", "AGO", "ACTIVATES"), "dim"), ] for r in rows: - missing = (r.get("next", NA) == NA and r.get("last", NA) == NA) - attr = "dim" if missing else ("yellow" if r.get("next", NA) == NA else "normal") + if not isinstance(r, dict): + continue + next_val = _safe_str(r.get("next"), NA) + last_val = _safe_str(r.get("last"), NA) + missing = (next_val == NA and last_val == NA) + attr = "dim" if missing else ("yellow" if next_val == NA else "normal") lines.append(( "%-32s %-11s %-9s %-11s %-9s %s" % ( - r.get("timer", "?")[:32], r.get("next", NA)[:11], - r.get("next_rel", NA)[:9], r.get("last", NA)[:11], - r.get("last_rel", NA)[:9], - r.get("activates", NA)[:28]), attr)) + _safe_str(r.get("timer"), "?")[:32], next_val[:11], + _safe_str(r.get("next_rel"), NA)[:9], last_val[:11], + _safe_str(r.get("last_rel"), NA)[:9], + _safe_str(r.get("activates"), NA)[:28]), attr)) if not rows: lines.append(("(No fleet timers discovered in systemd/)", "dim")) if data.get("note"): lines.append(("", "normal")) - lines.append(("note: %s" % data["note"][: w - 10], "yellow")) + lines.append(("note: %s" % _safe_str(data["note"])[: max(0, w - 10)], "yellow")) self._body(h, w, "USER TIMER STATES (NEXT/LAST)", lines) def _render_harvest(self, h: int, w: int) -> None: @@ -1116,23 +1126,30 @@ class BoxFleetTUI: % ("AGENT", "THREAD", "WATERMARK", "LAST", "AGO", "FRESH"), "dim"), ] for r in rows: - fresh = r.get("freshness", "unknown") + if not isinstance(r, dict): + continue + fresh = _safe_str(r.get("freshness"), "unknown") if fresh == "fresh": attr = "green" elif fresh == "aging": attr = "yellow" elif fresh == "stale": attr = "red" - elif r.get("watermark", NA) == NA: + elif _safe_str(r.get("watermark"), NA) == NA: attr = "dim" + else: + attr = "normal" lines.append(( "%-6s %-20s %-24s %-11s %-9s %s" % ( - r.get("agent", "?"), r.get("thread_name", "?")[:20], - r.get("watermark", NA)[:24], r.get("last", NA)[:11], - r.get("last_rel", NA)[:9], fresh), attr)) + _safe_str(r.get("agent"), "?"), _safe_str(r.get("thread_name"), "?")[:20], + _safe_str(r.get("watermark"), NA)[:24], _safe_str(r.get("last"), NA)[:11], + _safe_str(r.get("last_rel"), NA)[:9], fresh), attr)) + if not rows: + lines.append(("(No harvest watermarks found in %s)" + % WATERMARKS_FILE.name, "dim")) if data.get("note"): lines.append(("", "normal")) - lines.append(("note: %s" % data["note"][: w - 10], "yellow")) + lines.append(("note: %s" % _safe_str(data["note"])[: max(0, w - 10)], "yellow")) self._body(h, w, "HARVEST WATERMARKS PER AGENT (freshness highlighted)", lines) def _render_followups(self, h: int, w: int) -> None: @@ -1158,16 +1175,16 @@ class BoxFleetTUI: lines.append((" %-8s %d" % (agent, n), "normal")) lines.append(("", "normal")) oldest = data.get("oldest_pending") - if oldest: + if isinstance(oldest, dict): lines.append(("Oldest pending: %s %s %s sent %s (%s)" % ( - oldest.get("dm_id", "?"), oldest.get("route", "?"), - oldest.get("target", "?"), oldest.get("sent", NA), - oldest.get("sent_rel", NA)), "yellow")) + _safe_str(oldest.get("dm_id"), "?"), _safe_str(oldest.get("route"), "?"), + _safe_str(oldest.get("target"), "?"), _safe_str(oldest.get("sent"), NA), + _safe_str(oldest.get("sent_rel"), NA)), "yellow")) else: lines.append(("Oldest pending: none", "dim")) if data.get("note"): lines.append(("", "normal")) - lines.append(("note: %s" % data["note"][: w - 10], "yellow")) + lines.append(("note: %s" % _safe_str(data["note"])[: max(0, w - 10)], "yellow")) self._body(h, w, "FOLLOW-UP QUEUE COUNTS BY STATUS", lines) def _render_approvals(self, h: int, w: int) -> None: @@ -1185,14 +1202,16 @@ class BoxFleetTUI: lines.append(("%-6s %-12s %-32s %s" % ("NODE", "STATUS", "TARGET", "TITLE"), "dim")) for p in pending: + if not isinstance(p, dict): + continue lines.append(( "%-6s %-12s %-32s %s" % ( - p.get("node", "?"), p.get("status", "?")[:12], - p.get("target", "-")[:32], - p.get("title", "")[: w - 58]), "red")) + _safe_str(p.get("node"), "?"), _safe_str(p.get("status"), "?")[:12], + _safe_str(p.get("target"), "-")[:32], + _safe_str(p.get("title"), "")[: max(0, w - 58)]), "red")) if data.get("note"): lines.append(("", "normal")) - lines.append(("note: %s" % data["note"][: w - 10], "yellow")) + lines.append(("note: %s" % _safe_str(data["note"])[: max(0, w - 10)], "yellow")) self._body(h, w, "PENDING APPROVALS", lines) def _render_activity(self, h: int, w: int) -> None: @@ -1203,18 +1222,20 @@ class BoxFleetTUI: % ("AGENT", "LAST", "AGO", "SRC", "FRESH", "DETAIL"), "dim"), ] for r in rows: - fresh = r.get("freshness", "unknown") + if not isinstance(r, dict): + continue + fresh = _safe_str(r.get("freshness"), "unknown") attr = "green" if fresh == "fresh" else ( "yellow" if fresh == "aging" else ( "red" if fresh == "stale" else "dim")) lines.append(( "%-6s %-11s %-9s %-6s %-7s %s" % ( - r.get("agent", "?"), r.get("last", NA)[:11], - r.get("last_rel", NA)[:9], r.get("source", NA)[:6], - fresh, r.get("detail", NA)[: w - 48]), attr)) + _safe_str(r.get("agent"), "?"), _safe_str(r.get("last"), NA)[:11], + _safe_str(r.get("last_rel"), NA)[:9], _safe_str(r.get("source"), NA)[:6], + fresh, _safe_str(r.get("detail"), NA)[: max(0, w - 48)]), attr)) if data.get("note"): lines.append(("", "normal")) - lines.append(("note: %s" % data["note"][: w - 10], "yellow")) + lines.append(("note: %s" % _safe_str(data["note"])[: max(0, w - 10)], "yellow")) self._body(h, w, "PER-AGENT LAST ACTIVITY", lines) def _render_runtimes(self, h: int, w: int) -> None: @@ -1235,70 +1256,74 @@ class BoxFleetTUI: ] items: List[Dict[str, str]] = [] line_of_item: Dict[int, int] = {} - for agent in data.get("agents", []): + for agent in data.get("agents") or []: + if not isinstance(agent, dict): + continue live = agent.get("live") briefed = agent.get("briefed") if isinstance(briefed, bool): briefed_txt = "yes" if briefed else "NO" else: - briefed_txt = str(briefed) + briefed_txt = str(briefed) if briefed is not None else "-" attr = "dim" if not agent.get("enabled") else ( "red" if not live else ( "yellow" if briefed_txt in ("NO", "stale") else "normal")) idx = len(items) if live: - items.append({"socket": agent.get("socket", "?"), - "session": agent.get("session", "?")}) + items.append({"socket": _safe_str(agent.get("socket"), "?"), + "session": _safe_str(agent.get("session"), "?")}) line_of_item[idx] = len(lines) else: idx = -1 marker = ">" if idx == self.rt_sel else " " lines.append(( "%s%-25s %-8s %-5s %-5s %-11s %-10s %-7s %s" % ( - marker, agent.get("session", "?")[:25], - agent.get("hat", "?")[:8], + marker, _safe_str(agent.get("session"), "?")[:25], + _safe_str(agent.get("hat"), "?")[:8], "YES" if live else "NO", - agent.get("pane", NA)[:5], - agent.get("state", NA)[:11], - agent.get("mode", NA)[:10], + _safe_str(agent.get("pane"), NA)[:5], + _safe_str(agent.get("state"), NA)[:11], + _safe_str(agent.get("mode"), NA)[:10], "ALIVE" if agent.get("watcher") else "-", briefed_txt), attr)) - strays = data.get("strays", []) + strays = data.get("strays") or [] if strays: lines.append(("", "normal")) lines.append(("Stray sessions (fleet sockets, not in manifest):", "dim")) for stray in strays: + if not isinstance(stray, dict): + continue idx = len(items) - items.append({"socket": stray.get("socket", "?"), - "session": stray.get("session", "?")}) + items.append({"socket": _safe_str(stray.get("socket"), "?"), + "session": _safe_str(stray.get("session"), "?")}) line_of_item[idx] = len(lines) marker = ">" if idx == self.rt_sel else " " lines.append(( "%s%-25s %-8s %-5s %-5s %-11s %-10s %-7s %s" % ( - marker, stray.get("session", "?")[:25], + marker, _safe_str(stray.get("session"), "?")[:25], "stray", - "YES", stray.get("pane", NA)[:5], - stray.get("state", NA)[:11], "", "", - (stray.get("cmd", "") or "")[:20]), "yellow")) + "YES", _safe_str(stray.get("pane"), NA)[:5], + _safe_str(stray.get("state"), NA)[:11], "", "", + _safe_str(stray.get("cmd"))[:20]), "yellow")) lines.append(("", "normal")) lines.append(("Task queue: pending %d | claimed %s | done %d" - % (queue.get("pending", 0), claimed_txt[: w - 48], + % (queue.get("pending", 0), claimed_txt[: max(0, w - 48)], queue.get("done", 0)), "cyan")) plan_bits = [] for key in ("launch", "brief", "nudge", "failed"): names = plan.get(key, []) or [] if names: - plan_bits.append("%s: %s" % (key, ", ".join(names))) + plan_bits.append("%s: %s" % (key, ", ".join(str(n) for n in names))) lines.append(("", "normal")) lines.append(("Reconcile plan (dry-run): %s" % ("; ".join(plan_bits) or "steady state"), "cyan")) for err in data.get("plan_errors", []) or []: - lines.append(("plan error: %s" % str(err)[: w - 14], "red")) + lines.append(("plan error: %s" % str(err)[: max(0, w - 14)], "red")) if data.get("note"): lines.append(("", "normal")) - lines.append(("note: %s" % data["note"][: w - 10], "yellow")) + lines.append(("note: %s" % _safe_str(data["note"])[: max(0, w - 10)], "yellow")) self.rt_items = items if self.rt_items: self.rt_sel = max(0, min(self.rt_sel, len(self.rt_items) - 1)) @@ -1363,16 +1388,16 @@ class BoxFleetTUI: ] for agent in FLEET_AGENTS: - st = cog.get(agent, {}) - badge = st.get("badge") or st.get("status") or "UNKNOWN" + st = cog.get(agent) or {} + badge = _safe_str(st.get("badge") or st.get("status"), "UNKNOWN") lock = "LOCKED" if st.get("cognitive_lock") else "OPEN" scr = st.get("screen") or {} - url = scr.get("url") or "-" - title = scr.get("title") or "-" + url = _safe_str(scr.get("url"), "-") + title = _safe_str(scr.get("title"), "-") lock_attr = "red" if lock == "LOCKED" else "green" lines.append(( "%-6s %-16s %-8s %-24s %s" % ( - agent, badge[:16], lock, url[:24], title[:w - 60] + agent, badge[:16], lock, url[:24], title[: max(0, w - 60)] ), lock_attr )) @@ -1383,17 +1408,19 @@ class BoxFleetTUI: if not issues: lines.append(("(No recent build tickets found on tea.muse-dev.online)", "dim")) else: - for idx, iss in enumerate(issues[:8]): - inum = "#%s" % iss.get("number") - istate = iss.get("state", "").upper() - asg = (iss.get("assignee") or {}).get("username") or "unassigned" - created = (iss.get("created_at") or "")[:10] - title = iss.get("title") or "" + for idx, iss in enumerate((issues or [])[:8]): + if not isinstance(iss, dict): + continue + inum = "#%s" % _safe_str(iss.get("number"), "?") + istate = _safe_str(iss.get("state"), "").upper() + asg = _safe_str((iss.get("assignee") or {}).get("username"), "unassigned") + created = _safe_str(iss.get("created_at"))[:10] + title = _safe_str(iss.get("title"), "") marker = ">" if idx == self.work_sel else " " i_attr = "green" if istate == "CLOSED" else "yellow" lines.append(( "%s%-5s %-10s %-10s %-12s %s" % ( - marker, inum, istate, asg[:10], created, title[:w - 45] + marker, inum, istate, asg[:10], created, title[: max(0, w - 45)] ), i_attr )) @@ -1435,10 +1462,10 @@ class BoxFleetTUI: self.status_msg = "Heal @%s: %s" % (agent.upper(), "Restored" if rc == 0 else "Alerts") def _render_help(self, h: int, w: int) -> None: - modal_w = min(64, w - 6) - modal_h = 15 - top = (h - modal_h) // 2 - left = (w - modal_w) // 2 + modal_w = max(10, min(64, w - 6)) + modal_h = max(5, min(15, h - 2)) + top = max(0, (h - modal_h) // 2) + left = max(0, (w - modal_w) // 2) for y in range(top, top + modal_h): self.safe_addstr(y, left, " " * modal_w, self._attr("row_sel")) self.safe_addstr(top, left, "+" + "-" * (modal_w - 2) + "+", diff --git a/bin/muse-tui.py b/bin/muse-tui.py index 666a7dd..b933d0a 100755 --- a/bin/muse-tui.py +++ b/bin/muse-tui.py @@ -2033,8 +2033,11 @@ class MuseTUI: # Muse Conversational View (Split Sidebar + Transcript) # ----------------------------------------------------------------------- def _render_muse_view(self, y: int, x: int, h: int, w: int): - sidebar_w = min(36, max(26, w // 4)) - chat_w = w - sidebar_w - 1 + if w < 40: + sidebar_w = min(20, max(10, w // 2)) + else: + sidebar_w = min(36, max(24, w // 4)) + chat_w = max(0, w - sidebar_w - 1) is_sidebar_focused = (self.focus_pane in ("sidebar", "sidechats", "fleet") and self.editor_mode == "NORMAL") is_transcript_focused = (self.focus_pane == "transcript" and self.editor_mode == "NORMAL") @@ -2878,7 +2881,7 @@ class MuseTUI: break node = self.data.nodes[idx] is_sel = (idx == self.fleet_sel_idx) - st = statuses.get(node, {}) + st = statuses.get(node) or {} status_text = st.get("status", "ACTIVE") is_blocked = st.get("blocked", False) if is_blocked: @@ -2890,7 +2893,7 @@ class MuseTUI: tab_name = st.get("tab") or "-" # Cognitive state & Lock - cog = cog_cache.get(node, {}) + cog = cog_cache.get(node) or {} c_st = cog.get("status") or ("INPUT_WAIT" if is_blocked else ("ACTIVE" if status_text == "ACTIVE" else "IDLE")) c_locked = cog.get("cognitive_lock", False) lock_str = "LOCKED" if c_locked else "OPEN" @@ -2899,8 +2902,10 @@ class MuseTUI: # Assigned build ticket assigned = None for iss in work_issues: - if iss.get("assignee", {}).get("username") == node and iss.get("state") == "open": - assigned = f"#{iss.get('number')} {iss.get('title', '')[:28]}" + if not isinstance(iss, dict): + continue + if (iss.get("assignee") or {}).get("username") == node and iss.get("state") == "open": + assigned = f"#{iss.get('number')} {(iss.get('title') or '')[:28]}" break work_txt = assigned or tab_name @@ -2976,7 +2981,7 @@ class MuseTUI: tun = "UP" if node in ("opm", "646", "dev", "muse-main", "muse") else "DARK" tun_attr = self._attr("success") if tun == "UP" else self._attr("dim") - cog = cog_cache.get(node, {}) + cog = cog_cache.get(node) or {} c_st = cog.get("status") or "IDLE" if node == "muse-main": c_st = "HOST" @@ -2986,8 +2991,10 @@ class MuseTUI: assigned = None for iss in issues: - if iss.get("assignee", {}).get("username") == node and iss.get("state") == "open": - assigned = f"#{iss.get('number')} {iss.get('title', '')[:30]}" + if not isinstance(iss, dict): + continue + if (iss.get("assignee") or {}).get("username") == node and iss.get("state") == "open": + assigned = f"#{iss.get('number')} {(iss.get('title') or '')[:30]}" break work_txt = assigned or ("Ready for assignment" if not c_locked else "Busy in thread") @@ -3014,11 +3021,13 @@ class MuseTUI: row_y += 1 else: for iss in issues[:6]: - num = f"#{iss.get('number')}" - st = iss.get("state", "").upper() - asg = iss.get("assignee", {}).get("username", "-") if iss.get("assignee") else "-" - tit = iss.get("title", "")[:44] - lbls = ", ".join(l.get("name") for l in iss.get("labels", [])) + if not isinstance(iss, dict): + continue + num = f"#{iss.get('number') or '?'}" + st = (iss.get("state") or "").upper() + asg = (iss.get("assignee") or {}).get("username") or "-" + tit = (iss.get("title") or "")[:44] + lbls = ", ".join(((l or {}).get("name") or "") for l in (iss.get("labels") or [])) st_attr = self._attr("dim") if st == "CLOSED" else self._attr("success") self.safe_addstr(self.stdscr, row_y, x + 3, f"{num:<8}", self._attr("bold")) @@ -3041,11 +3050,13 @@ class MuseTUI: row_y += 1 else: for pr in prs[:3]: - pnum = f"#{pr.get('number')}" + if not isinstance(pr, dict): + continue + pnum = f"#{pr.get('number') or '?'}" merged = pr.get("merged", False) - p_st = "MERGED" if merged else pr.get("state", "").upper() - p_head = pr.get("head", {}).get("ref", "-")[:32] - p_tit = pr.get("title", "")[:42] + p_st = "MERGED" if merged else (pr.get("state") or "").upper() + p_head = ((pr.get("head") or {}).get("ref") or "-")[:32] + p_tit = (pr.get("title") or "")[:42] st_attr = self._attr("success") if merged else self._attr("warn_banner") self.safe_addstr(self.stdscr, row_y, x + 3, f"{pnum:<8}", self._attr("bold")) @@ -3081,17 +3092,19 @@ class MuseTUI: row_y = y + 3 for idx, it in enumerate(apps): + if not isinstance(it, dict): + continue if row_y >= y + h - 2: break - n = it.get("node", "-") - st = it.get("status", "CLEAR") + n = it.get("node") or "-" + st = it.get("status") or "CLEAR" is_sel = (idx == self.approvals_sel_idx) tgt = it.get("target") or it.get("purpose") or it.get("title") or "-" if st == "INPUT_WAIT": waits = it.get("input_waits") or [] if waits: - tgt = "; ".join(f"{w.get('task')}" for w in waits) + tgt = "; ".join(f"{(w or {}).get('task')}" for w in waits) else: tgt = "Subagent waiting for operator input" @@ -3114,7 +3127,7 @@ class MuseTUI: row_y += 1 hint_y = y + h - 1 - sel_node = apps[self.approvals_sel_idx].get("node", "").upper() if apps else "-" + sel_node = (apps[self.approvals_sel_idx].get("node") or "").upper() if apps else "-" hints = f"Selected: [{sel_node}] [j/k]: Head [1/2/3]: Allow/Always/Deny [r]: Proceed [x]: Dismiss [c]: Clear All [Enter]: Chat" self.safe_addstr(self.stdscr, hint_y, x + 1, hints[:w - 2], self._attr("dim")) @@ -3150,12 +3163,14 @@ class MuseTUI: if idx >= len(jobs): break j = jobs[idx] + if not isinstance(j, dict): + continue is_sel = (idx == self.jobs_sel_idx) - name = j.get("name", "") - agent = j.get("agent", "-") - sched = j.get("schedule", "-") - timeout = f"{j.get('timeout', '-')}s" - desc = j.get("description", "-") or "-" + name = j.get("name") or "" + agent = j.get("agent") or "-" + sched = j.get("schedule") or "-" + timeout = f"{j.get('timeout') or '-'}s" + desc = j.get("description") or "-" head = "▶ " if is_sel else " " row_attr = self._attr("selected") if is_sel else self._attr("normal") @@ -3203,7 +3218,7 @@ class MuseTUI: self.tmux_sel_idx = max(0, len(sessions) - 1) row_y = y + 3 - for idx, s in enumerate(sessions[:h - 5]): + for idx, s in enumerate(sessions[:max(0, h - 5)]): is_sel = (idx == self.tmux_sel_idx) head = "▶ " if is_sel else " " row_attr = self._attr("selected") if is_sel else self._attr("normal") @@ -3226,13 +3241,15 @@ class MuseTUI: logs = list(self.data.dm_logs_cache) row_y = y + 2 - for it in logs[:h - 4]: - l_type = it.get("type", "") - agent = it.get("agent", "") - to = it.get("to", "") - target = it.get("target", "") - ts = it.get("ts", "")[:19].replace("T", " ") - mid = it.get("id", "") + for it in logs[:max(0, h - 4)]: + if not isinstance(it, dict): + continue + l_type = it.get("type") or "" + agent = it.get("agent") or "" + to = it.get("to") or "" + target = it.get("target") or "" + ts = (it.get("ts") or "")[:19].replace("T", " ") + mid = it.get("id") or "" color = self._attr("success") if l_type == "verified" else (self._attr("user") if l_type == "sent" else self._attr("dim")) line_str = f"[{ts}] {l_type:<9} {agent} -> {to} ({target}) [{mid}]" diff --git a/tests/test_box_tui_render.py b/tests/test_box_tui_render.py index 9d3ae8e..d5da781 100644 --- a/tests/test_box_tui_render.py +++ b/tests/test_box_tui_render.py @@ -16,6 +16,10 @@ spec = importlib.util.spec_from_file_location("muse_tui", str(BIN_DIR / "muse-tu muse_tui = importlib.util.module_from_spec(spec) spec.loader.exec_module(muse_tui) +spec_fleet = importlib.util.spec_from_file_location("box_fleet_tui", str(BIN_DIR / "box-fleet-tui.py")) +box_fleet_tui = importlib.util.module_from_spec(spec_fleet) +spec_fleet.loader.exec_module(box_fleet_tui) + class TestBoxTUIHeadless(unittest.TestCase): def setUp(self): @@ -134,6 +138,146 @@ class TestBoxTUIHeadless(unittest.TestCase): app._handle_key(ord('\t')) mock_load.assert_called_with("pip", "upcoming") + def test_muse_tui_all_tabs_fuzz_dimensions_and_corrupt_data(self): + dimensions = [(8, 20), (12, 40), (24, 80), (50, 160)] + + corrupt_dataset = { + "cognitive": { + "muse": None, + "pip": {"status": None, "cognitive_lock": None, "screen": None}, + "dev": {"status": "THINKING", "cognitive_lock": True, "screen": {"url": None, "title": None}} + }, + "issues": [ + {"number": 1, "state": None, "title": None, "assignee": None, "labels": None}, + {"number": 2, "state": "open", "title": "Test", "assignee": {"username": None}, "labels": [None, {"name": None}]} + ], + "prs": [ + {"number": 1, "state": None, "merged": False, "title": None, "head": None}, + {"number": 2, "state": "open", "merged": True, "title": "PR test", "head": {"ref": None}} + ], + "approvals": [ + {"node": None, "status": None, "target": None}, + {"node": "pip", "status": "INPUT_WAIT", "input_waits": [None, {"task": None}]} + ], + "jobs": [ + {"name": None, "agent": None, "schedule": None, "timeout": None, "description": None} + ], + "tmux": ["", "invalid:session:colon"], + "dm_logs": [ + {"type": None, "agent": None, "to": None, "target": None, "ts": None, "id": None} + ] + } + + for h, w in dimensions: + stdscr = MagicMock() + stdscr.getmaxyx.return_value = (h, w) + stdscr.getch.return_value = -1 + + app = muse_tui.MuseTUI(stdscr, initial_mode="box", initial_tab="chat") + app._term_dims = (h, w) + + # Test with empty caches + app.data.cognitive_cache = {} + app.data.work_issues_cache = [] + app.data.work_prs_cache = [] + app.data.approvals_cache = [] + app.data.jobs_cache = [] + app.data.dm_logs_cache = [] + + content_h = max(1, h - 4) + for tab_idx in range(8): + app.box_tab = tab_idx + if tab_idx == 0: + app._render_muse_view(2, 0, content_h, w) + elif tab_idx == 1: + app._render_fleet_table(2, 0, content_h, w) + elif tab_idx == 2: + app._render_approvals_view(2, 0, content_h, w) + elif tab_idx == 3: + app._render_jobs_view(2, 0, content_h, w) + elif tab_idx == 4: + with patch("subprocess.run") as mock_sub: + mock_sub.return_value = MagicMock(returncode=0, stdout="s1: 1 windows\n") + app._render_tmux_view(2, 0, content_h, w) + elif tab_idx == 5: + app._render_dm_logs_view(2, 0, content_h, w) + elif tab_idx == 6: + app._render_ssh_view(2, 0, content_h, w) + elif tab_idx == 7: + app._render_work_view(2, 0, content_h, w) + + # Test with corrupt / null caches + app.data.cognitive_cache = corrupt_dataset["cognitive"] + app.data.work_issues_cache = corrupt_dataset["issues"] + app.data.work_prs_cache = corrupt_dataset["prs"] + app.data.approvals_cache = corrupt_dataset["approvals"] + app.data.jobs_cache = corrupt_dataset["jobs"] + app.data.dm_logs_cache = corrupt_dataset["dm_logs"] + + for tab_idx in range(8): + app.box_tab = tab_idx + if tab_idx == 0: + app._render_muse_view(2, 0, content_h, w) + elif tab_idx == 1: + app._render_fleet_table(2, 0, content_h, w) + elif tab_idx == 2: + app._render_approvals_view(2, 0, content_h, w) + elif tab_idx == 3: + app._render_jobs_view(2, 0, content_h, w) + elif tab_idx == 4: + with patch("subprocess.run") as mock_sub: + mock_sub.return_value = MagicMock(returncode=0, stdout="s1: 1 windows\n") + app._render_tmux_view(2, 0, content_h, w) + elif tab_idx == 5: + app._render_dm_logs_view(2, 0, content_h, w) + elif tab_idx == 6: + app._render_ssh_view(2, 0, content_h, w) + elif tab_idx == 7: + app._render_work_view(2, 0, content_h, w) + + def test_box_fleet_tui_all_tabs_fuzz_dimensions_and_corrupt_data(self): + dimensions = [(8, 20), (12, 40), (24, 80), (50, 160)] + for h, w in dimensions: + stdscr = MagicMock() + stdscr.getmaxyx.return_value = (h, w) + stdscr.getch.return_value = -1 + + fleet_app = box_fleet_tui.BoxFleetTUI(stdscr) + + # 1. Empty snapshot + fleet_app.snapshot = {} + renderers = [fleet_app._render_timers, fleet_app._render_harvest, + fleet_app._render_followups, fleet_app._render_approvals, + fleet_app._render_activity, fleet_app._render_runtimes, + fleet_app._render_work] + + for tab_idx, renderer in enumerate(renderers): + fleet_app.current_tab = tab_idx + renderer(h, w) + + # 2. Corrupted / null snapshot + fleet_app.snapshot = { + "timers": {"rows": [{"timer": None, "next": None, "next_rel": None, "last": None, "last_rel": None, "activates": None}]}, + "harvest": {"rows": [{"freshness": None, "agent": None, "thread": None, "watermark": None, "last": None, "ago": None}]}, + "followups": {"counts": {"pending": 2}, "by_recipient": {"pip": 1}, "oldest_pending": None}, + "approvals": {"total_pending": 1, "checked": 1, "unreachable": [], "pending": [{"node": None, "status": None, "target": None, "title": None}]}, + "activity": {"rows": [{"freshness": None, "agent": None, "last": None, "last_rel": None, "source": None, "detail": None}]}, + "runtimes": {"agents": [{"live": True, "briefed": None, "enabled": True, "session": None, "hat": None, "pane": None, "state": None, "mode": None, "watcher": None}], "queue": {}, "plan": {}}, + "work": { + "cognitive": {"dev": None}, + "issues": [{"number": None, "state": None, "assignee": None, "created_at": None, "title": None}], + "loopback_active": True + } + } + + for tab_idx, renderer in enumerate(renderers): + fleet_app.current_tab = tab_idx + renderer(h, w) + + fleet_app._render_header(w) + fleet_app._render_footer(h, w) + fleet_app._render_help(h, w) + if __name__ == "__main__": unittest.main()