diff --git a/.claude/rules/core-architecture.md b/.claude/rules/core-architecture.md index b96d7bce..293b8c27 100644 --- a/.claude/rules/core-architecture.md +++ b/.claude/rules/core-architecture.md @@ -74,7 +74,7 @@ joins the session of any sibling span on the same `trace_id`. **Rule: both cache fields belong on both sides of every token sum and every price comparison** — an omission that recurs in `core/optimize/`. **(a) In aggregates**, sum cache reads AND cache writes in the same expression; `COALESCE(SUM(input_tokens + output_tokens + cache_tokens + cache_write_tokens), 0)` (the form in `optimize/runner.py`) is canonical for per-cluster and per-session totals — grep `SUM.*cache_tokens` when adding one. **(b) In pricing comparisons**, an actual-vs-alternative-model savings figure must price cache identically on both sides: a hand-rolled alt-price helper omitting cache-write while the actual `cost_usd` bills it inflates every derived saving by the candidate's full cache-write cost, and evades that grep because the SQL is correct and the Python pricing is not — the omission lives in a per-row helper, not a `SUM`. **Keep exactly one pricer:** route every alternative-side computation through `calculate_cost`, which covers all four token classes. -- **`alerts.py`** — `AlertEngine` (the kinds it can fire are the `AlertType` enum in `core/models.py`; read the enum, never a count kept here), `CooldownTracker` (in-memory, per agent+type, resets on restart), `AlertDispatcher` routing to the channel types the `_build_channel` factory knows (stdout, file, ntfy, webhook, Discord, Telegram — that factory is the live set). `AlertEngine.fire()` is the entry point for `SchemaValidator` and `DriftDetector`. Suppressed alerts are still persisted, just not dispatched. Thresholds are module-level constants, not config — read them for values: **retry loop** fires when the same tool call repeats often enough inside a trailing span window (`_RETRY_LOOP_THRESHOLD` / `_RETRY_LOOP_WINDOW`); **failure rate** when the error share over a trailing span window crosses a ceiling, re-evaluated only every Nth error to bound cost (`_FAILURE_RATE_THRESHOLD` / `_FAILURE_RATE_WINDOW` / `_FAILURE_RATE_CHECK_INTERVAL`); **session duration** past a default wall-clock ceiling (`_SESSION_DURATION_DEFAULT`). Stdout and file channels always include full detail regardless of `include_captured_content`. +- **`alerts.py`** — `AlertEngine` (the kinds it can fire are the `AlertType` enum in `core/models.py`; read the enum, never a count kept here), `CooldownTracker` (in-memory, per agent+type, resets on restart), `AlertDispatcher` routing to the channel types the `_build_channel` factory knows (stdout, file, ntfy, webhook, Discord, Telegram, Mast — that factory is the live set). `AlertEngine.fire()` is the entry point for `SchemaValidator` and `DriftDetector`. Suppressed alerts are still persisted, just not dispatched. Thresholds are module-level constants, not config — read them for values: **retry loop** fires when the same tool call repeats often enough inside a trailing span window (`_RETRY_LOOP_THRESHOLD` / `_RETRY_LOOP_WINDOW`); **failure rate** when the error share over a trailing span window crosses a ceiling, re-evaluated only every Nth error to bound cost (`_FAILURE_RATE_THRESHOLD` / `_FAILURE_RATE_WINDOW` / `_FAILURE_RATE_CHECK_INTERVAL`); **session duration** past a default wall-clock ceiling (`_SESSION_DURATION_DEFAULT`). Stdout and file channels always include full detail regardless of `include_captured_content`. - **`drift.py`** — `DriftDetector`, Z-score based behavioral drift detection, fires at session end. diff --git a/docs/alerts.md b/docs/alerts.md index 0f416a3c..b316a16a 100644 --- a/docs/alerts.md +++ b/docs/alerts.md @@ -30,6 +30,12 @@ Configure where alerts go in `.tj/config.toml`. Multiple channels work simultane type = "ntfy" topic = "my-agent-alerts" +# Mast pager app for iPhone and Android. Free, with or without an account (limits below). +# The key in the URL is the only credential. +[[alerts.channels]] +type = "mast" +channel_url = "https://mast.tissue.dev/mk_0123456789abcdef0123456789abcdef01234567" + # Discord webhook [[alerts.channels]] type = "discord" @@ -56,6 +62,51 @@ path = "~/.tj/alerts.log" type = "stdout" ``` +### Mast + +[Mast](https://tissue.systems/mast) is a pager app for iPhone and Android. Install it, make a +channel, and paste the channel URL into the config above. It works without an account, but a +channel made that way is lost if the app is reinstalled, and the only sign on this side is a +"No such channel." line in the log. Sign in to a free Tissue account if you want the channel +to survive that. + +Both ways are free but capped. With no account you get 2 channels and 1,000 messages a month; +a free account gets 10 channels and 10,000 messages. Mast keeps accepting sends past the monthly +number, but if your agents alert a lot, check [Mast's limits](https://tissue.systems/mast/docs/limits/) +before relying on it. + +`tj` posts a JSON body with `title`, `body`, `priority` and `key`. Mast documents those fields +on its [sending page](https://tissue.systems/mast/docs/send/), which is the place to look if +sends start failing. + +`tj` maps severity onto three of Mast's four priorities: `info` sends `quiet`, `warning` +sends `normal`, `critical` sends `loud`. Nothing maps to the fourth, `page`. It repeats until +someone acks it and ignores mute and quiet hours, so you have to ask for it, which is what +you'd do on a channel that's meant to wake someone up: + +```toml +[[alerts.channels]] +type = "mast" +channel_url = "https://mast.tissue.dev/mk_..." +priority = "page" +min_severity = "critical" +``` + +Each channel also has a ceiling, shown as **Highest priority** in the app, and a send that +asks for more than that is delivered at the ceiling. It's set when the channel is created. The +iOS app doesn't go below `loud`, the Android app uses whatever default priority you picked, and +a channel made from an API token starts at `loud`. So a channel created on Android with the +default of `Normal` will deliver `critical` alerts as `normal`, and `priority = "page"` won't +page anyone until you raise the ceiling on the phone. Worth checking before you rely on it. + +Each send carries a dedupe key built from the same things the cooldown uses: the alert type, +the agent, and for `cost_budget_session` and `session_duration` the session too. If the same +alert fires again inside the channel's dedupe window (5 minutes unless you change it), Mast +adds to the count on the card already on the phone instead of sending a new one. Two sessions +over their own budget are different alerts, so they get separate cards. The default cooldown +is 60 seconds, so a repeat usually lands inside the window; if you raise `cooldown_seconds` +past the window, repeats arrive as new cards. + ## Sensitive actions Define which tool calls should trigger immediate alerts: @@ -88,7 +139,7 @@ Cooldown state survives a daemon restart: `AlertEngine` rehydrates it from the p ## Content stripping -By default, alert payloads sent to external channels (Discord, Telegram, webhook, ntfy) have sensitive content stripped: `prompt_content`, `completion_content`, `tool_input`, `tool_output`. To include full content: +By default, alert payloads sent to external channels (Discord, Telegram, webhook, ntfy, Mast) have sensitive content stripped: `prompt_content`, `completion_content`, `tool_input`, `tool_output`. To include full content: ```toml [alerts] diff --git a/docs/architecture.md b/docs/architecture.md index 217c1232..9fb9844e 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -35,7 +35,7 @@ flowchart TD Ingest["IngestPipeline\nSanitization · Session continuity · Attribute extraction"] Ingest --> Cost["CostEngine\npricing.toml"] - Ingest --> Alerts["AlertEngine\n13 types · 6 channels"] + Ingest --> Alerts["AlertEngine\n13 types · 7 channels"] Ingest --> Schema["SchemaValidator\nexplicit JSON Schema"] Cost --> DB["DuckDB\nlocal · embedded"] @@ -179,7 +179,7 @@ The alert engine evaluates rules after every span ingest. Alert types fall into ### Dispatch and cooldown -`AlertDispatcher` routes fired alerts to 6 channel types: stdout, file, ntfy, webhook, Discord, Telegram. `CooldownTracker` prevents alert storms by suppressing repeat alerts of the same type for the same agent within a configurable window. Suppressed alerts are still persisted to the database but not dispatched. +`AlertDispatcher` routes fired alerts to 7 channel types: stdout, file, ntfy, webhook, Discord, Telegram, Mast. `CooldownTracker` prevents alert storms by suppressing repeat alerts of the same type for the same agent within a configurable window. Suppressed alerts are still persisted to the database but not dispatched. Content stripping removes `prompt_content`, `completion_content`, `tool_input`, `tool_output` from payloads sent to external channels unless `alerts.include_captured_content = true`. Stdout and file channels always receive the full payload. diff --git a/docs/configuration.md b/docs/configuration.md index c908bdcf..06a7912a 100644 --- a/docs/configuration.md +++ b/docs/configuration.md @@ -91,6 +91,10 @@ include_captured_content = false type = "discord" webhook_url = "https://discord.com/api/webhooks/..." + [[alerts.channels]] + type = "mast" + channel_url = "https://mast.tissue.dev/mk_..." # iPhone/Android pager app, free; see docs/alerts.md + [export] [export.otlp] endpoint = "" # set to forward spans to an external OTel backend @@ -225,7 +229,7 @@ The four flags are independent: capture prompts without completions, or tool inp **Where the gate runs.** All spans flow through `IngestPipeline.process()`, which calls `strip_captured_content()` once before any DB write. The receiver, the OTLP HTTP endpoint, and the Claude Code JSONL backfill all go through this single gate — there's no per-source duplication. -**Precedence with alert channels.** `alerts.include_captured_content = true` lets external alert channels (webhook, Discord, Telegram, ntfy) receive full payloads. If the corresponding `[capture]` flag is `false`, the content is already stripped at ingest, so the alert flag is moot — channels can't include what was never captured. Stdout and file alert channels always receive full payloads from the in-memory span before the DB strip, so they still see content during the brief in-process window. +**Precedence with alert channels.** `alerts.include_captured_content = true` lets external alert channels (webhook, Discord, Telegram, ntfy, Mast) receive full payloads. If the corresponding `[capture]` flag is `false`, the content is already stripped at ingest, so the alert flag is moot — channels can't include what was never captured. Stdout and file alert channels always receive full payloads from the in-memory span before the DB strip, so they still see content during the brief in-process window. **What this means for downstream analyzers.** diff --git a/docs/nemoclaw-integration.md b/docs/nemoclaw-integration.md index f217c570..5522074f 100644 --- a/docs/nemoclaw-integration.md +++ b/docs/nemoclaw-integration.md @@ -22,7 +22,7 @@ The NemoClaw observer listens on the OpenShell Gateway WebSocket for sandbox enf | Syscall denied | `syscall_denied` | critical | | Inference rerouted | `inference_rerouted` | warning | -These alerts flow through the standard alert pipeline — cooldown, dispatch to configured channels (ntfy, Discord, Telegram, webhook), and persistence to the database. +These alerts flow through the standard alert pipeline — cooldown, dispatch to configured channels (ntfy, Mast, Discord, Telegram, webhook), and persistence to the database. ## Configuration diff --git a/tests/unit/test_cmd_policy.py b/tests/unit/test_cmd_policy.py index 13ef8cdf..937f6b01 100644 --- a/tests/unit/test_cmd_policy.py +++ b/tests/unit/test_cmd_policy.py @@ -157,6 +157,24 @@ def test_alerts_row_summarises_channel_count_and_flags(): assert "path=/tmp/alerts.log" in ch1.setting +def test_channel_summary_redacts_the_mast_channel_key(): + # The URL carries the channel key, so it must not show up in the table. + summary = _channel_summary(AlertChannelConfig( + type="mast", + channel_url="https://mast.tissue.dev/mk_0123456789abcdef", + )) + assert "type=mast" in summary + assert "channel_url=" in summary + assert "mk_0123456789abcdef" not in summary + + +def test_channel_summary_shows_a_mast_priority_override(): + summary = _channel_summary(AlertChannelConfig( + type="mast", channel_url="https://mast.tissue.dev/mk_x", priority="page", + )) + assert "priority=page" in summary + + def test_drift_overrides_default_detects_change(): assert _drift_overrides_default(DriftConfig()) is False assert _drift_overrides_default(DriftConfig(token_threshold=3.0)) is True diff --git a/tests/unit/test_config.py b/tests/unit/test_config.py index 989662c5..9d81a5bc 100644 --- a/tests/unit/test_config.py +++ b/tests/unit/test_config.py @@ -246,6 +246,24 @@ def test_alerts_channels_parsed(self): assert len(config.alerts.channels) == 2 assert config.alerts.channels[1].topic == "my-topic" + def test_alerts_mast_channel_parsed(self): + raw = { + "alerts": { + "channels": [ + { + "type": "mast", + "channel_url": "https://mast.tissue.dev/mk_abc", + "priority": "page", + }, + ], + } + } + config = _parse(raw) + channel = config.alerts.channels[0] + assert channel.type == "mast" + assert channel.channel_url == "https://mast.tissue.dev/mk_abc" + assert channel.priority == "page" + def test_default_alert_channel_is_stdout(self): config = _parse({}) assert len(config.alerts.channels) == 1 diff --git a/tests/unit/test_mast_channel.py b/tests/unit/test_mast_channel.py new file mode 100644 index 00000000..2320281d --- /dev/null +++ b/tests/unit/test_mast_channel.py @@ -0,0 +1,283 @@ +"""Mast alert channel. httpx is faked, nothing leaves the process.""" +from __future__ import annotations + +import logging +from typing import Any + +import pytest + +from tokenjam.core.alerts import CooldownTracker, MastChannel, _build_channel +from tokenjam.core.config import AlertChannelConfig +from tokenjam.core.models import Alert, AlertType, Severity +from tokenjam.utils.time_parse import utcnow + +CHANNEL_URL = "https://mast.tissue.dev/mk_0123456789abcdef0123456789abcdef01234567" + + +class _FakeResponse: + def __init__(self, status_code: int = 202, body: Any = None) -> None: + self.status_code = status_code + self._body = body + + def json(self) -> Any: + if self._body is None: + raise ValueError("no JSON body") + return self._body + + +class _FakeClient: + """Stands in for httpx.Client, recording the one POST a send makes.""" + + calls: list[dict[str, Any]] = [] + status_code = 202 + body: Any = None + + def __init__(self, *args: Any, **kwargs: Any) -> None: + pass + + def __enter__(self) -> _FakeClient: + return self + + def __exit__(self, *exc: Any) -> None: + return None + + def post(self, url: str, json: dict[str, Any] | None = None, **kwargs: Any) -> _FakeResponse: + type(self).calls.append({"url": url, "json": json}) + return _FakeResponse(type(self).status_code, type(self).body) + + +@pytest.fixture +def fake_http(monkeypatch: pytest.MonkeyPatch) -> type[_FakeClient]: + _FakeClient.calls = [] + _FakeClient.status_code = 202 + _FakeClient.body = None + monkeypatch.setattr("tokenjam.core.alerts.httpx.Client", _FakeClient) + return _FakeClient + + +def _alert( + severity: Severity = Severity.CRITICAL, + detail: dict[str, Any] | None = None, + agent_id: str | None = "my-agent", + title: str = "Budget exceeded", + alert_type: AlertType = AlertType.COST_BUDGET_DAILY, + session_id: str | None = None, +) -> Alert: + return Alert( + alert_id="al_1", + fired_at=utcnow(), + type=alert_type, + severity=severity, + title=title, + detail=detail if detail is not None else {"message": "daily spend over limit"}, + agent_id=agent_id, + session_id=session_id, + ) + + +def _channel(**overrides: Any) -> MastChannel: + config = AlertChannelConfig(type="mast", channel_url=CHANNEL_URL, **overrides) + return MastChannel(config, include_captured_content=False) + + +# ── Factory registration ────────────────────────────────────────────────── + +def test_build_channel_returns_a_mast_channel_for_the_mast_type(): + ch = _build_channel( + AlertChannelConfig(type="mast", channel_url=CHANNEL_URL), + include_captured_content=False, + ) + assert isinstance(ch, MastChannel) + assert ch.channel_url == CHANNEL_URL + + +def test_min_severity_is_set_for_the_dispatcher(): + # AlertDispatcher.dispatch() reads min_severity off the channel; without + # it every severity gets through. + ch = _channel(min_severity="critical") + assert ch.min_severity is Severity.CRITICAL + + +# ── Priority mapping ────────────────────────────────────────────────────── + +@pytest.mark.parametrize( + ("severity", "expected"), + [ + (Severity.INFO, "quiet"), + (Severity.WARNING, "normal"), + (Severity.CRITICAL, "loud"), + ], +) +def test_severity_maps_onto_a_mast_priority(fake_http, severity, expected): + _channel().send(_alert(severity=severity)) + assert fake_http.calls[0]["json"]["priority"] == expected + + +def test_no_severity_maps_to_page(fake_http): + # page repeats until acked and ignores quiet hours, so you have to ask + # for it. + for severity in (Severity.INFO, Severity.WARNING, Severity.CRITICAL): + _channel().send(_alert(severity=severity)) + assert all(call["json"]["priority"] != "page" for call in fake_http.calls) + + +def test_configured_priority_overrides_the_severity_mapping(fake_http): + _channel(priority="page").send(_alert(severity=Severity.INFO)) + assert fake_http.calls[0]["json"]["priority"] == "page" + + +def test_a_misspelt_priority_fails_at_startup(): + # Otherwise it's a 400 in the log the first time something fires. + with pytest.raises(ValueError, match="quiet, normal, loud, page"): + _channel(priority="high") + + +# ── Payload ─────────────────────────────────────────────────────────────── + +def test_send_posts_to_the_channel_url_with_title_and_body(fake_http): + _channel().send(_alert()) + call = fake_http.calls[0] + assert call["url"] == CHANNEL_URL + assert call["json"]["title"] == "Budget exceeded" + assert call["json"]["body"] == "daily spend over limit" + + +def _key(fake_http, **alert_args: Any) -> str: + _channel().send(_alert(**alert_args)) + return fake_http.calls[-1]["json"]["key"] + + +def test_repeats_of_the_same_alert_share_a_key(fake_http): + first = _key(fake_http) + assert first.startswith("cost_budget_daily:") + assert _key(fake_http) == first + + +def test_an_alert_with_no_agent_still_gets_a_key(fake_http): + assert _key(fake_http, agent_id=None).startswith("cost_budget_daily:") + + +def test_different_agents_get_different_keys(fake_http): + assert _key(fake_http, agent_id="a") != _key(fake_http, agent_id="b") + + +def test_two_sessions_over_budget_get_two_cards(fake_http): + # Mast would fold the second into the first and only bump its counter, + # so the second session's alert would never show. + one = _key(fake_http, alert_type=AlertType.COST_BUDGET_SESSION, session_id="s1") + two = _key(fake_http, alert_type=AlertType.COST_BUDGET_SESSION, session_id="s2") + assert one != two + + +def test_session_id_is_ignored_for_types_that_are_not_per_session(fake_http): + one = _key(fake_http, session_id="s1") + two = _key(fake_http, session_id="s2") + assert one == two + + +def test_long_agent_ids_with_a_shared_prefix_dont_collide(fake_http): + prefix = "x" * 200 + assert _key(fake_http, agent_id=prefix + "-a") != _key(fake_http, agent_id=prefix + "-b") + + +def test_key_fits_in_mast_limit(fake_http): + longest = max(AlertType, key=lambda t: len(t.value)) + key = _key(fake_http, agent_id="x" * 1000, alert_type=longest, session_id="y" * 1000) + assert len(key) <= 120 + + +@pytest.mark.parametrize( + ("first", "second"), + [ + ({}, {}), + ({}, {"agent_id": "other"}), + ({}, {"alert_type": AlertType.RETRY_LOOP}), + ({"session_id": "s1"}, {"session_id": "s2"}), + ( + {"alert_type": AlertType.COST_BUDGET_SESSION, "session_id": "s1"}, + {"alert_type": AlertType.COST_BUDGET_SESSION, "session_id": "s2"}, + ), + ( + {"alert_type": AlertType.SESSION_DURATION, "session_id": "s1"}, + {"alert_type": AlertType.SESSION_DURATION, "session_id": "s1"}, + ), + ], +) +def test_key_matches_what_the_cooldown_treats_as_the_same_alert(fake_http, first, second): + a, b = _alert(**first), _alert(**second) + tracker = CooldownTracker(cooldown_seconds=300) + tracker.record(a.agent_id, a.type, a.session_id) + same_for_cooldown = tracker.is_suppressed(b.agent_id, b.type, b.session_id) + assert (_key(fake_http, **first) == _key(fake_http, **second)) == same_for_cooldown + + +def test_title_and_body_are_cut_to_mast_limits(fake_http): + # Past the limit Mast answers 400 and the alert is lost. + _channel().send(_alert(title="t" * 500, detail={"message": "b" * 9000})) + payload = fake_http.calls[0]["json"] + assert len(payload["title"]) == MastChannel.TITLE_MAXLEN + assert len(payload["body"]) == MastChannel.BODY_MAXLEN + + +def test_a_channel_with_no_url_sends_nothing(fake_http): + MastChannel(AlertChannelConfig(type="mast"), include_captured_content=False).send(_alert()) + assert fake_http.calls == [] + + +# ── Critical Rule 5: captured-content stripping ─────────────────────────── + +def test_captured_content_is_stripped_by_default(fake_http): + detail = {"prompt_content": "secret prompt", "tool_output": "secret output", "cost": 1.5} + _channel().send(_alert(detail=detail)) + body = fake_http.calls[0]["json"]["body"] + assert "secret prompt" not in body + assert "secret output" not in body + assert "1.5" in body + + +def test_captured_content_is_included_when_the_flag_is_set(fake_http): + detail = {"prompt_content": "secret prompt", "cost": 1.5} + config = AlertChannelConfig(type="mast", channel_url=CHANNEL_URL) + MastChannel(config, include_captured_content=True).send(_alert(detail=detail)) + assert "secret prompt" in fake_http.calls[0]["json"]["body"] + + +# ── Delivery failure is reported ────────────────────────────────────────── + +def test_a_rejected_send_logs_the_status_and_message(fake_http, caplog): + # Mast's error body is {"error": {"code", "message"}}; the message is + # the part that tells you it was a rotated key. + fake_http.status_code = 404 + fake_http.body = {"error": {"code": "not_found", "message": "No such channel."}} + with caplog.at_level(logging.WARNING, logger="tokenjam.core.alerts"): + _channel().send(_alert()) + assert "rejected" in caplog.text + assert "404" in caplog.text + assert "No such channel." in caplog.text + + +def test_a_rejection_without_a_json_body_still_logs_the_status(fake_http, caplog): + fake_http.status_code = 429 + fake_http.body = None + with caplog.at_level(logging.WARNING, logger="tokenjam.core.alerts"): + _channel().send(_alert()) + assert "429" in caplog.text + + +def test_a_5xx_is_not_logged_as_a_rejection(fake_http, caplog): + # A 502 may have come after the message was stored. Call it rejected + # and whoever reads the log retries and pages twice. + fake_http.status_code = 502 + fake_http.body = {"error": {"code": "storage_error", "message": "The message could not be stored."}} + with caplog.at_level(logging.WARNING, logger="tokenjam.core.alerts"): + _channel().send(_alert()) + assert "502" in caplog.text + assert "may or may not have gone out" in caplog.text + assert "rejected" not in caplog.text + + +def test_an_accepted_send_logs_nothing(fake_http, caplog): + fake_http.status_code = 202 + with caplog.at_level(logging.WARNING, logger="tokenjam.core.alerts"): + _channel().send(_alert()) + assert caplog.text == "" diff --git a/tokenjam/cli/cmd_policy.py b/tokenjam/cli/cmd_policy.py index 53ef68b2..e8abce69 100644 --- a/tokenjam/cli/cmd_policy.py +++ b/tokenjam/cli/cmd_policy.py @@ -318,6 +318,13 @@ def _channel_summary(ch: AlertChannelConfig) -> str: parts.append("webhook_url=") elif ch.type == "telegram" and ch.chat_id: parts.append(f"chat_id={ch.chat_id}") + elif ch.type == "mast": + if ch.priority: + parts.append(f"priority={ch.priority}") + if ch.channel_url: + # The URL contains the channel key, which is the whole + # credential, so don't print it. + parts.append("channel_url=") return ", ".join(parts) diff --git a/tokenjam/core/alerts.py b/tokenjam/core/alerts.py index f40a384e..5af73378 100644 --- a/tokenjam/core/alerts.py +++ b/tokenjam/core/alerts.py @@ -5,6 +5,7 @@ """ from __future__ import annotations +import hashlib import json import logging import threading @@ -57,6 +58,20 @@ AlertType.SESSION_DURATION, }) + +def _alert_identity( + agent_id: str | None, alert_type: AlertType, session_id: str | None = None, +) -> tuple[str, str, str]: + """What makes two alerts "the same one" for cooldown purposes. + + Session-scoped types include the session id, so two sessions of one agent + that each go over their own budget are two alerts. Everything else is per + agent and type. MastChannel builds its dedupe key from this too. + """ + scope = session_id if alert_type in _SESSION_SCOPED_ALERT_TYPES else "" + return (agent_id or "", alert_type.value, scope or "") + + # Row cap for the startup queries that rebuild CooldownTracker/_failure_rate_fired # from the alerts table (#592). Generous rather than exact: missing a stale row # only means an old alert is treated as new, the same behavior as before this fix. @@ -213,9 +228,7 @@ def __init__(self, cooldown_seconds: int = 60) -> None: def is_suppressed( self, agent_id: str | None, alert_type: AlertType, session_id: str | None = None, ) -> bool: - scope = session_id if alert_type in _SESSION_SCOPED_ALERT_TYPES else "" - key = (agent_id or "", alert_type.value, scope or "") - last = self._last_fired.get(key) + last = self._last_fired.get(_alert_identity(agent_id, alert_type, session_id)) if last is None: return False return (utcnow() - last).total_seconds() < self.cooldown_seconds @@ -223,9 +236,7 @@ def is_suppressed( def record( self, agent_id: str | None, alert_type: AlertType, session_id: str | None = None, ) -> None: - scope = session_id if alert_type in _SESSION_SCOPED_ALERT_TYPES else "" - key = (agent_id or "", alert_type.value, scope or "") - self._last_fired[key] = utcnow() + self._last_fired[_alert_identity(agent_id, alert_type, session_id)] = utcnow() def hydrate(self, alerts: list[Alert]) -> None: """Seed `_last_fired` from persisted, non-suppressed alert rows. @@ -237,8 +248,7 @@ def hydrate(self, alerts: list[Alert]) -> None: for alert in alerts: if alert.suppressed: continue - scope = alert.session_id if alert.type in _SESSION_SCOPED_ALERT_TYPES else "" - key = (alert.agent_id or "", alert.type.value, scope or "") + key = _alert_identity(alert.agent_id, alert.type, alert.session_id) existing = self._last_fired.get(key) if existing is None or alert.fired_at > existing: self._last_fired[key] = alert.fired_at @@ -767,6 +777,8 @@ def _build_channel( return DiscordChannel(config, include_captured_content) case "telegram": return TelegramChannel(config, include_captured_content) + case "mast": + return MastChannel(config, include_captured_content) case _: raise ValueError(f"Unknown alert channel type: {config.type!r}") @@ -966,10 +978,101 @@ def send(self, alert: Alert) -> None: client.post(url, json=payload) +class MastChannel(AlertChannel): + """POST to a Mast channel URL, which pages a phone (https://tissue.systems/mast).""" + + # Mast refuses a longer title or body with a 400 naming the field. + TITLE_MAXLEN = 250 + BODY_MAXLEN = 4096 + + PRIORITIES = ("quiet", "normal", "loud", "page") + + def __init__(self, config: AlertChannelConfig, include_captured_content: bool) -> None: + self.channel_url = config.channel_url or "" + # Checked here, like min_severity, so a typo fails at startup instead + # of as a 400 in the log the first time an alert fires. + if config.priority is not None and config.priority not in self.PRIORITIES: + raise ValueError( + f"Mast channel priority must be one of {', '.join(self.PRIORITIES)}, " + f"not {config.priority!r}" + ) + self.priority = config.priority + self._include_captured_content = include_captured_content + self.min_severity = Severity(config.min_severity) + + def send(self, alert: Alert) -> None: + if not self.channel_url: + return + body = _format_detail_text(alert, strip=not self._include_captured_content) + payload = { + "title": alert.title[: self.TITLE_MAXLEN], + "body": body[: self.BODY_MAXLEN], + "priority": self.priority or _MAST_SEVERITY_PRIORITY.get(alert.severity, "normal"), + "key": _mast_dedupe_key(alert), + } + with httpx.Client(timeout=5.0) as client: + response = client.post(self.channel_url, json=payload) + # 202 means stored. Log anything else, because nobody notices a + # pager that fails quietly, and include the error message from the + # body: it says what was actually wrong (rotated key, field over its + # limit, rate limit) where the status alone doesn't. A 5xx gets + # different wording since the message may have been stored before + # things went wrong, and retrying on the strength of the log could + # page someone twice. + if response.status_code in (200, 202): + return + detail = _mast_error_message(response) + if response.status_code >= 500: + logger.warning( + "Mast gave HTTP %s for alert %s; it may or may not have gone out%s", + response.status_code, alert.alert_id, f": {detail}" if detail else "", + ) + else: + logger.warning( + "Mast rejected alert %s (HTTP %s%s)", + alert.alert_id, response.status_code, f": {detail}" if detail else "", + ) + + # ── Helpers ──────────────────────────────────────────────────────────────── _SEVERITY_RANK = {Severity.INFO: 0, Severity.WARNING: 1, Severity.CRITICAL: 2} +def _mast_error_message(response: Any) -> str: + """Pull error.message out of a Mast error body. Empty string if there isn't one.""" + try: + payload = response.json() + except Exception: + return "" + error = payload.get("error") if isinstance(payload, dict) else None + if isinstance(error, dict): + return str(error.get("message") or "") + return "" + + +def _mast_dedupe_key(alert: Alert) -> str: + """Dedupe key for a Mast send: the alert type, then a hash of its identity. + + Mast folds a repeat of a key into the card already on the phone (it bumps + a counter and keeps the first card's text), so the key has to tell apart + exactly the alerts the cooldown tells apart. Hashed because agent ids can + be any length and Mast caps keys at 120 characters; cutting the string + instead would merge two long agent ids that share a prefix. + """ + agent, kind, scope = _alert_identity(alert.agent_id, alert.type, alert.session_id) + digest = hashlib.sha256("\0".join((kind, agent, scope)).encode()).hexdigest() + return f"{kind}:{digest[:32]}" + + +# Mast has a fourth priority, "page", that repeats until someone acks it and +# ignores mute and quiet hours. Nothing maps to it; you have to ask for it +# with priority= on the channel. +_MAST_SEVERITY_PRIORITY = { + Severity.INFO: "quiet", + Severity.WARNING: "normal", + Severity.CRITICAL: "loud", +} + def _severity_rank(sev: Severity) -> int: return _SEVERITY_RANK.get(sev, 0) diff --git a/tokenjam/core/config.py b/tokenjam/core/config.py index ffb6cc01..a072636c 100644 --- a/tokenjam/core/config.py +++ b/tokenjam/core/config.py @@ -159,6 +159,13 @@ class AlertChannelConfig: # telegram bot_token: str | None = None chat_id: str | None = None + # mast + channel_url: str | None = None + # Pins every alert on this channel to one Mast priority. Leave it unset + # and severity picks: info -> quiet, warning -> normal, critical -> loud. + # "page" repeats until acked and ignores quiet hours, so nothing maps to + # it on its own; this is the only way to get it. + priority: str | None = None # shared min_severity: str = "info"