Terminal dashboard for monitoring Claude Code and Claude API health.
Initial commit: claude-status TUI dashboard
Textual-based TUI that monitors Claude Code and Claude API health
from status.claude.com, with NDJSON history and cmux notifications.
Matthew J. Schultz
(Apr 28, 2026, 6:01 PM EDT)
eb0de34e
+2097
+84
CLAUDE.md
+84
CLAUDE.md
···
1
+
# CLAUDE.md — claude-status
2
+
3
+
TUI dashboard that monitors Claude Code and Claude API health from the official statuspage.
4
+
5
+
## Run
6
+
7
+
```bash
8
+
uv run claude_status.py
9
+
# or directly (shebang handles uv invocation):
10
+
./claude_status.py
11
+
```
12
+
13
+
No install step — `uv` fetches dependencies (`textual==8.2.3`, `certifi`) on first run.
14
+
15
+
Press `q` to quit.
16
+
17
+
## Tests
18
+
19
+
```bash
20
+
uv run --with pytest --with "textual==8.2.3" --with certifi pytest tests/ -q
21
+
```
22
+
23
+
## Project Structure
24
+
25
+
| File | Purpose |
26
+
|------|---------|
27
+
| `claude_status.py` | Entry point. PEP 723 inline metadata for `uv run`. |
28
+
| `app.py` | Textual `ClaudeStatusApp`. Layout, polling loop, UI refresh. |
29
+
| `poller.py` | Fetches `/api/v2/components.json` + `/api/v2/incidents/unresolved.json`. Returns `FetchResult`. |
30
+
| `history.py` | NDJSON read/write at `~/.local/share/claude-status/history.json`. `PollRecord` dataclass. |
31
+
| `notifier.py` | Detects operational ↔ degraded transitions. Fires `cmux notify` if `CMUX_WORKSPACE_ID` is set. |
32
+
| `tests/` | pytest unit tests for poller, history, app logic, and notifier. |
33
+
34
+
## Architecture
35
+
36
+
- **Polling**: `_poll_worker` runs in a background thread (`@work(thread=True, exit_on_error=False)`) every 60 seconds and on mount.
37
+
- **Thread safety**: Results are dispatched back to the main thread via `self.call_from_thread(self._handle_result, result)`.
38
+
- **Persistence**: Every successful poll appends a `PollRecord` to NDJSON history. History is loaded on startup and seeded into the in-memory list.
39
+
- **Bar chart**: `ComponentRow._redraw()` renders one `█` per historical record (last N to fit width). Colors encode status.
40
+
- **SSL**: Uses `certifi` CA bundle via `ssl.create_default_context(cafile=certifi.where())` — required on macOS.
41
+
42
+
## Statuspage Component IDs
43
+
44
+
```python
45
+
CLAUDE_CODE_ID = "yyzkbfz2thpt" # "Claude Code" on status.claude.com
46
+
CLAUDE_API_ID = "k8w3r06qmzrp" # "Claude API (api.anthropic.com)"
47
+
```
48
+
49
+
These are stable IDs from the statuspage API. If components ever stop returning data, verify against:
50
+
```
51
+
curl https://status.claude.com/api/v2/components.json | python -m json.tool | grep -A2 '"name"'
52
+
```
53
+
54
+
## Known Gotchas
55
+
56
+
### `shutil` vs `os` for terminal size
57
+
`shutil.get_terminal_size(fallback=(80, 24))` accepts a `fallback` kwarg. `os.get_terminal_size()` does NOT — it raises `TypeError: posix.get_terminal_size() takes no keyword arguments`.
58
+
59
+
The fallback path in `ComponentRow._redraw()` is triggered when `self.size.width == 0`, which happens during `on_mount` before Textual has laid out the widget. This only occurs when history exists on startup (causing `_refresh_rows()` to run immediately on mount). **Always use `shutil.get_terminal_size`** for the fallback.
60
+
61
+
### History file format
62
+
`~/.local/share/claude-status/history.json` is NDJSON (one JSON object per line). Corrupt lines are silently skipped — the file is not all-or-nothing.
63
+
64
+
### Partial fetch failure handling
65
+
`fetch_status()` uses two separate `try` blocks:
66
+
1. Fetch components → build `PollRecord`. If this fails, return `FetchResult(record=None, error=...)`.
67
+
2. Fetch incidents. If this fails, return `FetchResult(record=record, incidents=[], error=...)`.
68
+
69
+
This means a valid `record` is preserved even when the incidents endpoint is unreachable. `_handle_result` in `app.py` checks `if result.record is None` (not `if result.error`) — don't change this.
70
+
71
+
### cmux notifications
72
+
`notifier.py` only calls `cmux notify` when `CMUX_WORKSPACE_ID` is set in the environment. Transitions from/to `"unknown"` are suppressed (polling failures don't trigger alerts).
73
+
74
+
### Textual widget sizing
75
+
`ComponentRow` has `height: 5` in CSS (label line + bar line + footer line + 2 border lines). If you add content rows, bump this.
76
+
77
+
## Status Values
78
+
79
+
From the statuspage API:
80
+
- `operational`
81
+
- `degraded_performance`
82
+
- `partial_outage`
83
+
- `major_outage`
84
+
- `unknown` (synthesized locally when component is missing from API response)
+30
README.md
+30
README.md
···
1
+
# claude-status
2
+
3
+
Terminal dashboard for monitoring Claude Code and Claude API health.
4
+
5
+
## Run
6
+
7
+
```bash
8
+
uv run claude_status.py
9
+
```
10
+
11
+
Or directly (shebang handles it):
12
+
13
+
```bash
14
+
./claude_status.py
15
+
```
16
+
17
+
No install step needed — `uv` fetches dependencies on first run.
18
+
19
+
Press `q` to quit.
20
+
21
+
## What it shows
22
+
23
+
- **Claude Code** and **Claude API** status (Operational / Degraded / Outage)
24
+
- Session-accumulated bar chart — one bar per 60s poll, persisted across restarts
25
+
- Active incident banner when the statuspage reports an open incident
26
+
- cmux notification on status change (when running inside cmux)
27
+
28
+
## Data
29
+
30
+
Poll history stored at `~/.local/share/claude-status/history.json` (NDJSON, one record per line).
+236
app.py
+236
app.py
···
1
+
import os
2
+
import shutil
3
+
from datetime import datetime, timezone
4
+
5
+
from textual.app import App, ComposeResult
6
+
from textual.widgets import Static
7
+
from textual import work
8
+
9
+
from history import PollRecord, load_history, append_record
10
+
from poller import fetch_status, FetchResult, Incident
11
+
from notifier import check_and_notify
12
+
13
+
POLL_INTERVAL = 60
14
+
15
+
_STATUS_COLORS = {
16
+
"operational": "green",
17
+
"degraded_performance": "orange3",
18
+
"partial_outage": "red",
19
+
"major_outage": "bold red",
20
+
"unknown": "dim",
21
+
}
22
+
23
+
_STATUS_LABELS = {
24
+
"operational": "Operational",
25
+
"degraded_performance": "Degraded Performance",
26
+
"partial_outage": "Partial Outage",
27
+
"major_outage": "Major Outage",
28
+
"unknown": "Unknown",
29
+
}
30
+
31
+
_BAR_CHARS: dict[str, tuple[str, str]] = {
32
+
"operational": ("▉", "green"),
33
+
"degraded_performance": ("▉", "orange3"),
34
+
"partial_outage": ("▉", "red"),
35
+
"major_outage": ("▉", "bold red"),
36
+
"unknown": ("░", "dim"),
37
+
}
38
+
39
+
40
+
def uptime_percent(history: list[PollRecord], field: str) -> float:
41
+
if not history:
42
+
return 0.0
43
+
operational = sum(1 for r in history if getattr(r, field) == "operational")
44
+
return (operational / len(history)) * 100.0
45
+
46
+
47
+
def render_bar(history: list[PollRecord], field: str, width: int) -> str:
48
+
records = history[-width:] if len(history) > width else history
49
+
parts = []
50
+
for r in records:
51
+
status = getattr(r, field)
52
+
char, color = _BAR_CHARS.get(status, ("░", "dim"))
53
+
parts.append(f"[{color}]{char}[/{color}]")
54
+
return "".join(parts)
55
+
56
+
57
+
def _format_age(created_at: str) -> str:
58
+
try:
59
+
ts = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
60
+
delta = max(0, int((datetime.now(timezone.utc) - ts).total_seconds()))
61
+
if delta < 60:
62
+
return f"{delta}s ago"
63
+
elif delta < 3600:
64
+
return f"{delta // 60} min ago"
65
+
else:
66
+
return f"{delta // 3600} hr ago"
67
+
except Exception:
68
+
return "unknown age"
69
+
70
+
71
+
class ComponentRow(Static):
72
+
def __init__(self, label: str, field: str, **kwargs) -> None:
73
+
super().__init__("Loading...", **kwargs)
74
+
self._label = label
75
+
self._field = field
76
+
self._history: list[PollRecord] = []
77
+
self._current_status: str = "unknown"
78
+
79
+
def update_data(self, history: list[PollRecord], current_status: str) -> None:
80
+
self._history = history
81
+
self._current_status = current_status
82
+
self._redraw()
83
+
84
+
def _redraw(self) -> None:
85
+
width = self.size.width - 4 if self.size.width > 4 else shutil.get_terminal_size(fallback=(80, 24)).columns - 4
86
+
color = _STATUS_COLORS.get(self._current_status, "dim")
87
+
label_str = _STATUS_LABELS.get(self._current_status, "Unknown")
88
+
pct = uptime_percent(self._history, self._field)
89
+
bars = render_bar(self._history, self._field, width)
90
+
91
+
badge = f"[{color}]● {label_str}[/{color}]"
92
+
padding = max(1, width - len(self._label) - len(label_str) - 3)
93
+
top = f"[bold]{self._label}[/bold]{' ' * padding}{badge}"
94
+
95
+
dash = max(1, (width - 22) // 2)
96
+
footer = f"[dim]Session start {'─' * dash} {pct:.1f}% uptime {'─' * dash} Now[/dim]"
97
+
98
+
self.update(f"{top}\n{bars}\n{footer}")
99
+
100
+
101
+
class IncidentBanner(Static):
102
+
def __init__(self, **kwargs) -> None:
103
+
super().__init__("", **kwargs)
104
+
105
+
def update_incidents(self, incidents: list[Incident]) -> None:
106
+
if not incidents:
107
+
self.update("")
108
+
self.add_class("hidden")
109
+
return
110
+
self.remove_class("hidden")
111
+
lines = []
112
+
for incident in incidents:
113
+
age = _format_age(incident.created_at)
114
+
lines.append(f"[yellow]⚠[/yellow] [bold]{incident.name}[/bold]")
115
+
lines.append(f" [dim]{incident.status.capitalize()} · {age}[/dim]")
116
+
self.update("\n".join(lines))
117
+
118
+
119
+
class StatusFooter(Static):
120
+
def __init__(self, **kwargs) -> None:
121
+
super().__init__("Initializing...", **kwargs)
122
+
self._last_updated: str = "—"
123
+
self._next_poll: int = POLL_INTERVAL
124
+
self._error: str | None = None
125
+
126
+
def on_mount(self) -> None:
127
+
self.set_interval(1, self._tick)
128
+
129
+
def _tick(self) -> None:
130
+
if self._next_poll > 0:
131
+
self._next_poll -= 1
132
+
self._redraw()
133
+
134
+
def mark_polled(self, error: str | None = None) -> None:
135
+
self._last_updated = datetime.now().strftime("%H:%M:%S")
136
+
self._next_poll = POLL_INTERVAL
137
+
self._error = error
138
+
self._redraw()
139
+
140
+
def _redraw(self) -> None:
141
+
if self._error:
142
+
status = f"[red]Fetch failed: {self._error[:50]}[/red]"
143
+
else:
144
+
status = f"Last updated: [bold]{self._last_updated}[/bold]"
145
+
self.update(f"{status} · Next poll: {self._next_poll}s")
146
+
147
+
148
+
class ClaudeStatusApp(App):
149
+
TITLE = "Claude Status"
150
+
CSS = """
151
+
Screen {
152
+
layout: vertical;
153
+
background: $surface;
154
+
}
155
+
ComponentRow {
156
+
height: 5;
157
+
border: solid $panel-lighten-1;
158
+
padding: 0;
159
+
margin: 0;
160
+
}
161
+
IncidentBanner {
162
+
height: auto;
163
+
background: $warning 15%;
164
+
padding: 0 1;
165
+
margin: 0;
166
+
}
167
+
IncidentBanner.hidden {
168
+
display: none;
169
+
}
170
+
StatusFooter {
171
+
height: 1;
172
+
dock: bottom;
173
+
background: $panel;
174
+
padding: 0 1;
175
+
}
176
+
"""
177
+
178
+
BINDINGS = [("q", "quit", "Quit")]
179
+
180
+
def __init__(self) -> None:
181
+
super().__init__()
182
+
self._history: list[PollRecord] = []
183
+
self._prev_claude_code: str = "unknown"
184
+
self._prev_claude_api: str = "unknown"
185
+
186
+
def compose(self) -> ComposeResult:
187
+
yield ComponentRow("Claude Code", "claude_code", id="row-claude-code")
188
+
yield ComponentRow("Claude API (api.anthropic.com)", "claude_api", id="row-claude-api")
189
+
yield IncidentBanner(id="incident-banner")
190
+
yield StatusFooter(id="status-footer")
191
+
192
+
def on_mount(self) -> None:
193
+
self._history = load_history()
194
+
if self._history:
195
+
latest = self._history[-1]
196
+
self._prev_claude_code = latest.claude_code
197
+
self._prev_claude_api = latest.claude_api
198
+
self._refresh_rows()
199
+
self._poll_worker()
200
+
self.set_interval(POLL_INTERVAL, self._poll_worker)
201
+
202
+
def _refresh_rows(self) -> None:
203
+
if not self._history:
204
+
return
205
+
latest = self._history[-1]
206
+
self.query_one("#row-claude-code", ComponentRow).update_data(
207
+
self._history, latest.claude_code
208
+
)
209
+
self.query_one("#row-claude-api", ComponentRow).update_data(
210
+
self._history, latest.claude_api
211
+
)
212
+
213
+
@work(thread=True, exit_on_error=False)
214
+
def _poll_worker(self) -> None:
215
+
result = fetch_status()
216
+
self.call_from_thread(self._handle_result, result)
217
+
218
+
def _handle_result(self, result: FetchResult) -> None:
219
+
footer = self.query_one("#status-footer", StatusFooter)
220
+
221
+
if result.record is None:
222
+
footer.mark_polled(error=result.error or "unknown error")
223
+
return
224
+
225
+
record = result.record
226
+
append_record(record)
227
+
self._history.append(record)
228
+
229
+
check_and_notify("Claude Code", self._prev_claude_code, record.claude_code)
230
+
check_and_notify("Claude API", self._prev_claude_api, record.claude_api)
231
+
self._prev_claude_code = record.claude_code
232
+
self._prev_claude_api = record.claude_api
233
+
234
+
self._refresh_rows()
235
+
self.query_one("#incident-banner", IncidentBanner).update_incidents(result.incidents)
236
+
footer.mark_polled(error=result.error)
+9
claude_status.py
+9
claude_status.py
+1069
docs/superpowers/plans/2026-04-15-claude-status-tui.md
+1069
docs/superpowers/plans/2026-04-15-claude-status-tui.md
···
1
+
# Claude Status TUI Implementation Plan
2
+
3
+
> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
4
+
5
+
**Goal:** Build a Python/Textual TUI that polls the Claude statuspage API every 60s, displays a session-accumulated bar chart + current status badge for Claude Code and Claude API, shows active incident banners, and fires cmux notifications on status changes.
6
+
7
+
**Architecture:** Four focused modules (`history`, `poller`, `notifier`, `app`) keep data persistence, API fetching, alerting, and UI strictly separated. The Textual app wires them together: on mount it loads persisted history, immediately polls, then polls every 60s via a thread worker. Each poll result is appended to an NDJSON file so history survives restarts.
8
+
9
+
**Tech Stack:** Python 3.12, Textual 8.2.3, pytest, stdlib only for HTTP (`urllib.request`)
10
+
11
+
---
12
+
13
+
## File Structure
14
+
15
+
```
16
+
~/mjs/claude-status/
17
+
├── claude_status.py # entry point: #!/usr/bin/env python3.12
18
+
├── app.py # Textual App + all widgets
19
+
├── poller.py # API fetch + data structures
20
+
├── history.py # PollRecord dataclass + NDJSON persistence
21
+
├── notifier.py # cmux notify + transition detection
22
+
├── requirements.txt # textual==8.2.3
23
+
├── tests/
24
+
│ ├── __init__.py
25
+
│ ├── test_history.py
26
+
│ ├── test_poller.py
27
+
│ ├── test_notifier.py
28
+
│ └── test_app_logic.py
29
+
└── docs/
30
+
└── superpowers/
31
+
├── specs/2026-04-15-claude-status-tui-design.md
32
+
└── plans/2026-04-15-claude-status-tui.md
33
+
```
34
+
35
+
**Responsibility of each file:**
36
+
- `history.py` — `PollRecord` dataclass, `StatusValue` type, `load_history()`, `append_record()`. Zero UI, zero network.
37
+
- `poller.py` — `Incident` dataclass, `FetchResult` dataclass, `fetch_status()`. Zero UI, zero disk I/O.
38
+
- `notifier.py` — `detect_transition()`, `notify_cmux()`, `check_and_notify()`. Zero UI, zero network.
39
+
- `app.py` — `ComponentRow`, `IncidentBanner`, `StatusFooter`, `ClaudeStatusApp`. Imports all three modules, owns the poll loop and UI state.
40
+
- `claude_status.py` — one line: `ClaudeStatusApp().run()`.
41
+
42
+
---
43
+
44
+
## Task 1: Project Setup
45
+
46
+
**Files:**
47
+
- Create: `requirements.txt`
48
+
- Create: `tests/__init__.py`
49
+
- Create: `pytest.ini`
50
+
51
+
- [ ] **Step 1: Write requirements.txt**
52
+
53
+
```
54
+
textual==8.2.3
55
+
pytest==8.3.5
56
+
```
57
+
58
+
- [ ] **Step 2: Write pytest.ini**
59
+
60
+
```ini
61
+
[pytest]
62
+
testpaths = tests
63
+
```
64
+
65
+
- [ ] **Step 3: Create tests/__init__.py**
66
+
67
+
```python
68
+
```
69
+
70
+
(Empty file — makes `tests/` a package so imports resolve correctly.)
71
+
72
+
- [ ] **Step 4: Install dependencies**
73
+
74
+
Run:
75
+
```bash
76
+
pip3.12 install textual==8.2.3 pytest==8.3.5
77
+
```
78
+
79
+
Expected: both packages install without error.
80
+
81
+
- [ ] **Step 5: Verify Python and Textual available**
82
+
83
+
Run:
84
+
```bash
85
+
python3.12 -c "import textual; print(textual.__version__)"
86
+
```
87
+
88
+
Expected: `8.2.3`
89
+
90
+
- [ ] **Step 6: Commit**
91
+
92
+
```bash
93
+
cd ~/mjs/claude-status
94
+
git add requirements.txt pytest.ini tests/__init__.py
95
+
git commit -m "chore: project setup, deps, pytest config"
96
+
```
97
+
98
+
---
99
+
100
+
## Task 2: History Module
101
+
102
+
**Files:**
103
+
- Create: `history.py`
104
+
- Create: `tests/test_history.py`
105
+
106
+
- [ ] **Step 1: Write the failing tests**
107
+
108
+
Create `tests/test_history.py`:
109
+
110
+
```python
111
+
import json
112
+
import pytest
113
+
from pathlib import Path
114
+
from unittest.mock import patch
115
+
from history import PollRecord, load_history, append_record
116
+
117
+
118
+
def test_load_history_returns_empty_when_file_missing(tmp_path):
119
+
with patch("history.HISTORY_FILE", tmp_path / "history.json"), \
120
+
patch("history.HISTORY_DIR", tmp_path):
121
+
assert load_history() == []
122
+
123
+
124
+
def test_append_and_load_round_trips(tmp_path):
125
+
record = PollRecord(ts=1713196800, claude_code="operational", claude_api="operational")
126
+
with patch("history.HISTORY_FILE", tmp_path / "history.json"), \
127
+
patch("history.HISTORY_DIR", tmp_path):
128
+
append_record(record)
129
+
loaded = load_history()
130
+
assert loaded == [record]
131
+
132
+
133
+
def test_load_history_multiple_records(tmp_path):
134
+
test_file = tmp_path / "history.json"
135
+
test_file.write_text(
136
+
'{"ts": 1713196800, "claude_code": "operational", "claude_api": "operational"}\n'
137
+
'{"ts": 1713196860, "claude_code": "degraded_performance", "claude_api": "operational"}\n'
138
+
)
139
+
with patch("history.HISTORY_FILE", test_file):
140
+
loaded = load_history()
141
+
assert len(loaded) == 2
142
+
assert loaded[0] == PollRecord(ts=1713196800, claude_code="operational", claude_api="operational")
143
+
assert loaded[1] == PollRecord(ts=1713196860, claude_code="degraded_performance", claude_api="operational")
144
+
145
+
146
+
def test_load_history_corrupt_file_returns_empty(tmp_path):
147
+
test_file = tmp_path / "history.json"
148
+
test_file.write_text("not valid json\n")
149
+
with patch("history.HISTORY_FILE", test_file):
150
+
assert load_history() == []
151
+
152
+
153
+
def test_append_creates_directory_if_missing(tmp_path):
154
+
data_dir = tmp_path / "nested" / "dir"
155
+
data_file = data_dir / "history.json"
156
+
record = PollRecord(ts=1, claude_code="operational", claude_api="operational")
157
+
with patch("history.HISTORY_FILE", data_file), \
158
+
patch("history.HISTORY_DIR", data_dir):
159
+
append_record(record)
160
+
assert data_file.exists()
161
+
```
162
+
163
+
- [ ] **Step 2: Run to confirm they all fail**
164
+
165
+
```bash
166
+
cd ~/mjs/claude-status
167
+
python3.12 -m pytest tests/test_history.py -v
168
+
```
169
+
170
+
Expected: `ModuleNotFoundError: No module named 'history'`
171
+
172
+
- [ ] **Step 3: Write history.py**
173
+
174
+
Create `history.py`:
175
+
176
+
```python
177
+
import json
178
+
from dataclasses import dataclass, asdict
179
+
from pathlib import Path
180
+
from typing import Literal
181
+
182
+
StatusValue = Literal[
183
+
"operational",
184
+
"degraded_performance",
185
+
"partial_outage",
186
+
"major_outage",
187
+
"unknown",
188
+
]
189
+
190
+
HISTORY_DIR = Path.home() / ".local" / "share" / "claude-status"
191
+
HISTORY_FILE = HISTORY_DIR / "history.json"
192
+
193
+
194
+
@dataclass
195
+
class PollRecord:
196
+
ts: int
197
+
claude_code: str
198
+
claude_api: str
199
+
200
+
201
+
def load_history() -> list[PollRecord]:
202
+
if not HISTORY_FILE.exists():
203
+
return []
204
+
records: list[PollRecord] = []
205
+
try:
206
+
with open(HISTORY_FILE) as f:
207
+
for line in f:
208
+
line = line.strip()
209
+
if line:
210
+
d = json.loads(line)
211
+
records.append(PollRecord(**d))
212
+
except (json.JSONDecodeError, KeyError, TypeError):
213
+
return []
214
+
return records
215
+
216
+
217
+
def append_record(record: PollRecord) -> None:
218
+
HISTORY_DIR.mkdir(parents=True, exist_ok=True)
219
+
with open(HISTORY_FILE, "a") as f:
220
+
f.write(json.dumps(asdict(record)) + "\n")
221
+
```
222
+
223
+
- [ ] **Step 4: Run tests and confirm they pass**
224
+
225
+
```bash
226
+
python3.12 -m pytest tests/test_history.py -v
227
+
```
228
+
229
+
Expected: 5 tests pass.
230
+
231
+
- [ ] **Step 5: Commit**
232
+
233
+
```bash
234
+
git add history.py tests/test_history.py
235
+
git commit -m "feat: history module — PollRecord, load/append NDJSON"
236
+
```
237
+
238
+
---
239
+
240
+
## Task 3: Poller Module
241
+
242
+
**Files:**
243
+
- Create: `poller.py`
244
+
- Create: `tests/test_poller.py`
245
+
246
+
- [ ] **Step 1: Write the failing tests**
247
+
248
+
Create `tests/test_poller.py`:
249
+
250
+
```python
251
+
import json
252
+
import urllib.error
253
+
import pytest
254
+
from unittest.mock import MagicMock, patch
255
+
from poller import fetch_status, CLAUDE_CODE_ID, CLAUDE_API_ID, FetchResult
256
+
257
+
258
+
MOCK_COMPONENTS = {
259
+
"components": [
260
+
{"id": CLAUDE_CODE_ID, "name": "Claude Code", "status": "operational"},
261
+
{"id": CLAUDE_API_ID, "name": "Claude API (api.anthropic.com)", "status": "degraded_performance"},
262
+
{"id": "other-id", "name": "claude.ai", "status": "operational"},
263
+
]
264
+
}
265
+
266
+
MOCK_INCIDENTS_EMPTY = {"incidents": []}
267
+
268
+
MOCK_INCIDENTS = {
269
+
"incidents": [
270
+
{
271
+
"id": "abc123",
272
+
"name": "Elevated errors on Claude Code",
273
+
"status": "identified",
274
+
"created_at": "2026-04-15T14:53:02.000Z",
275
+
"impact": "critical",
276
+
}
277
+
]
278
+
}
279
+
280
+
281
+
def _make_response(data: dict) -> MagicMock:
282
+
mock = MagicMock()
283
+
mock.read.return_value = json.dumps(data).encode()
284
+
mock.__enter__ = lambda s: s
285
+
mock.__exit__ = MagicMock(return_value=False)
286
+
return mock
287
+
288
+
289
+
def _make_urlopen(responses: list) -> callable:
290
+
it = iter(responses)
291
+
return lambda url, timeout: next(it)
292
+
293
+
294
+
def test_fetch_status_parses_claude_code_and_api(monkeypatch):
295
+
monkeypatch.setattr(
296
+
"poller.urllib.request.urlopen",
297
+
_make_urlopen([_make_response(MOCK_COMPONENTS), _make_response(MOCK_INCIDENTS_EMPTY)]),
298
+
)
299
+
result = fetch_status()
300
+
assert result.error is None
301
+
assert result.record.claude_code == "operational"
302
+
assert result.record.claude_api == "degraded_performance"
303
+
assert result.incidents == []
304
+
305
+
306
+
def test_fetch_status_parses_incidents(monkeypatch):
307
+
monkeypatch.setattr(
308
+
"poller.urllib.request.urlopen",
309
+
_make_urlopen([_make_response(MOCK_COMPONENTS), _make_response(MOCK_INCIDENTS)]),
310
+
)
311
+
result = fetch_status()
312
+
assert len(result.incidents) == 1
313
+
assert result.incidents[0].name == "Elevated errors on Claude Code"
314
+
assert result.incidents[0].status == "identified"
315
+
assert result.incidents[0].impact == "critical"
316
+
317
+
318
+
def test_fetch_status_network_error_returns_error(monkeypatch):
319
+
monkeypatch.setattr(
320
+
"poller.urllib.request.urlopen",
321
+
lambda url, timeout: (_ for _ in ()).throw(urllib.error.URLError("connection refused")),
322
+
)
323
+
result = fetch_status()
324
+
assert result.error is not None
325
+
assert result.record is None
326
+
assert result.incidents == []
327
+
328
+
329
+
def test_fetch_status_missing_component_returns_unknown(monkeypatch):
330
+
components_without_code = {"components": [
331
+
{"id": CLAUDE_API_ID, "name": "Claude API", "status": "operational"},
332
+
]}
333
+
monkeypatch.setattr(
334
+
"poller.urllib.request.urlopen",
335
+
_make_urlopen([_make_response(components_without_code), _make_response(MOCK_INCIDENTS_EMPTY)]),
336
+
)
337
+
result = fetch_status()
338
+
assert result.record.claude_code == "unknown"
339
+
assert result.record.claude_api == "operational"
340
+
341
+
342
+
def test_fetch_status_records_timestamp(monkeypatch):
343
+
monkeypatch.setattr(
344
+
"poller.urllib.request.urlopen",
345
+
_make_urlopen([_make_response(MOCK_COMPONENTS), _make_response(MOCK_INCIDENTS_EMPTY)]),
346
+
)
347
+
import time
348
+
before = int(time.time())
349
+
result = fetch_status()
350
+
after = int(time.time())
351
+
assert before <= result.record.ts <= after
352
+
```
353
+
354
+
- [ ] **Step 2: Run to confirm they all fail**
355
+
356
+
```bash
357
+
python3.12 -m pytest tests/test_poller.py -v
358
+
```
359
+
360
+
Expected: `ModuleNotFoundError: No module named 'poller'`
361
+
362
+
- [ ] **Step 3: Write poller.py**
363
+
364
+
Create `poller.py`:
365
+
366
+
```python
367
+
import json
368
+
import time
369
+
import urllib.request
370
+
import urllib.error
371
+
from dataclasses import dataclass, field
372
+
373
+
from history import PollRecord
374
+
375
+
COMPONENTS_URL = "https://status.claude.com/api/v2/components.json"
376
+
INCIDENTS_URL = "https://status.claude.com/api/v2/incidents/unresolved.json"
377
+
378
+
CLAUDE_CODE_ID = "yyzkbfz2thpt"
379
+
CLAUDE_API_ID = "k8w3r06qmzrp"
380
+
381
+
382
+
@dataclass
383
+
class Incident:
384
+
name: str
385
+
status: str
386
+
created_at: str
387
+
impact: str
388
+
389
+
390
+
@dataclass
391
+
class FetchResult:
392
+
record: PollRecord | None
393
+
incidents: list[Incident] = field(default_factory=list)
394
+
error: str | None = None
395
+
396
+
397
+
def _fetch_json(url: str) -> dict:
398
+
with urllib.request.urlopen(url, timeout=10) as resp:
399
+
return json.loads(resp.read())
400
+
401
+
402
+
def fetch_status() -> FetchResult:
403
+
try:
404
+
components_data = _fetch_json(COMPONENTS_URL)
405
+
by_id = {c["id"]: c["status"] for c in components_data["components"]}
406
+
407
+
record = PollRecord(
408
+
ts=int(time.time()),
409
+
claude_code=by_id.get(CLAUDE_CODE_ID, "unknown"),
410
+
claude_api=by_id.get(CLAUDE_API_ID, "unknown"),
411
+
)
412
+
413
+
incidents_data = _fetch_json(INCIDENTS_URL)
414
+
incidents = [
415
+
Incident(
416
+
name=i["name"],
417
+
status=i["status"],
418
+
created_at=i["created_at"],
419
+
impact=i["impact"],
420
+
)
421
+
for i in incidents_data.get("incidents", [])
422
+
]
423
+
424
+
return FetchResult(record=record, incidents=incidents)
425
+
426
+
except Exception as exc:
427
+
return FetchResult(record=None, error=str(exc))
428
+
```
429
+
430
+
- [ ] **Step 4: Run tests and confirm they pass**
431
+
432
+
```bash
433
+
python3.12 -m pytest tests/test_poller.py -v
434
+
```
435
+
436
+
Expected: 5 tests pass.
437
+
438
+
- [ ] **Step 5: Commit**
439
+
440
+
```bash
441
+
git add poller.py tests/test_poller.py
442
+
git commit -m "feat: poller module — fetch_status with Incident/FetchResult"
443
+
```
444
+
445
+
---
446
+
447
+
## Task 4: Notifier Module
448
+
449
+
**Files:**
450
+
- Create: `notifier.py`
451
+
- Create: `tests/test_notifier.py`
452
+
453
+
- [ ] **Step 1: Write the failing tests**
454
+
455
+
Create `tests/test_notifier.py`:
456
+
457
+
```python
458
+
import os
459
+
import pytest
460
+
from unittest.mock import patch, call
461
+
from notifier import detect_transition, notify_cmux, check_and_notify
462
+
463
+
464
+
def test_detect_degraded():
465
+
assert detect_transition("operational", "degraded_performance") == "degraded"
466
+
467
+
468
+
def test_detect_outage():
469
+
assert detect_transition("operational", "partial_outage") == "degraded"
470
+
471
+
472
+
def test_detect_major_outage():
473
+
assert detect_transition("operational", "major_outage") == "degraded"
474
+
475
+
476
+
def test_detect_recovered():
477
+
assert detect_transition("degraded_performance", "operational") == "recovered"
478
+
479
+
480
+
def test_detect_no_change_operational():
481
+
assert detect_transition("operational", "operational") is None
482
+
483
+
484
+
def test_detect_no_change_degraded():
485
+
assert detect_transition("degraded_performance", "partial_outage") is None
486
+
487
+
488
+
def test_notify_skipped_when_not_in_cmux():
489
+
with patch.dict(os.environ, {}, clear=True):
490
+
with patch("notifier.subprocess.run") as mock_run:
491
+
notify_cmux("title", "body")
492
+
mock_run.assert_not_called()
493
+
494
+
495
+
def test_notify_fires_when_in_cmux():
496
+
with patch.dict(os.environ, {"CMUX_WORKSPACE_ID": "workspace:1"}):
497
+
with patch("notifier.subprocess.run") as mock_run:
498
+
notify_cmux("Claude Code degraded", "Degraded Performance")
499
+
mock_run.assert_called_once_with(
500
+
["cmux", "notify", "--title", "Claude Code degraded", "--body", "Degraded Performance"],
501
+
capture_output=True,
502
+
)
503
+
504
+
505
+
def test_check_and_notify_fires_on_degradation():
506
+
with patch.dict(os.environ, {"CMUX_WORKSPACE_ID": "workspace:1"}):
507
+
with patch("notifier.subprocess.run") as mock_run:
508
+
check_and_notify("Claude Code", "operational", "degraded_performance")
509
+
args = mock_run.call_args[0][0]
510
+
assert "Claude Code degraded" in args
511
+
512
+
513
+
def test_check_and_notify_fires_on_recovery():
514
+
with patch.dict(os.environ, {"CMUX_WORKSPACE_ID": "workspace:1"}):
515
+
with patch("notifier.subprocess.run") as mock_run:
516
+
check_and_notify("Claude Code", "degraded_performance", "operational")
517
+
args = mock_run.call_args[0][0]
518
+
assert "Claude Code recovered" in args
519
+
520
+
521
+
def test_check_and_notify_silent_on_no_change():
522
+
with patch.dict(os.environ, {"CMUX_WORKSPACE_ID": "workspace:1"}):
523
+
with patch("notifier.subprocess.run") as mock_run:
524
+
check_and_notify("Claude Code", "operational", "operational")
525
+
mock_run.assert_not_called()
526
+
```
527
+
528
+
- [ ] **Step 2: Run to confirm they all fail**
529
+
530
+
```bash
531
+
python3.12 -m pytest tests/test_notifier.py -v
532
+
```
533
+
534
+
Expected: `ModuleNotFoundError: No module named 'notifier'`
535
+
536
+
- [ ] **Step 3: Write notifier.py**
537
+
538
+
Create `notifier.py`:
539
+
540
+
```python
541
+
import os
542
+
import subprocess
543
+
544
+
545
+
def detect_transition(old: str, new: str) -> str | None:
546
+
"""Returns 'degraded', 'recovered', or None."""
547
+
was_ok = old == "operational"
548
+
is_ok = new == "operational"
549
+
if was_ok and not is_ok:
550
+
return "degraded"
551
+
if not was_ok and is_ok:
552
+
return "recovered"
553
+
return None
554
+
555
+
556
+
def notify_cmux(title: str, body: str) -> None:
557
+
if not os.environ.get("CMUX_WORKSPACE_ID"):
558
+
return
559
+
subprocess.run(
560
+
["cmux", "notify", "--title", title, "--body", body],
561
+
capture_output=True,
562
+
)
563
+
564
+
565
+
def check_and_notify(component_name: str, old_status: str, new_status: str) -> None:
566
+
transition = detect_transition(old_status, new_status)
567
+
if transition == "degraded":
568
+
label = new_status.replace("_", " ").title()
569
+
notify_cmux(f"{component_name} degraded", label)
570
+
elif transition == "recovered":
571
+
notify_cmux(f"{component_name} recovered", "Back to operational")
572
+
```
573
+
574
+
- [ ] **Step 4: Run tests and confirm they pass**
575
+
576
+
```bash
577
+
python3.12 -m pytest tests/test_notifier.py -v
578
+
```
579
+
580
+
Expected: 10 tests pass.
581
+
582
+
- [ ] **Step 5: Commit**
583
+
584
+
```bash
585
+
git add notifier.py tests/test_notifier.py
586
+
git commit -m "feat: notifier module — transition detection + cmux notify"
587
+
```
588
+
589
+
---
590
+
591
+
## Task 5: App Logic (Pure Functions)
592
+
593
+
**Files:**
594
+
- Create: `tests/test_app_logic.py`
595
+
- Create: `app.py` (logic functions only — no Textual imports yet)
596
+
597
+
These functions are pure and fast to test without spinning up a Textual app.
598
+
599
+
- [ ] **Step 1: Write the failing tests**
600
+
601
+
Create `tests/test_app_logic.py`:
602
+
603
+
```python
604
+
from history import PollRecord
605
+
from app import render_bar, uptime_percent
606
+
607
+
608
+
def _op(ts: int) -> PollRecord:
609
+
return PollRecord(ts=ts, claude_code="operational", claude_api="operational")
610
+
611
+
612
+
def _deg(ts: int) -> PollRecord:
613
+
return PollRecord(ts=ts, claude_code="degraded_performance", claude_api="operational")
614
+
615
+
616
+
def test_uptime_100_percent():
617
+
history = [_op(i) for i in range(10)]
618
+
assert uptime_percent(history, "claude_code") == 100.0
619
+
620
+
621
+
def test_uptime_50_percent():
622
+
history = [_op(1), _deg(2)]
623
+
assert uptime_percent(history, "claude_code") == 50.0
624
+
625
+
626
+
def test_uptime_empty_history():
627
+
assert uptime_percent([], "claude_code") == 0.0
628
+
629
+
630
+
def test_render_bar_operational_uses_green():
631
+
bar = render_bar([_op(1)], "claude_code", width=10)
632
+
assert "green" in bar
633
+
assert "█" in bar
634
+
635
+
636
+
def test_render_bar_degraded_uses_orange():
637
+
bar = render_bar([_deg(1)], "claude_code", width=10)
638
+
assert "orange3" in bar
639
+
640
+
641
+
def test_render_bar_truncates_to_width():
642
+
history = [_op(i) for i in range(20)]
643
+
bar = render_bar(history, "claude_code", width=5)
644
+
# Rich markup wraps each char; count raw block chars
645
+
assert bar.count("█") == 5
646
+
647
+
648
+
def test_render_bar_shows_all_when_fewer_than_width():
649
+
history = [_op(i) for i in range(3)]
650
+
bar = render_bar(history, "claude_code", width=10)
651
+
assert bar.count("█") == 3
652
+
653
+
654
+
def test_render_bar_unknown_uses_dim_block():
655
+
record = PollRecord(ts=1, claude_code="unknown", claude_api="operational")
656
+
bar = render_bar([record], "claude_code", width=10)
657
+
assert "░" in bar
658
+
assert "dim" in bar
659
+
```
660
+
661
+
- [ ] **Step 2: Run to confirm they fail**
662
+
663
+
```bash
664
+
python3.12 -m pytest tests/test_app_logic.py -v
665
+
```
666
+
667
+
Expected: `ModuleNotFoundError: No module named 'app'`
668
+
669
+
- [ ] **Step 3: Write app.py with just the logic functions**
670
+
671
+
Create `app.py`:
672
+
673
+
```python
674
+
import os
675
+
import subprocess
676
+
from datetime import datetime, timezone
677
+
678
+
from textual.app import App, ComposeResult
679
+
from textual.widgets import Static
680
+
from textual import work
681
+
682
+
from history import PollRecord, load_history, append_record
683
+
from poller import fetch_status, Incident
684
+
from notifier import check_and_notify
685
+
686
+
POLL_INTERVAL = 60
687
+
688
+
_STATUS_COLORS = {
689
+
"operational": "green",
690
+
"degraded_performance": "orange3",
691
+
"partial_outage": "red",
692
+
"major_outage": "bold red",
693
+
"unknown": "dim",
694
+
}
695
+
696
+
_STATUS_LABELS = {
697
+
"operational": "Operational",
698
+
"degraded_performance": "Degraded Performance",
699
+
"partial_outage": "Partial Outage",
700
+
"major_outage": "Major Outage",
701
+
"unknown": "Unknown",
702
+
}
703
+
704
+
_BAR_CHARS: dict[str, tuple[str, str]] = {
705
+
"operational": ("█", "green"),
706
+
"degraded_performance": ("█", "orange3"),
707
+
"partial_outage": ("█", "red"),
708
+
"major_outage": ("█", "bold red"),
709
+
"unknown": ("░", "dim"),
710
+
}
711
+
712
+
713
+
def uptime_percent(history: list[PollRecord], field: str) -> float:
714
+
if not history:
715
+
return 0.0
716
+
operational = sum(1 for r in history if getattr(r, field) == "operational")
717
+
return (operational / len(history)) * 100.0
718
+
719
+
720
+
def render_bar(history: list[PollRecord], field: str, width: int) -> str:
721
+
records = history[-width:] if len(history) > width else history
722
+
parts = []
723
+
for r in records:
724
+
status = getattr(r, field)
725
+
char, color = _BAR_CHARS.get(status, ("░", "dim"))
726
+
parts.append(f"[{color}]{char}[/{color}]")
727
+
return "".join(parts)
728
+
729
+
730
+
def _format_age(created_at: str) -> str:
731
+
try:
732
+
ts = datetime.fromisoformat(created_at.replace("Z", "+00:00"))
733
+
delta = int((datetime.now(timezone.utc) - ts).total_seconds())
734
+
if delta < 60:
735
+
return f"{delta}s ago"
736
+
elif delta < 3600:
737
+
return f"{delta // 60} min ago"
738
+
else:
739
+
return f"{delta // 3600} hr ago"
740
+
except Exception:
741
+
return "unknown age"
742
+
743
+
744
+
class ComponentRow(Static):
745
+
def __init__(self, label: str, field: str, **kwargs) -> None:
746
+
super().__init__("Loading...", **kwargs)
747
+
self._label = label
748
+
self._field = field
749
+
self._history: list[PollRecord] = []
750
+
self._current_status: str = "unknown"
751
+
752
+
def update_data(self, history: list[PollRecord], current_status: str) -> None:
753
+
self._history = history
754
+
self._current_status = current_status
755
+
self._redraw()
756
+
757
+
def _redraw(self) -> None:
758
+
width = os.get_terminal_size(fallback=(80, 24)).columns - 4
759
+
color = _STATUS_COLORS.get(self._current_status, "dim")
760
+
label_str = _STATUS_LABELS.get(self._current_status, "Unknown")
761
+
pct = uptime_percent(self._history, self._field)
762
+
bars = render_bar(self._history, self._field, width)
763
+
764
+
badge = f"[{color}]● {label_str}[/{color}]"
765
+
padding = max(1, width - len(self._label) - len(label_str) - 3)
766
+
top = f"[bold]{self._label}[/bold]{' ' * padding}{badge}"
767
+
768
+
dash = max(1, (width - 22) // 2)
769
+
footer = f"[dim]Session start {'─' * dash} {pct:.1f}% uptime {'─' * dash} Now[/dim]"
770
+
771
+
self.update(f"{top}\n{bars}\n{footer}")
772
+
773
+
774
+
class IncidentBanner(Static):
775
+
def __init__(self, **kwargs) -> None:
776
+
super().__init__("", **kwargs)
777
+
778
+
def update_incidents(self, incidents: list[Incident]) -> None:
779
+
if not incidents:
780
+
self.update("")
781
+
self.add_class("hidden")
782
+
return
783
+
self.remove_class("hidden")
784
+
lines = []
785
+
for incident in incidents:
786
+
age = _format_age(incident.created_at)
787
+
lines.append(f"[yellow]⚠[/yellow] [bold]{incident.name}[/bold]")
788
+
lines.append(f" [dim]{incident.status.capitalize()} · {age}[/dim]")
789
+
self.update("\n".join(lines))
790
+
791
+
792
+
class StatusFooter(Static):
793
+
def __init__(self, **kwargs) -> None:
794
+
super().__init__("Initializing...", **kwargs)
795
+
self._last_updated: str = "—"
796
+
self._next_poll: int = POLL_INTERVAL
797
+
self._error: str | None = None
798
+
799
+
def on_mount(self) -> None:
800
+
self.set_interval(1, self._tick)
801
+
802
+
def _tick(self) -> None:
803
+
if self._next_poll > 0:
804
+
self._next_poll -= 1
805
+
self._redraw()
806
+
807
+
def mark_polled(self, error: str | None = None) -> None:
808
+
self._last_updated = datetime.now().strftime("%H:%M:%S")
809
+
self._next_poll = POLL_INTERVAL
810
+
self._error = error
811
+
self._redraw()
812
+
813
+
def _redraw(self) -> None:
814
+
if self._error:
815
+
status = f"[red]Fetch failed: {self._error[:50]}[/red]"
816
+
else:
817
+
status = f"Last updated: [bold]{self._last_updated}[/bold]"
818
+
self.update(f"{status} · Next poll: {self._next_poll}s")
819
+
820
+
821
+
class ClaudeStatusApp(App):
822
+
TITLE = "Claude Status"
823
+
CSS = """
824
+
Screen {
825
+
layout: vertical;
826
+
background: $surface;
827
+
}
828
+
ComponentRow {
829
+
height: 5;
830
+
border: solid $panel-lighten-1;
831
+
padding: 0 1;
832
+
margin: 0;
833
+
}
834
+
IncidentBanner {
835
+
height: auto;
836
+
background: $warning 15%;
837
+
padding: 0 1;
838
+
margin: 0;
839
+
}
840
+
IncidentBanner.hidden {
841
+
display: none;
842
+
}
843
+
StatusFooter {
844
+
height: 1;
845
+
dock: bottom;
846
+
background: $panel;
847
+
padding: 0 1;
848
+
}
849
+
"""
850
+
851
+
BINDINGS = [("q", "quit", "Quit")]
852
+
853
+
def __init__(self) -> None:
854
+
super().__init__()
855
+
self._history: list[PollRecord] = []
856
+
self._prev_claude_code: str = "unknown"
857
+
self._prev_claude_api: str = "unknown"
858
+
859
+
def compose(self) -> ComposeResult:
860
+
yield ComponentRow("Claude Code", "claude_code", id="row-claude-code")
861
+
yield ComponentRow("Claude API (api.anthropic.com)", "claude_api", id="row-claude-api")
862
+
yield IncidentBanner(id="incident-banner")
863
+
yield StatusFooter(id="status-footer")
864
+
865
+
def on_mount(self) -> None:
866
+
self._history = load_history()
867
+
if self._history:
868
+
latest = self._history[-1]
869
+
self._prev_claude_code = latest.claude_code
870
+
self._prev_claude_api = latest.claude_api
871
+
self._refresh_rows()
872
+
self._do_poll()
873
+
self.set_interval(POLL_INTERVAL, self._do_poll)
874
+
875
+
def _refresh_rows(self) -> None:
876
+
if not self._history:
877
+
return
878
+
latest = self._history[-1]
879
+
self.query_one("#row-claude-code", ComponentRow).update_data(
880
+
self._history, latest.claude_code
881
+
)
882
+
self.query_one("#row-claude-api", ComponentRow).update_data(
883
+
self._history, latest.claude_api
884
+
)
885
+
886
+
def _do_poll(self) -> None:
887
+
self._poll_worker()
888
+
889
+
@work(thread=True)
890
+
def _poll_worker(self) -> None:
891
+
result = fetch_status()
892
+
self.call_from_thread(self._handle_result, result)
893
+
894
+
def _handle_result(self, result) -> None:
895
+
footer = self.query_one("#status-footer", StatusFooter)
896
+
897
+
if result.error or result.record is None:
898
+
footer.mark_polled(error=result.error or "unknown error")
899
+
return
900
+
901
+
record = result.record
902
+
append_record(record)
903
+
self._history.append(record)
904
+
905
+
check_and_notify("Claude Code", self._prev_claude_code, record.claude_code)
906
+
check_and_notify("Claude API", self._prev_claude_api, record.claude_api)
907
+
self._prev_claude_code = record.claude_code
908
+
self._prev_claude_api = record.claude_api
909
+
910
+
self._refresh_rows()
911
+
self.query_one("#incident-banner", IncidentBanner).update_incidents(result.incidents)
912
+
footer.mark_polled()
913
+
```
914
+
915
+
- [ ] **Step 4: Run the logic tests**
916
+
917
+
```bash
918
+
python3.12 -m pytest tests/test_app_logic.py -v
919
+
```
920
+
921
+
Expected: 9 tests pass.
922
+
923
+
- [ ] **Step 5: Run all tests so far**
924
+
925
+
```bash
926
+
python3.12 -m pytest -v
927
+
```
928
+
929
+
Expected: All tests pass (history + poller + notifier + app_logic).
930
+
931
+
- [ ] **Step 6: Commit**
932
+
933
+
```bash
934
+
git add app.py tests/test_app_logic.py
935
+
git commit -m "feat: app module — widgets, logic functions, ClaudeStatusApp"
936
+
```
937
+
938
+
---
939
+
940
+
## Task 6: Entry Point + Smoke Test
941
+
942
+
**Files:**
943
+
- Create: `claude_status.py`
944
+
945
+
- [ ] **Step 1: Write claude_status.py**
946
+
947
+
Create `claude_status.py`:
948
+
949
+
```python
950
+
#!/usr/bin/env python3.12
951
+
from app import ClaudeStatusApp
952
+
953
+
if __name__ == "__main__":
954
+
ClaudeStatusApp().run()
955
+
```
956
+
957
+
- [ ] **Step 2: Make it executable**
958
+
959
+
```bash
960
+
chmod +x ~/mjs/claude-status/claude_status.py
961
+
```
962
+
963
+
- [ ] **Step 3: Run a live smoke test**
964
+
965
+
```bash
966
+
cd ~/mjs/claude-status
967
+
python3.12 claude_status.py
968
+
```
969
+
970
+
Expected:
971
+
- TUI launches in terminal
972
+
- Two `ComponentRow` widgets visible: "Claude Code" and "Claude API (api.anthropic.com)"
973
+
- Status badges colored appropriately (green = operational, orange = degraded)
974
+
- Bar chart fills from the left as poll records accumulate
975
+
- Status footer shows "Last updated: HH:MM:SS · Next poll: 60s"
976
+
- If an active incident exists, a yellow `⚠` banner appears
977
+
- Press `q` to quit
978
+
979
+
- [ ] **Step 4: Verify history file was written**
980
+
981
+
```bash
982
+
cat ~/.local/share/claude-status/history.json
983
+
```
984
+
985
+
Expected: One or more lines of NDJSON, e.g.:
986
+
```json
987
+
{"ts": 1713196800, "claude_code": "degraded_performance", "claude_api": "operational"}
988
+
```
989
+
990
+
- [ ] **Step 5: Run again and verify history loads**
991
+
992
+
```bash
993
+
python3.12 claude_status.py
994
+
```
995
+
996
+
Expected: Bar chart starts with the previously saved records visible (not empty).
997
+
998
+
- [ ] **Step 6: Commit**
999
+
1000
+
```bash
1001
+
git add claude_status.py
1002
+
git commit -m "feat: entry point — claude_status.py"
1003
+
```
1004
+
1005
+
---
1006
+
1007
+
## Task 7: Final Polish + README
1008
+
1009
+
**Files:**
1010
+
- Create: `README.md`
1011
+
1012
+
- [ ] **Step 1: Run full test suite one final time**
1013
+
1014
+
```bash
1015
+
cd ~/mjs/claude-status
1016
+
python3.12 -m pytest -v
1017
+
```
1018
+
1019
+
Expected: All tests pass, no warnings.
1020
+
1021
+
- [ ] **Step 2: Write README.md**
1022
+
1023
+
Create `README.md`:
1024
+
1025
+
```markdown
1026
+
# claude-status
1027
+
1028
+
Terminal dashboard for monitoring Claude Code and Claude API health.
1029
+
1030
+
## Setup
1031
+
1032
+
```bash
1033
+
pip3.12 install textual==8.2.3
1034
+
```
1035
+
1036
+
## Run
1037
+
1038
+
```bash
1039
+
python3.12 claude_status.py
1040
+
```
1041
+
1042
+
Press `q` to quit.
1043
+
1044
+
## What it shows
1045
+
1046
+
- **Claude Code** and **Claude API** status (Operational / Degraded / Outage)
1047
+
- Session-accumulated bar chart — one bar per 60s poll, persisted across restarts
1048
+
- Active incident banner when the statuspage reports an open incident
1049
+
- cmux notification on status change (when running inside cmux)
1050
+
1051
+
## Data
1052
+
1053
+
Poll history stored at `~/.local/share/claude-status/history.json` (NDJSON, one record per line).
1054
+
```
1055
+
1056
+
- [ ] **Step 3: Commit**
1057
+
1058
+
```bash
1059
+
git add README.md
1060
+
git commit -m "docs: README with setup and usage"
1061
+
```
1062
+
1063
+
---
1064
+
1065
+
## Self-Review Checklist
1066
+
1067
+
- [x] **Spec coverage:** Layout (ComponentRow × 2, IncidentBanner, StatusFooter) ✓ · Session history (load on mount, append each poll) ✓ · Status colors ✓ · 60s poll interval ✓ · cmux notify on transition ✓ · Incident banner ✓ · Error handling (stale indicator, corrupt history) ✓ · Terminal-width-responsive bars ✓ · Minimum width fallback (fallback=(80,24)) ✓
1068
+
- [x] **Placeholder scan:** No TBDs. All code complete.
1069
+
- [x] **Type consistency:** `PollRecord` defined in Task 2, used consistently in Tasks 3–5. `Incident`/`FetchResult` defined in Task 3, used in Tasks 5. `detect_transition` returns `str | None` in Task 4, consumed in `check_and_notify` correctly. `mark_polled` defined and called consistently.
+152
docs/superpowers/specs/2026-04-15-claude-status-tui-design.md
+152
docs/superpowers/specs/2026-04-15-claude-status-tui-design.md
···
1
+
# Claude Status TUI — Design Spec
2
+
3
+
**Date:** 2026-04-15
4
+
**Status:** Approved
5
+
6
+
---
7
+
8
+
## Overview
9
+
10
+
A Textual-based TUI dashboard that polls the Claude statuspage API to monitor the health of Claude Code and the Claude API in real time. Displays a session-accumulated bar chart history, a current status callout, an active incident banner, and fires cmux notifications on status changes.
11
+
12
+
The history reflects the user's local session — not global 90-day uptime — making it contextually relevant to their actual dev experience.
13
+
14
+
---
15
+
16
+
## Architecture
17
+
18
+
### Single-file script
19
+
20
+
`~/dev/claude-status/claude_status.py`
21
+
22
+
Run directly: `python claude_status.py`
23
+
24
+
One dependency beyond stdlib: `textual`. No packaging, no virtualenv required beyond `pip install textual`.
25
+
26
+
### Data persistence
27
+
28
+
Poll results are appended to:
29
+
30
+
```
31
+
~/.local/share/claude-status/history.json
32
+
```
33
+
34
+
Format — one JSON object per line (newline-delimited JSON):
35
+
36
+
```json
37
+
{"ts": 1713196800, "claude_code": "degraded_performance", "claude_api": "operational"}
38
+
```
39
+
40
+
On startup, existing history is loaded and rendered. Each poll appends a new record. File is created automatically on first run.
41
+
42
+
### Polling
43
+
44
+
- **Interval:** 60 seconds
45
+
- **Components endpoint:** `https://status.claude.com/api/v2/components.json`
46
+
- Claude Code component ID: `yyzkbfz2thpt`
47
+
- Claude API component ID: `k8w3r06qmzrp`
48
+
- **Incidents endpoint:** `https://status.claude.com/api/v2/incidents/unresolved.json`
49
+
- Both fetched on each poll tick. Network errors are caught; previous status is retained with a visual "stale" indicator.
50
+
51
+
---
52
+
53
+
## Layout
54
+
55
+
```
56
+
┌─────────────────────────────────────────────────────────┐
57
+
│ Claude Code ● Degraded Performance │
58
+
│ ▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▒▒░░░░▒▒▓▓▓▓▓▓▓▓▓▓▓▓▓▒▒▒▒ │
59
+
│ Session start ──────────────── 97.2% uptime ─────── Now │
60
+
├─────────────────────────────────────────────────────────┤
61
+
│ Claude API (api.anthropic.com) ● Operational │
62
+
│ ▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓▓ │
63
+
│ Session start ──────────────── 100% uptime ──────── Now │
64
+
├─────────────────────────────────────────────────────────┤
65
+
│ ⚠ Elevated errors on Claude.ai, API, Claude Code │
66
+
│ Identified · 15 min ago │
67
+
├─────────────────────────────────────────────────────────┤
68
+
│ Last updated: 14:32:07 · Next poll: 47s │
69
+
└─────────────────────────────────────────────────────────┘
70
+
```
71
+
72
+
- Two `ComponentRow` widgets stacked vertically, one per monitored service
73
+
- `IncidentBanner` widget below — hidden when no active incidents
74
+
- `StatusBar` at the bottom with last-updated timestamp and countdown to next poll
75
+
- Terminal-width-responsive: bar chart fills available width, truncates oldest bars first when narrow
76
+
77
+
---
78
+
79
+
## Components
80
+
81
+
### `ComponentRow`
82
+
83
+
Renders one service. Props: `name`, `history: list[PollRecord]`, `current_status`.
84
+
85
+
- **Header line:** service name left, colored status badge right
86
+
- **Bar chart:** one bar character per poll record, colored by status, fills terminal width (most recent = rightmost, oldest dropped when space runs out)
87
+
- **Footer line:** "Session start" left, uptime % center, "Now" right
88
+
89
+
Bar character mapping:
90
+
- `operational` → `█` green
91
+
- `degraded_performance` → `█` orange
92
+
- `partial_outage` → `█` red
93
+
- `major_outage` → `█` bold red
94
+
- `unknown` / fetch error → `░` dim (stale indicator)
95
+
96
+
### `IncidentBanner`
97
+
98
+
Shown only when `incidents/unresolved.json` returns one or more incidents. Displays each active incident as its own line: name, status label, and age ("15 min ago", "2 hr ago"). Hidden when no active incidents. In practice there is rarely more than one active incident; if there are multiple, all are shown stacked.
99
+
100
+
### `StatusBar`
101
+
102
+
Single footer line: last poll timestamp + countdown to next poll (counts down from 60).
103
+
104
+
---
105
+
106
+
## Notifications
107
+
108
+
On each poll, compare new status to previous status per component:
109
+
110
+
- **Degraded/outage** (was operational, now not): fire `cmux notify --title "Claude Code degraded" --body "<status label>"`
111
+
- **Recovery** (was not operational, now operational): fire `cmux notify --title "Claude Code recovered" --body "Back to operational"`
112
+
113
+
cmux detection: check `CMUX_WORKSPACE_ID` env var. If not set, skip notify silently — the TUI still updates normally.
114
+
115
+
Notify once per transition, not on every poll.
116
+
117
+
---
118
+
119
+
## Error Handling
120
+
121
+
- **Network error on poll:** retain last known status, mark bars as stale (`░`), show "fetch failed" in status bar. Retry next interval.
122
+
- **History file unreadable/corrupt:** start fresh, log warning to status bar, do not crash.
123
+
- **Terminal too narrow to render bars:** show at least the status badge and name; suppress bar chart below a minimum width (40 chars).
124
+
125
+
---
126
+
127
+
## File Structure
128
+
129
+
```
130
+
~/dev/claude-status/
131
+
├── claude_status.py # single-file TUI app
132
+
├── requirements.txt # textual pinned to latest stable at implementation time
133
+
└── docs/
134
+
└── superpowers/
135
+
└── specs/
136
+
└── 2026-04-15-claude-status-tui-design.md
137
+
```
138
+
139
+
Data directory (outside repo):
140
+
```
141
+
~/.local/share/claude-status/
142
+
└── history.json # newline-delimited JSON poll records
143
+
```
144
+
145
+
---
146
+
147
+
## Out of Scope
148
+
149
+
- More than two components (can be added later by changing a config list)
150
+
- Interactive keyboard navigation / scrolling history
151
+
- Alerting via channels other than cmux (email, Slack, etc.)
152
+
- Packaging / distribution
+48
history.py
+48
history.py
···
1
+
import json
2
+
from dataclasses import dataclass, asdict
3
+
from pathlib import Path
4
+
from typing import Literal
5
+
6
+
StatusValue = Literal[
7
+
"operational",
8
+
"degraded_performance",
9
+
"partial_outage",
10
+
"major_outage",
11
+
"unknown",
12
+
]
13
+
14
+
HISTORY_DIR = Path.home() / ".local" / "share" / "claude-status"
15
+
HISTORY_FILE = HISTORY_DIR / "history.json"
16
+
17
+
18
+
@dataclass
19
+
class PollRecord:
20
+
ts: int
21
+
claude_code: StatusValue
22
+
claude_api: StatusValue
23
+
24
+
25
+
def load_history() -> list[PollRecord]:
26
+
if not HISTORY_FILE.exists():
27
+
return []
28
+
records: list[PollRecord] = []
29
+
try:
30
+
with open(HISTORY_FILE) as f:
31
+
for line in f:
32
+
line = line.strip()
33
+
if not line:
34
+
continue
35
+
try:
36
+
d = json.loads(line)
37
+
records.append(PollRecord(**d))
38
+
except (json.JSONDecodeError, KeyError, TypeError):
39
+
continue
40
+
except OSError:
41
+
return []
42
+
return records
43
+
44
+
45
+
def append_record(record: PollRecord) -> None:
46
+
HISTORY_DIR.mkdir(parents=True, exist_ok=True)
47
+
with open(HISTORY_FILE, "a") as f:
48
+
f.write(json.dumps(asdict(record)) + "\n")
+47
notifier.py
+47
notifier.py
···
1
+
import os
2
+
import subprocess
3
+
4
+
5
+
_DEGRADED_STATUSES = frozenset({
6
+
"degraded_performance",
7
+
"partial_outage",
8
+
"major_outage",
9
+
})
10
+
11
+
12
+
def detect_transition(old: str, new: str) -> str | None:
13
+
"""Returns 'degraded', 'recovered', or None.
14
+
15
+
'unknown' is treated as neutral — it represents a polling failure,
16
+
not an actual service status change, so it never triggers notifications.
17
+
"""
18
+
if old == "unknown" or new == "unknown":
19
+
return None
20
+
was_ok = old == "operational"
21
+
is_ok = new == "operational"
22
+
if was_ok and new in _DEGRADED_STATUSES:
23
+
return "degraded"
24
+
if old in _DEGRADED_STATUSES and is_ok:
25
+
return "recovered"
26
+
return None
27
+
28
+
29
+
def notify_cmux(title: str, body: str) -> None:
30
+
if not os.environ.get("CMUX_WORKSPACE_ID"):
31
+
return
32
+
try:
33
+
subprocess.run(
34
+
["cmux", "notify", "--title", title, "--body", body],
35
+
capture_output=True,
36
+
)
37
+
except (FileNotFoundError, OSError):
38
+
pass # cmux not available; notification is best-effort
39
+
40
+
41
+
def check_and_notify(component_name: str, old_status: str, new_status: str) -> None:
42
+
transition = detect_transition(old_status, new_status)
43
+
if transition == "degraded":
44
+
label = new_status.replace("_", " ").title()
45
+
notify_cmux(f"{component_name} degraded", label)
46
+
elif transition == "recovered":
47
+
notify_cmux(f"{component_name} recovered", "Back to operational")
+67
poller.py
+67
poller.py
···
1
+
import json
2
+
import ssl
3
+
import time
4
+
import urllib.request
5
+
from dataclasses import dataclass, field
6
+
7
+
import certifi
8
+
9
+
from history import PollRecord
10
+
11
+
COMPONENTS_URL = "https://status.claude.com/api/v2/components.json"
12
+
INCIDENTS_URL = "https://status.claude.com/api/v2/incidents/unresolved.json"
13
+
14
+
CLAUDE_CODE_ID = "yyzkbfz2thpt"
15
+
CLAUDE_API_ID = "k8w3r06qmzrp"
16
+
17
+
18
+
@dataclass
19
+
class Incident:
20
+
name: str
21
+
status: str
22
+
created_at: str
23
+
impact: str
24
+
25
+
26
+
@dataclass
27
+
class FetchResult:
28
+
record: PollRecord | None
29
+
incidents: list[Incident] = field(default_factory=list)
30
+
error: str | None = None
31
+
32
+
33
+
def _fetch_json(url: str) -> dict:
34
+
ctx = ssl.create_default_context(cafile=certifi.where())
35
+
with urllib.request.urlopen(url, timeout=10, context=ctx) as resp:
36
+
data = json.loads(resp.read())
37
+
if not isinstance(data, dict):
38
+
raise ValueError(f"Expected JSON object from {url}, got {type(data).__name__}")
39
+
return data
40
+
41
+
42
+
def fetch_status() -> FetchResult:
43
+
try:
44
+
components_data = _fetch_json(COMPONENTS_URL)
45
+
by_id = {c["id"]: c["status"] for c in components_data["components"]}
46
+
record = PollRecord(
47
+
ts=int(time.time()),
48
+
claude_code=by_id.get(CLAUDE_CODE_ID, "unknown"),
49
+
claude_api=by_id.get(CLAUDE_API_ID, "unknown"),
50
+
)
51
+
except Exception as exc:
52
+
return FetchResult(record=None, error=str(exc))
53
+
54
+
try:
55
+
incidents_data = _fetch_json(INCIDENTS_URL)
56
+
incidents = [
57
+
Incident(
58
+
name=i["name"],
59
+
status=i["status"],
60
+
created_at=i["created_at"],
61
+
impact=i["impact"],
62
+
)
63
+
for i in incidents_data.get("incidents", [])
64
+
]
65
+
return FetchResult(record=record, incidents=incidents)
66
+
except Exception as exc:
67
+
return FetchResult(record=record, incidents=[], error=str(exc))
tests/__init__.py
tests/__init__.py
This is a binary file and will not be displayed.
+85
tests/test_app_logic.py
+85
tests/test_app_logic.py
···
1
+
from history import PollRecord
2
+
from app import render_bar, uptime_percent, _format_age
3
+
4
+
5
+
def _op(ts: int) -> PollRecord:
6
+
return PollRecord(ts=ts, claude_code="operational", claude_api="operational")
7
+
8
+
9
+
def _deg(ts: int) -> PollRecord:
10
+
return PollRecord(ts=ts, claude_code="degraded_performance", claude_api="operational")
11
+
12
+
13
+
def test_uptime_100_percent():
14
+
history = [_op(i) for i in range(10)]
15
+
assert uptime_percent(history, "claude_code") == 100.0
16
+
17
+
18
+
def test_uptime_50_percent():
19
+
history = [_op(1), _deg(2)]
20
+
assert uptime_percent(history, "claude_code") == 50.0
21
+
22
+
23
+
def test_uptime_empty_history():
24
+
assert uptime_percent([], "claude_code") == 0.0
25
+
26
+
27
+
def test_render_bar_operational_uses_green():
28
+
bar = render_bar([_op(1)], "claude_code", width=10)
29
+
assert "green" in bar
30
+
assert "▉" in bar
31
+
32
+
33
+
def test_render_bar_degraded_uses_orange():
34
+
bar = render_bar([_deg(1)], "claude_code", width=10)
35
+
assert "orange3" in bar
36
+
37
+
38
+
def test_render_bar_truncates_to_width():
39
+
history = [_op(i) for i in range(20)]
40
+
# 1 char per bar; width=10 fits exactly 10 bars
41
+
bar = render_bar(history, "claude_code", width=10)
42
+
assert bar.count("▉") == 10
43
+
44
+
45
+
def test_render_bar_no_separator_character():
46
+
history = [_op(i) for i in range(3)]
47
+
bar = render_bar(history, "claude_code", width=20)
48
+
# No explicit separator — relies on natural cell boundary between █ chars
49
+
assert "│" not in bar
50
+
assert " " not in bar
51
+
52
+
53
+
def test_render_bar_shows_all_when_fewer_than_width():
54
+
history = [_op(i) for i in range(3)]
55
+
bar = render_bar(history, "claude_code", width=10)
56
+
assert bar.count("▉") == 3
57
+
58
+
59
+
def test_render_bar_unknown_uses_dim_block():
60
+
record = PollRecord(ts=1, claude_code="unknown", claude_api="operational")
61
+
bar = render_bar([record], "claude_code", width=10)
62
+
assert "░" in bar
63
+
assert "dim" in bar
64
+
65
+
66
+
def test_format_age_seconds():
67
+
from datetime import datetime, timezone, timedelta
68
+
ts = (datetime.now(timezone.utc) - timedelta(seconds=30)).isoformat()
69
+
assert _format_age(ts) == "30s ago"
70
+
71
+
72
+
def test_format_age_minutes():
73
+
from datetime import datetime, timezone, timedelta
74
+
ts = (datetime.now(timezone.utc) - timedelta(minutes=15)).isoformat()
75
+
assert _format_age(ts) == "15 min ago"
76
+
77
+
78
+
def test_format_age_hours():
79
+
from datetime import datetime, timezone, timedelta
80
+
ts = (datetime.now(timezone.utc) - timedelta(hours=3)).isoformat()
81
+
assert _format_age(ts) == "3 hr ago"
82
+
83
+
84
+
def test_format_age_malformed_returns_unknown():
85
+
assert _format_age("not-a-date") == "unknown age"
+63
tests/test_history.py
+63
tests/test_history.py
···
1
+
import json
2
+
from pathlib import Path
3
+
from unittest.mock import patch
4
+
from history import PollRecord, load_history, append_record
5
+
6
+
7
+
def test_load_history_returns_empty_when_file_missing(tmp_path):
8
+
with patch("history.HISTORY_FILE", tmp_path / "history.json"), \
9
+
patch("history.HISTORY_DIR", tmp_path):
10
+
assert load_history() == []
11
+
12
+
13
+
def test_append_and_load_round_trips(tmp_path):
14
+
record = PollRecord(ts=1713196800, claude_code="operational", claude_api="operational")
15
+
with patch("history.HISTORY_FILE", tmp_path / "history.json"), \
16
+
patch("history.HISTORY_DIR", tmp_path):
17
+
append_record(record)
18
+
loaded = load_history()
19
+
assert loaded == [record]
20
+
21
+
22
+
def test_load_history_multiple_records(tmp_path):
23
+
test_file = tmp_path / "history.json"
24
+
test_file.write_text(
25
+
'{"ts": 1713196800, "claude_code": "operational", "claude_api": "operational"}\n'
26
+
'{"ts": 1713196860, "claude_code": "degraded_performance", "claude_api": "operational"}\n'
27
+
)
28
+
with patch("history.HISTORY_FILE", test_file):
29
+
loaded = load_history()
30
+
assert len(loaded) == 2
31
+
assert loaded[0] == PollRecord(ts=1713196800, claude_code="operational", claude_api="operational")
32
+
assert loaded[1] == PollRecord(ts=1713196860, claude_code="degraded_performance", claude_api="operational")
33
+
34
+
35
+
def test_load_history_corrupt_file_returns_empty(tmp_path):
36
+
test_file = tmp_path / "history.json"
37
+
test_file.write_text("not valid json\n")
38
+
with patch("history.HISTORY_FILE", test_file):
39
+
assert load_history() == []
40
+
41
+
42
+
def test_append_creates_directory_if_missing(tmp_path):
43
+
data_dir = tmp_path / "nested" / "dir"
44
+
data_file = data_dir / "history.json"
45
+
record = PollRecord(ts=1, claude_code="operational", claude_api="operational")
46
+
with patch("history.HISTORY_FILE", data_file), \
47
+
patch("history.HISTORY_DIR", data_dir):
48
+
append_record(record)
49
+
assert data_file.exists()
50
+
51
+
52
+
def test_load_history_skips_corrupt_lines_keeps_valid(tmp_path):
53
+
test_file = tmp_path / "history.json"
54
+
test_file.write_text(
55
+
'{"ts": 1, "claude_code": "operational", "claude_api": "operational"}\n'
56
+
'not valid json\n'
57
+
'{"ts": 2, "claude_code": "operational", "claude_api": "operational"}\n'
58
+
)
59
+
with patch("history.HISTORY_FILE", test_file):
60
+
loaded = load_history()
61
+
assert len(loaded) == 2
62
+
assert loaded[0].ts == 1
63
+
assert loaded[1].ts == 2
+83
tests/test_notifier.py
+83
tests/test_notifier.py
···
1
+
import os
2
+
from unittest.mock import patch
3
+
from notifier import detect_transition, notify_cmux, check_and_notify
4
+
5
+
6
+
def test_detect_degraded():
7
+
assert detect_transition("operational", "degraded_performance") == "degraded"
8
+
9
+
10
+
def test_detect_outage():
11
+
assert detect_transition("operational", "partial_outage") == "degraded"
12
+
13
+
14
+
def test_detect_major_outage():
15
+
assert detect_transition("operational", "major_outage") == "degraded"
16
+
17
+
18
+
def test_detect_recovered():
19
+
assert detect_transition("degraded_performance", "operational") == "recovered"
20
+
21
+
22
+
def test_detect_no_change_operational():
23
+
assert detect_transition("operational", "operational") is None
24
+
25
+
26
+
def test_detect_no_change_degraded():
27
+
assert detect_transition("degraded_performance", "partial_outage") is None
28
+
29
+
30
+
def test_notify_skipped_when_not_in_cmux():
31
+
with patch.dict(os.environ, {}, clear=True):
32
+
with patch("notifier.subprocess.run") as mock_run:
33
+
notify_cmux("title", "body")
34
+
mock_run.assert_not_called()
35
+
36
+
37
+
def test_notify_fires_when_in_cmux():
38
+
with patch.dict(os.environ, {"CMUX_WORKSPACE_ID": "workspace:1"}):
39
+
with patch("notifier.subprocess.run") as mock_run:
40
+
notify_cmux("Claude Code degraded", "Degraded Performance")
41
+
mock_run.assert_called_once_with(
42
+
["cmux", "notify", "--title", "Claude Code degraded", "--body", "Degraded Performance"],
43
+
capture_output=True,
44
+
)
45
+
46
+
47
+
def test_check_and_notify_fires_on_degradation():
48
+
with patch.dict(os.environ, {"CMUX_WORKSPACE_ID": "workspace:1"}):
49
+
with patch("notifier.subprocess.run") as mock_run:
50
+
check_and_notify("Claude Code", "operational", "degraded_performance")
51
+
args = mock_run.call_args[0][0]
52
+
assert "Claude Code degraded" in args
53
+
assert "Degraded Performance" in args
54
+
55
+
56
+
def test_check_and_notify_fires_on_recovery():
57
+
with patch.dict(os.environ, {"CMUX_WORKSPACE_ID": "workspace:1"}):
58
+
with patch("notifier.subprocess.run") as mock_run:
59
+
check_and_notify("Claude Code", "degraded_performance", "operational")
60
+
args = mock_run.call_args[0][0]
61
+
assert "Claude Code recovered" in args
62
+
63
+
64
+
def test_check_and_notify_silent_on_no_change():
65
+
with patch.dict(os.environ, {"CMUX_WORKSPACE_ID": "workspace:1"}):
66
+
with patch("notifier.subprocess.run") as mock_run:
67
+
check_and_notify("Claude Code", "operational", "operational")
68
+
mock_run.assert_not_called()
69
+
70
+
71
+
def test_detect_unknown_old_returns_none():
72
+
assert detect_transition("unknown", "operational") is None
73
+
74
+
75
+
def test_detect_unknown_new_returns_none():
76
+
assert detect_transition("operational", "unknown") is None
77
+
78
+
79
+
def test_notify_cmux_binary_missing_does_not_raise():
80
+
with patch.dict(os.environ, {"CMUX_WORKSPACE_ID": "workspace:1"}):
81
+
with patch("notifier.subprocess.run", side_effect=FileNotFoundError):
82
+
# should not raise
83
+
notify_cmux("title", "body")
+116
tests/test_poller.py
+116
tests/test_poller.py
···
1
+
import json
2
+
import urllib.error
3
+
from unittest.mock import MagicMock
4
+
from poller import fetch_status, CLAUDE_CODE_ID, CLAUDE_API_ID, FetchResult
5
+
6
+
7
+
MOCK_COMPONENTS = {
8
+
"components": [
9
+
{"id": CLAUDE_CODE_ID, "name": "Claude Code", "status": "operational"},
10
+
{"id": CLAUDE_API_ID, "name": "Claude API (api.anthropic.com)", "status": "degraded_performance"},
11
+
{"id": "other-id", "name": "claude.ai", "status": "operational"},
12
+
]
13
+
}
14
+
15
+
MOCK_INCIDENTS_EMPTY = {"incidents": []}
16
+
17
+
MOCK_INCIDENTS = {
18
+
"incidents": [
19
+
{
20
+
"id": "abc123",
21
+
"name": "Elevated errors on Claude Code",
22
+
"status": "identified",
23
+
"created_at": "2026-04-15T14:53:02.000Z",
24
+
"impact": "critical",
25
+
}
26
+
]
27
+
}
28
+
29
+
30
+
def _make_response(data: dict) -> MagicMock:
31
+
mock = MagicMock()
32
+
mock.read.return_value = json.dumps(data).encode()
33
+
mock.__enter__ = lambda s: s
34
+
mock.__exit__ = MagicMock(return_value=False)
35
+
return mock
36
+
37
+
38
+
def _make_urlopen(responses: list) -> callable:
39
+
it = iter(responses)
40
+
return lambda url, timeout, **kwargs: next(it)
41
+
42
+
43
+
def test_fetch_status_parses_claude_code_and_api(monkeypatch):
44
+
monkeypatch.setattr(
45
+
"poller.urllib.request.urlopen",
46
+
_make_urlopen([_make_response(MOCK_COMPONENTS), _make_response(MOCK_INCIDENTS_EMPTY)]),
47
+
)
48
+
result = fetch_status()
49
+
assert result.error is None
50
+
assert result.record.claude_code == "operational"
51
+
assert result.record.claude_api == "degraded_performance"
52
+
assert result.incidents == []
53
+
54
+
55
+
def test_fetch_status_parses_incidents(monkeypatch):
56
+
monkeypatch.setattr(
57
+
"poller.urllib.request.urlopen",
58
+
_make_urlopen([_make_response(MOCK_COMPONENTS), _make_response(MOCK_INCIDENTS)]),
59
+
)
60
+
result = fetch_status()
61
+
assert len(result.incidents) == 1
62
+
assert result.incidents[0].name == "Elevated errors on Claude Code"
63
+
assert result.incidents[0].status == "identified"
64
+
assert result.incidents[0].impact == "critical"
65
+
66
+
67
+
def test_fetch_status_network_error_returns_error(monkeypatch):
68
+
monkeypatch.setattr(
69
+
"poller.urllib.request.urlopen",
70
+
lambda url, timeout: (_ for _ in ()).throw(urllib.error.URLError("connection refused")),
71
+
)
72
+
result = fetch_status()
73
+
assert result.error is not None
74
+
assert result.record is None
75
+
assert result.incidents == []
76
+
77
+
78
+
def test_fetch_status_missing_component_returns_unknown(monkeypatch):
79
+
components_without_code = {"components": [
80
+
{"id": CLAUDE_API_ID, "name": "Claude API", "status": "operational"},
81
+
]}
82
+
monkeypatch.setattr(
83
+
"poller.urllib.request.urlopen",
84
+
_make_urlopen([_make_response(components_without_code), _make_response(MOCK_INCIDENTS_EMPTY)]),
85
+
)
86
+
result = fetch_status()
87
+
assert result.record.claude_code == "unknown"
88
+
assert result.record.claude_api == "operational"
89
+
90
+
91
+
def test_fetch_status_records_timestamp(monkeypatch):
92
+
monkeypatch.setattr(
93
+
"poller.urllib.request.urlopen",
94
+
_make_urlopen([_make_response(MOCK_COMPONENTS), _make_response(MOCK_INCIDENTS_EMPTY)]),
95
+
)
96
+
import time
97
+
before = int(time.time())
98
+
result = fetch_status()
99
+
after = int(time.time())
100
+
assert before <= result.record.ts <= after
101
+
102
+
103
+
def test_fetch_status_preserves_record_when_incidents_fail(monkeypatch):
104
+
call_count = 0
105
+
def mock_urlopen(url, timeout, **kwargs):
106
+
nonlocal call_count
107
+
call_count += 1
108
+
if call_count == 1:
109
+
return _make_response(MOCK_COMPONENTS)
110
+
raise urllib.error.URLError("incidents endpoint unreachable")
111
+
monkeypatch.setattr("poller.urllib.request.urlopen", mock_urlopen)
112
+
result = fetch_status()
113
+
assert result.record is not None
114
+
assert result.record.claude_code == "operational"
115
+
assert result.error is not None
116
+
assert result.incidents == []