diff --git a/.env.example b/.env.example index 25cf320192..e910bf8251 100644 --- a/.env.example +++ b/.env.example @@ -396,6 +396,12 @@ OPERATOR_QUEUE_EXECUTION_ID_MAX=128 # max execution_id preserved in a OPERATOR_QUEUE_EMAIL_MAX=320 # ent#364: addressee email length clamp (RFC 5321) OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS=300 # one flood alert per agent per this window OPERATOR_ALERT_MAX_PENDING_PER_TYPE=5 # per-(agent, type) pending budget for agent-influenceable platform alerts (#1677) +# #3246: a platform alert is one row per condition — it expires this many days after +# its LAST reading (bad value → default + WARNING) … +OPERATOR_PLATFORM_ALERT_LIFETIME_DAYS=14 +# … and after a person ends one, the same reading files nothing for this many days +# unless its priority rises or a material field changes. +OPERATOR_PLATFORM_ALERT_SNOOZE_DAYS=7 # =========================================== # SERVICE URLS (Usually no need to change) diff --git a/docker-compose.hosted.yml b/docker-compose.hosted.yml index 230258cf0b..018e405ee9 100644 --- a/docker-compose.hosted.yml +++ b/docker-compose.hosted.yml @@ -283,6 +283,8 @@ services: - OPERATOR_QUEUE_EMAIL_MAX=${OPERATOR_QUEUE_EMAIL_MAX:-320} # ent#364: addressee email length clamp (RFC 5321) - OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS=${OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS:-300} # one flood alert / agent / window - OPERATOR_ALERT_MAX_PENDING_PER_TYPE=${OPERATOR_ALERT_MAX_PENDING_PER_TYPE:-5} # per-(agent,type) platform-alert budget (#1677) + - OPERATOR_PLATFORM_ALERT_LIFETIME_DAYS=${OPERATOR_PLATFORM_ALERT_LIFETIME_DAYS:-14} # #3246: platform alert expires N days after its last reading + - OPERATOR_PLATFORM_ALERT_SNOOZE_DAYS=${OPERATOR_PLATFORM_ALERT_SNOOZE_DAYS:-7} # #3246: a person's ending holds N days unless priority rises - GITHUB_PAT=${GITHUB_PAT} # Log Retention & Archival Configuration (#1039) # These were MISSING from prod — prod launches standalone (no base-compose diff --git a/docker-compose.prod.yml b/docker-compose.prod.yml index 0b1a4b386d..7e56bfff0d 100644 --- a/docker-compose.prod.yml +++ b/docker-compose.prod.yml @@ -220,6 +220,8 @@ services: - OPERATOR_QUEUE_EMAIL_MAX=${OPERATOR_QUEUE_EMAIL_MAX:-320} # ent#364: addressee email length clamp (RFC 5321) - OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS=${OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS:-300} # one flood alert / agent / window - OPERATOR_ALERT_MAX_PENDING_PER_TYPE=${OPERATOR_ALERT_MAX_PENDING_PER_TYPE:-5} # per-(agent,type) platform-alert budget (#1677) + - OPERATOR_PLATFORM_ALERT_LIFETIME_DAYS=${OPERATOR_PLATFORM_ALERT_LIFETIME_DAYS:-14} # #3246: platform alert expires N days after its last reading + - OPERATOR_PLATFORM_ALERT_SNOOZE_DAYS=${OPERATOR_PLATFORM_ALERT_SNOOZE_DAYS:-7} # #3246: a person's ending holds N days unless priority rises - GITHUB_PAT=${GITHUB_PAT} # Log Retention & Archival Configuration (#1039) # These were MISSING from prod — prod launches standalone (no base-compose diff --git a/docker-compose.yml b/docker-compose.yml index 46c4ba204b..4115503d58 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -162,6 +162,8 @@ services: - OPERATOR_QUEUE_EMAIL_MAX=${OPERATOR_QUEUE_EMAIL_MAX:-320} # ent#364: addressee email length clamp (RFC 5321) - OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS=${OPERATOR_QUEUE_FLOOD_ALERT_COOLDOWN_SECONDS:-300} # one flood alert / agent / window - OPERATOR_ALERT_MAX_PENDING_PER_TYPE=${OPERATOR_ALERT_MAX_PENDING_PER_TYPE:-5} # per-(agent,type) platform-alert budget (#1677) + - OPERATOR_PLATFORM_ALERT_LIFETIME_DAYS=${OPERATOR_PLATFORM_ALERT_LIFETIME_DAYS:-14} # #3246: platform alert expires N days after its last reading + - OPERATOR_PLATFORM_ALERT_SNOOZE_DAYS=${OPERATOR_PLATFORM_ALERT_SNOOZE_DAYS:-7} # #3246: a person's ending holds N days unless priority rises - GITHUB_PAT=${GITHUB_PAT:-} - HOST_TEMPLATES_PATH=${PWD}/config/agent-templates # First-run default system seed (trinity-enterprise#124): a path to an diff --git a/docs/memory/architecture/backend.md b/docs/memory/architecture/backend.md index 0d2a66891b..529e226bd3 100644 --- a/docs/memory/architecture/backend.md +++ b/docs/memory/architecture/backend.md @@ -178,6 +178,7 @@ - `monitoring_service.py` - Fleet-wide health monitoring, 30s loop — authoritative for aggregate status; lifespan-resumed from persisted `monitoring_config`, default OFF (MON-001, #1121) - `monitoring_alerts.py` - Alert threshold configuration - `heartbeat_service.py` - Agent push-heartbeat liveness layer — see [Heartbeat Liveness](reliability.md#heartbeat-liveness-reliability-004-307) +- `platform_alerts.py` - The platform-alert seam (#3246, OPS-001-PLATFORM-ALERTS, requirements §26.14). Stdlib-only leaf (DB imported lazily): the `Kind` registry (prefix, lifetime class, clear hook, `material_keys`), `observe(agent, kind, key, …)` → at most one pending row per subject `:` updated in place (`last_seen_at`, `context.seen_count`), `clear` / `reconcile` → ended `cancelled` by the **platform** (`condition_cleared`), lifetime (`OPERATOR_PLATFORM_ALERT_LIFETIME_DAYS`, 14) and snooze (`OPERATOR_PLATFORM_ALERT_SNOOZE_DAYS`, 7) readers, and the pure `plan_sweep` the upgrade migration runs. Never raises into an emitter. Only four emitter families use it in PR 1 (headroom, skills adoption, system agent, circuit breaker); the rest are ratcheted in `PREFIXES_NOT_YET_ON_SEAM` until PR 2. - `operator_queue_service.py` - Operating Room sync with agent containers (OPS-001); the agent-authored ingestion boundary enforces the per-agent depth/rate/size caps + reserved-id guard + leader lock (#1632) - `operator_queue_choices.py` - The one rule for what an approval's `response` may be (#2376, #3242): an offered option (exact match, else `ResponseNotOfferedError` with the list) or the reserved `SOMETHING_ELSE = "(something else)"` with an instruction in `response_text` (`InstructionRequiredError` / `ReservedValueError` / `NotOffMenuError`, all `ReservedAnswerError`); import-free leaf the ask sink calls before any write; also holds `OPTIONS_DROPPED_MARKER`. Mirrored in `utils/operatorQueue.js` + MCP `types.ts` (parity-tested) - `sync_health_service.py` - Git sync health polling — see [Git Sync Health](agent-lifecycle.md#git-sync-health-389390) @@ -193,7 +194,7 @@ - `runtime_secret_scrub.py` - Runtime secret-scrub seam (ent#279): identity-based redaction of a producer-**STAGED** value out of persisted output, catching arbitrary values the pattern-based `utils/credential_sanitizer` (prefix regexes + `KEY=value`) cannot. A generic OSS mechanism — mechanism-only by design, names no consumer. Sync API: `stage_secret(agent_name, value)` (fail **CLOSED** — any failure, Redis down or hard cap, raises `StagingUnavailable` so a producer that cannot stage refuses delivery), `get_staged_values()`, and `scrub_text`/`scrub_obj` for the execution-terminal persistence chokepoints to redact staged values (marker `***REDACTED***`; raw / once-JSON-escaped / base64 renditions; longest-first; falsy-passthrough) **before** the write. Store = ONE global Redis HASH `secret_scrub:staged` (field `sha256(value)` → AES-256-GCM envelope via the credential-encryption singleton) + a `secret_scrub:staged_at` ZSET giving 24h **per-member** expiry (Redis holds only envelopes; values ≥8 chars only). The scrub side fails **OPEN** (a Redis/decrypt error → no identity scrub + one throttled ERROR; the pattern pass still runs and the terminal still persists — no NEW secret is delivered during the outage because the stage side is closed). Behaviour-neutral for OSS (no staged values ⇒ every scrub is a no-op) — see [runtime-secret-scrub.md](../feature-flows/runtime-secret-scrub.md) - `subscription_service.py` - Subscription management (SUB-002); owns `derive_auth_mode` — the ONE auth-mode enum derivation shared by `AgentAuthStatus` and the #471 pressure batch endpoint. ent#582: `is_claude_auth_configured` (the `claude_auth_configured` flag's one definition) and `connect_agents_to_first_credential` — when a write gives the install its FIRST Claude credential, every agent created without one gets the new subscription assigned, and agents with a running container are recreated in the background via the SUB-003 `_restart_agent` under the #799 switch lock. Candidates come from the DB rows (`db.list_agents_awaiting_first_credential`: not ephemeral, no subscription, `use_platform_api_key` on, never a successful execution) filtered to Claude runtime, never from `list_all_agents_fast` (which reads `[]` on a Docker fault). The restart re-checks each agent under the lock and skips one with a running execution (`db.agent_has_running_execution`) or an already-current env, so the call is idempotent; returns an int count (`connected_agents` on both save responses). `system_seed_service._connect_seeded_agents` re-runs it after a seed pass that created agents (a create can straddle the save). Scope is deliberately "could not authenticate anyway": an established fleet, or an agent on its own `.env` key, is never touched. **#2572** adds the credential-less predicate and the adoption sweep: `instance_has_api_key` (which MUST stay `get_anthropic_api_key()` — env fallback included — never `has_secret_setting()`, a DB-only presence check that would adopt a fleet off a working env key), `credentialless_agent_names` (one `get_agent_subscription_map` read, `[]` whenever the instance has a key) and `adopt_for_credentialless_agents` (awaited decide+persist, backgrounded container apply). Its runtime gate is LABEL-STRICT via `docker_service.agent_container_runtime_labels` and fails CLOSED on an unreadable Docker — the opposite of ent#403's fail-open, because a UI affordance and a persisted credential assignment are not the same bet - `subscription_headroom_service.py` - Live subscription headroom (#471): actual 5h/7d utilization % + reset times read from the `anthropic-ratelimit-unified-*` headers of a `max_tokens=1` micro-ping probe (the `/api/oauth/usage` endpoint is 403-scope-dead for stored `sk-ant-oat01-` setup tokens — the closed-PR #2170 mechanism, established 2026-08-19). Click-to-refresh (floored ≥60s/subscription) + default-ON ambient refresh (`subscription_headroom_auto_refresh` setting, 15-min floor, demand-driven). **Fail-CLOSED without Redis** — `_read_snapshot` is tri-state (`(redis_ok, snapshot)`; a client object existing ≠ a reachable server), so a Redis outage yields observed-only, never a probe storm; the dashboard batch path uses `wait=False` (stale-while-revalidate via strong-ref background task) so a hung provider can't wedge the 60s poll. Snapshot: Redis `subscription:headroom:{id}` (7d TTL); probes single-flighted (#1920); a probe 429 updates snapshot status only, never `subscription_rate_limit_events`. `services/subscription_headroom_service.resolve_rate_limited_now` is the ONE `rate_limited_now` derivation (#2157 one-gate rule), consumed by both `decorate_usage` and `pressure_states`. **It is three-state, not an OR (#447):** fresh provider verdict says limited → limited; fresh verdict says allowed → NOT limited; no usable verdict → the 2h event predicate. It shipped as `db_predicate OR fresh_verdict`, which made a fresh, authoritative *"allowed, 32% used, resets 19:10"* structurally powerless to clear the badge — because **nothing clears a failure row on success**: `clear_rate_limit_events` has had zero production callers since #444 removed the one call (clearing was destroying auto-switch's detection signal), so the event half only ever decays with the clock. Observed live: two subscriptions wearing `LIMIT` while every agent on them answered normally. The db predicate is an *inference from past failures*; a probe is *ground truth about now*, so it wins in both directions. `_headroom_indicates_healthy` is deliberately **not** the negation of `_headroom_indicates_limited` — "not limited" is also true for a stale snapshot, a rejected token and a transport error, none of which are evidence of headroom, so all three fall through to the predicate rather than clearing it (#2353's rule, preserved). **Both predicates judge a window against ONE named allowlist, `NON_BLOCKING_WINDOW_STATUSES` (#2396)** — an ALLOWLIST of statuses meaning "requests are being served", never a blocklist of blockers, so an unrecognised status still reads as limited (the inverse mistake is the #848 deny-check lesson). It shipped as the bare literal `("allowed",)` in one predicate and `(None, "allowed")` in the other, which made the provider's own near-the-limit tier `allowed_warning` read as a hard rate limit on every surface — a healthy subscription approaching its weekly window wore `LIMIT` while the provider was still serving it (observed live: a snapshot recording `seven_day: 90%, allowed_warning` beside 47 successful executions and zero `subscription_rate_limit_events`). `allowed_warning` now also counts as **positive proof of headroom** — the quota was reached and the answer was yes — so it clears a stale 2h db predicate; excluding it would be #447 returning in a narrower window. Note the ordering that bounds the blast radius: a real HTTP 429 sets the top-level snapshot `status` and is checked BEFORE any window, and the db predicate is a third independent detector, so the window arm is the weakest of the three. This is the DISPLAY predicate only: auto-switch candidate filtering reads the kind-blind `has_recent_subscription_failures`, so a just-recovered subscription is still skipped as a switch target and #444's ping-pong cannot return. **The event half of that predicate counts `failure_kind = 'rate_limit'` ONLY (#2352)** — it was kind-blind, so an auth failure (401/403: a dead, expired, or `.env`-shadowed token) set the flag and every surface reported a credential problem as quota exhaustion, sending the operator to wait out a window that was never full; the display layer had honoured the split since #471 (`rateLimitEventCount` reads the `rate_limit` kind alone) and the predicate was the layer that had not caught up. NULL `failure_kind` (pre-#471) is excluded — unknown is never promoted to "429". This is a **split, not a filter**: `db.has_recent_subscription_failures` preserves the kind-BLIND semantics for the two *candidate-selection* listings (`list_viable_alternative_subscriptions`, `list_assignable_subscriptions` — since #2409 the db only FILTERS; the services rank, see `subscription_auto_switch.py`), which must keep skipping a subscription that failed for ANY reason — narrowing them in place would have auto-switched agents onto subscriptions whose token had just been rejected (the #444 class). Two predicates, two meanings; re-merging them is how the bug happened. `pressure_states` therefore also emits `auth_failures_24h` beside the total, since after the split a dead-token subscription is no longer `rate_limited_now` and a bare total cannot tell the badge which word to use. The observed (DB-derived) arm is always populated; SUB-004's windows are **deduped since #471** (executions = sole cost/context/turn source; `chat_messages` contributes `output_tokens` only — a `/chat`/persisted-`/task` turn writes cost into BOTH tables). **ent#433 adds the durable half** — `_probe_and_store` now also writes each probe to `subscription_headroom_history`, so utilization TRENDS survive a snapshot that overwrites itself every probe. Three properties are load-bearing: the history write runs **after** `_store_snapshot` and **off the event loop** (`asyncio.to_thread`; a sync SQLAlchemy write on the loop stalls everything for up to the 30s busy timeout when it lands during the 03:30 backup or 04:30 VACUUM, and `try/except` handles errors but not blocking) — order pinned by test; it catches `Exception`, never `BaseException`, so shutdown's `CancelledError` still propagates; and it adds **no probe**, inheriting #471's entire rate-bounding envelope (60s floor, single-flight, fail-closed ambient) by sitting inside `_probe_and_store`. `get_history(subscription_id, window)` is the read — pure DB, never probes, so viewing a trend costs no quota. **ent#434 adds a THIRD predicate beside the two binary ones — `classify_headroom` → `saturated | has_headroom | unassessable`** — because the alert path needs a state the binary pair cannot express. `_headroom_indicates_healthy` returns `False` for a stale snapshot, a rejected token, a transport error AND a genuinely saturated subscription, so it cannot tell *no evidence* from *no headroom*; #2396's docstring named ent#434 as its consumer and was wrong, since the fleet escalation would then be blocked by exactly the condition it reports. The docstring is corrected and the BODY left untouched (`resolve_rate_limited_now` consumes it — a behavioural edit moves every `LIMIT` badge). The classifier keys on **utilization, not window status**: `allowed_warning` is deliberately non-blocking, so a status-driven classifier would file a live 90% warning-tier reading as `has_headroom` — the exact reading the alert exists for. Status remains the *stronger* signal (a blocking 7d status or a probe 429 is `saturated` with or without a number, counting toward the fleet claim but raising no percentage-crossing alert). Every could-not-tell path returns `unassessable`, including the easily-missed 5h-only snapshot (`parse_unified_headers`'s top guard passes on `5h-utilization` alone, so a gate reading "ok status + non-blocking windows" would claim WEEKLY headroom on zero weekly evidence). `ensure_reading(sid, max_age)` is the ONE probe decision per subscription per cycle — two consumers each deciding for themselves means the second is floored out by `MIN_PROBE_INTERVAL_SECONDS` and reads "no data" on precisely the subscriptions the first just refreshed. `SAMPLE_INTERVAL_SECONDS` (3600, floored at `REFRESH_SECONDS`) is a **constant, not a knob** (#1644) — and literally so: it reads no env var, because neither compose uses `env_file` and an unforwarded read would be inert while still reading as configurable. The one real lever is `SUBSCRIPTION_SWEEP_CONCURRENCY` (probes per sweep chunk), forwarded in both composes **#2409 adds the auto-switch ranker beside the classifier, over ONE gate.** `headroom_reading(headroom, max_age_seconds)` is the usability gate both consume — fresh? probe answered (`ok`/`rate_limited`/`invalid_token`)? window shape? — and `classify_headroom` became six lines of policy over it with byte-identical verdicts (pinned by a differential test against a frozen copy of the pre-#2409 function over the full age × status × window × threshold product; the ent#434 suite is unchanged). `cached_headroom_readings(ids)` is the selector's read: ONE `MGET` over `subscription:headroom:{id}`, tri-state on Redis (client unbuildable or a raise ⇒ every candidate unknown in one attempt; a malformed snapshot blinds only itself), and it never calls `get_headroom`/`_locked_probe`. `rank_subscriptions` / `selection_verdict` / `selection_sort_key` are **threshold-free** — the alert threshold is an operator knob that must not steer where agents land, and it is `0` when alerts are off: tier `measured` (fresh, serving, weekly figure) sorted by the FULLER window (`primary` = nearest wall) in `HEADROOM_BAND_PCT`=10 bands with `agent_count` as the in-band tiebreak, then the other window, then name; `unknown` (no/stale/error/5h-only/non-finite, or a STALE refusal) = exactly today's `agent_count ASC, name ASC`; `refused` (a FRESH probe 429 / blocking window / rejected token) is dropped. Two bounds: `MAX_READING_AGE_SECONDS` (2h — now OWNED here, `= max(2h, 2 × SAMPLE_INTERVAL_SECONDS)` so it always covers two sampler intervals, re-exported by `subscription_headroom_alerts`) bounds the weekly figure, which is a lower bound within the provider's fixed window; `FRESHNESS_SECONDS` (30 min, the LIMIT-badge bound) bounds the 5h figure (it can fully reset inside 2h) and any refusal (a point-in-time verdict). `describe_reading` is the notification/activity shape — tiers and figures, never a token. ent#582: `check_token(token)` sends that same one-message probe (`_post_probe`) for a raw token BEFORE it is registered — `ok`/`rate_limited`/`invalid_token`/`error`, backing `POST /api/subscriptions/test`. -- `subscription_headroom_alerts.py` - The weekly-window alert (ent#434): pure projection + tier decision + deterministic emitter, no durable state. **The alert id IS the state machine** — the 7d window was measured fixed-with-reset (live history: `seven_day_resets_at` constant across five days of probes, then a +7d step), so utilization is monotonic within a window, a hysteresis floor is dead code, and `sub-headroom-{sid}-{reset-day}-{tier}` gives edge-trigger, cross-worker dedup and re-arm for free via `create_item`'s `UNIQUE(agent_name, request_id)` ON CONFLICT DO NOTHING (the id is quantised to the DAY as a belt: if some plan did behave as rolling it degrades to one alert/day, not one/probe). **The threshold fires; the projection ranks** — `projected_end = utilization_pct / fraction_of_window_elapsed`, so urgency is the operator's own burn rate rather than a configured time gap that is wrong for everyone; the alert is never withheld at the threshold, `priority` is `low` under 100% projected and `high` over. Fleet escalation needs ≥2 subscriptions (with one, "this is full" and "every one is full" are the same fact) and every member assessable — one `unassessable` blocks the claim and is named (ent#100). **The denominator is built from the subscription roster, never from the sweep's results**: a member whose sampling raised carries `classification: None` by design (the swallow protects #447) and a mid-cycle lease yield `break`s with a short result list, so a truthiness filter dropped exactly the members that should have blocked the claim — a 3-subscription instance emitted "All 2 registered subscriptions are at or past 75%" with the per-subscription alerts suppressed by the early return. `fleet_verdict` was always correct; the caller had narrowed its input. It states what was MEASURED and deliberately does NOT say "auto-switch has nowhere to go": since #2409 the selector ranks over the SAME cached readings this sweep samples, but its candidate set is also narrowed by the 2h failure filter the sweep never sees, so the body describes the measurement and never promises what the selector will do. Platform-only emitter on sentinel `_sub-headroom` (`_PLATFORM_ALARM_SENTINELS` + `_RESERVED_ID_PREFIXES` + the `test_1677` allowlist; `expires_at` None). Residual: `create_item` has no UPDATE path, so a 75% row still reads 75% at 92% — the escalation is a separate, self-contained item +- `subscription_headroom_alerts.py` - The weekly-window alert (ent#434): pure projection + tier decision + deterministic emitter, no durable state. *Since #3246 it reports through `platform_alerts.observe` (one row per subscription, updated in place, `reconcile` ends cleared ones, 14-day lifetime) — the id-as-state-machine below is history.* **The alert id WAS the state machine** — the 7d window was measured fixed-with-reset (live history: `seven_day_resets_at` constant across five days of probes, then a +7d step), so utilization is monotonic within a window, a hysteresis floor is dead code, and `sub-headroom-{sid}-{reset-day}-{tier}` gives edge-trigger, cross-worker dedup and re-arm for free via `create_item`'s `UNIQUE(agent_name, request_id)` ON CONFLICT DO NOTHING (the id is quantised to the DAY as a belt: if some plan did behave as rolling it degrades to one alert/day, not one/probe). **The threshold fires; the projection ranks** — `projected_end = utilization_pct / fraction_of_window_elapsed`, so urgency is the operator's own burn rate rather than a configured time gap that is wrong for everyone; the alert is never withheld at the threshold, `priority` is `low` under 100% projected and `high` over. Fleet escalation needs ≥2 subscriptions (with one, "this is full" and "every one is full" are the same fact) and every member assessable — one `unassessable` blocks the claim and is named (ent#100). **The denominator is built from the subscription roster, never from the sweep's results**: a member whose sampling raised carries `classification: None` by design (the swallow protects #447) and a mid-cycle lease yield `break`s with a short result list, so a truthiness filter dropped exactly the members that should have blocked the claim — a 3-subscription instance emitted "All 2 registered subscriptions are at or past 75%" with the per-subscription alerts suppressed by the early return. `fleet_verdict` was always correct; the caller had narrowed its input. It states what was MEASURED and deliberately does NOT say "auto-switch has nowhere to go": since #2409 the selector ranks over the SAME cached readings this sweep samples, but its candidate set is also narrowed by the 2h failure filter the sweep never sees, so the body describes the measurement and never promises what the selector will do. Platform-only emitter on sentinel `_sub-headroom` (`_PLATFORM_ALARM_SENTINELS` + `_RESERVED_ID_PREFIXES` + the `test_1677` allowlist; `expires_at` None). Residual: `create_item` has no UPDATE path, so a 75% row still reads 75% at 92% — the escalation is a separate, self-contained item - `subscription_auto_switch.py` - SUB-003 auto-switch orchestration (`handle_subscription_failure` → `_perform_auto_switch`, hot-reload #1089). **Since #2409 it owns alternative SELECTION** — it used to be a first-match pick inside the db layer (`ORDER BY agent_count ASC`, first survivor of the 2h skip-list), which read no headroom, moved agents onto subscriptions at 99% of their weekly window, and sorted an *unused dead-token* subscription FIRST (no agents ⇒ no failure rows). `select_best_alternative_subscription(current)` = filter → rank → first: the db lists the survivors (`list_viable_alternative_subscriptions`, kind-blind #444/#2352, `agent_count ASC, name ASC` — filter only, the name tiebreak makes the fallback order deterministic), the headroom service ranks them (`rank_subscriptions` over `cached_headroom_readings`, one `MGET`, never a probe — a candidate the filter dropped is never even read), and the pick returns with a `why`. Runs under the per-agent switch lock via `asyncio.to_thread` (both reads are blocking). **Fail-open on the ranking half only, and LOUDLY**: Redis down, or a lazy import that resolved to the wrong module (`importlib.import_module` answers from `sys.modules`, not from a package attribute a previous test left behind — learnings 2026-08-12), degrades to the db's order with a WARNING; an all-unknown set with ambient refresh OFF logs that the ranker is inert. The only case that now yields no target where it previously did is *every survivor is currently refused by the provider* (an approved deviation from the issue's literal AC #3, recorded on it). `_perform_auto_switch(..., destination_headroom=why)` surfaces the pick on the activity `details`, the notification `metadata` and the result, plus one clause in the notification text (on a two-subscription install the ranking cannot change the pick, so the explanation IS the value). `subscription_service.select_subscription_for_new_agent()` (#74 new-agent auto-assign) rides the same ranker — `db.list_assignable_subscriptions()` → rank → first candidate whose token decrypts (#340), one decrypt in the common case; `database` resolved at call time for the creation harnesses. `get_least_used_subscription` / the db `select_best_alternative_subscription` are gone. See [subscription-auto-switch.md](../feature-flows/subscription-auto-switch.md) **#2638 adds three things the reactive switch could not do.** (1) The 2h skip-list is OVERRIDABLE per candidate: `_readmit_recovered` reads the filter's complement (`db.list_recently_failed_alternatives`) and readmits one only on positive evidence — `subscription_headroom_service.recovery_verdict` = a FRESH not-refusing reading (`serving_now`), or a blocked window's own reset instant ELAPSED **and predating** the failure (`window_reset`; without that ordering a subscription that 429'd after its rollover is readmitted on a reset it already consumed). Absence of evidence readmits nothing, so #444's ping-pong — caused by FORGETTING a failure — stays closed, and the fail-open ranking branch readmits nobody by construction because it is precisely where the evidence could not be read. A `window_reset` candidate is handed to the ranker as UNKNOWN: its `blocked` flag describes the window that just rolled over, and `rank_subscriptions` would drop it as `refused` — the readmission would be inert in exactly the case it exists for. (2) `ensure_serviceable_subscription` switches BEFORE the first dispatch when the assigned subscription is already known-refused (fresh provider refusal, or a 429 in the 2h window via the 429-only DISPLAY predicate — an auth failure is a credential problem another subscription may share). It never raises, records NO failure event (nothing failed; a synthetic one would poison the skip-list it feeds), dispatches anyway when there is no alternative, and performs the SAME `_perform_auto_switch` so one activity/notification/hot-reload happens whichever path fired — `pre_dispatch=True` changes only the wording. (3) `fallback_to_api_key` is the last resort when no subscription can serve: clear the assignment, set `use_platform_api_key`, RESTART (the reload endpoint pushes an OAuth token; this needs the opposite change, which `lifecycle`'s auth block already derives from DB state). Setting `subscription_api_key_fallback` (default ON, fail-OPEN on a read error) with `key_configured` on the GET, because a toggle reading only "on" with no key stored describes a remedy that cannot run. `earliest_known_reset` feeds the portal's honest "quota resets at …" refusal. **#2572 adds a FOURTH caller of `agent_switch_lock` / `_restart_agent`** — `subscription_service`'s credential-less adoption sweep — and changes nothing in this module: SUB-003 stays failure-driven and its precondition 2 (*the agent already has a subscription*) is deliberately NOT relaxed, because that precondition is where the #2572 gap was written down, not a defect. The sweep also deliberately does not become a fourth `_hot_reload_subscription_token` producer: credential-less → subscription is an auth-MODE change, and that helper's docstring invariant is that all its producers are sub→sub by construction. - `ssh_service.py` - Ephemeral SSH credential generation - `email_service.py` - Email sending for verification codes. Provider, Resend key and sender resolve per send through `settings_service` (ent#582), not the import-frozen config values diff --git a/docs/memory/architecture/database.md b/docs/memory/architecture/database.md index 50c39bed79..b6e860f796 100644 --- a/docs/memory/architecture/database.md +++ b/docs/memory/architecture/database.md @@ -751,7 +751,7 @@ CREATE TABLE operator_queue ( -- trinity-enterprise#611 (SQLite `operator_queue_ask_object` / Alembic `0076_operator_queue_ask_object`; nullable, no backfill): disposition TEXT, -- answered|cancelled|expired — written in the SAME CAS UPDATE that flips `status`; NULL = ended before the ledger disposed_at TEXT, - disposed_by TEXT, -- person|timeout (an enum of two: only a person or the clock ends an ask) + disposed_by TEXT, -- person|timeout|platform — only a person or the clock ends an ASK; `platform` ends only a platform alert (#3130 superseded flood alarms, #3246 condition_cleared) disposed_by_email TEXT, -- NULL for timeout; withheld from machine keys on get/list disposition_reason TEXT, -- the operator's optional cancel reason (≤ 500); never in an audit row, never to a Workspace client batch_id TEXT, -- one uuid per bulk-cancel sweep; its re-select is the sweep's CAS winners @@ -761,6 +761,8 @@ CREATE TABLE operator_queue ( resolved_to TEXT, -- PR B: JSON list of person refs; withheld from machine keys proposal TEXT, -- PR B: JSON, the frozen action supersedes_expired TEXT, -- PR B: the agent's own expired predecessor (row uuid) + subject TEXT, -- #3246: a platform alert's condition key `:`; at most one PENDING row per (agent_name, subject) — partial unique index uq_operator_queue_pending_subject + last_seen_at TEXT, -- #3246: the latest reading of that condition (the count rides in context.seen_count) FOREIGN KEY (responded_by_id) REFERENCES users(id) ); CREATE INDEX idx_operator_queue_agent ON operator_queue(agent_name); diff --git a/docs/memory/feature-flows/internal-system-agent.md b/docs/memory/feature-flows/internal-system-agent.md index 9be2befccd..50a5095733 100644 --- a/docs/memory/feature-flows/internal-system-agent.md +++ b/docs/memory/feature-flows/internal-system-agent.md @@ -129,7 +129,7 @@ would assert only that a correct container is correct. | Boundary | Behaviour | |---|---| -| Backend boot, container **running** | **Read-only.** `check_base_image_state()` → `result["base_image_state"]` ∈ `stale \| current \| unknown`; `action: "none"`. WARNING names the remedy; an edge-triggered `base-image-stale-` operator-queue alarm fires on `stale` **only** (a fail-open probe must never manufacture an alert, so `unknown` never alarms). **No recreate call exists in this branch** — source-pinned. The alarm's cooldown cursor is **per-process**, and `ensure_deployed` runs once per worker's lifespan, so a stale boot files one item **per worker** (`--workers 2` ⇒ 2). Deliberate: a cross-worker cursor would put Redis or a DB read on the boot path for an advisory alarm, and the un-guessable timestamped id — which is what stops an agent pre-creating and silencing it — cannot be deduped by `on_conflict_do_nothing` anyway. | +| Backend boot, container **running** | **Read-only.** `check_base_image_state()` → `result["base_image_state"]` ∈ `stale \| current \| unknown`; `action: "none"`. WARNING names the remedy; an edge-triggered `base-image-stale-` operator-queue alarm fires on `stale` **only** (a fail-open probe must never manufacture an alert, so `unknown` never alarms). **No recreate call exists in this branch** — source-pinned. The alarm's cooldown cursor is **per-process**, and `ensure_deployed` runs once per worker's lifespan, so a stale boot files one item **per worker** (`--workers 2` ⇒ 2). *(Since #3246 the alarm goes through `platform_alerts.observe` under the subject `base_image_stale:trinity-system`: every worker's reading lands on the SAME pending row (seen count + 1), the platform ends it when the image is current again, and it expires after a 30-day net — no longer one item per worker that stays until acknowledged.)* Deliberate: a cross-worker cursor would put Redis or a DB read on the boot path for an advisory alarm, and the un-guessable timestamped id — which is what stops an agent pre-creating and silencing it — cannot be deduped by `on_conflict_do_nothing` anyway. | | Backend boot, container **stopped** | Delegates to `start_agent_internal(SYSTEM_AGENT_NAME)` instead of a bare `container_start`, inheriting the #1809 cold-start image gate, #1560 clear-before-recreate ordering, the 409/NotFound concurrent-recreate hardening and the post-recreate handle re-lookup. `recreated` / `recreate_reason` are surfaced into the result. **This is where adoption happens.** | | `POST /api/system-agent/restart` | Explicit stop, then `start_agent_internal` — same adoption path. The response re-fetches the container so it reports the replacement's status, not the removed handle's. | diff --git a/docs/memory/feature-flows/operating-room.md b/docs/memory/feature-flows/operating-room.md index 0f7a7a6ea0..d384fdacab 100644 --- a/docs/memory/feature-flows/operating-room.md +++ b/docs/memory/feature-flows/operating-room.md @@ -501,11 +501,11 @@ Writers are edge-triggered (`WHERE sync_state IS NULL OR != :v …`; the rowcoun **Why.** An ask ends in exactly one of three ways — answered, cancelled, expired — and before #611 five write sites ended one, each with its own subset of side effects: single cancel recorded no actor, audited nothing, broadcast nothing and woke nobody; respond audited nothing; expiry never broadcast; bulk-cancel audited the ids it was ASKED to cancel. No surface could say who ended an ask or when (ResolvedCard showed `created_at` as the ending time), and an agent with no next turn waited forever on a cancelled ask. And an agent-scoped key resolved to its owner, so an agent could answer or cancel any ask its owner could reach — its own approval included — recorded as the owner. -**The ledger.** Written in the SAME compare-and-set UPDATE that flips `status`, so a writer that loses the race records nothing: `disposition` (`answered | cancelled | dismissed | expired` — `dismissed`, trinity-enterprise#748, rides `status='cancelled'`), `disposed_at`, `disposed_by` (`person | timeout` — an enum of two), `disposed_by_email` (NULL for timeout), `disposition_reason` (the operator's optional cancel reason, ≤ 500), `batch_id` (one uuid per bulk sweep). Nullable, no backfill — a row that ended before the ledger reads from `status`, and no surface presents its `created_at` as an ending time. The same migration (`operator_queue_ask_object` / Alembic `0076_operator_queue_ask_object`) adds `raised_by`, `channel`, `to_role`, `resolved_to`, `proposal`, `supersedes_expired` for PR B (agent-raised asks); the file poller already stamps `channel='file', raised_by='agent'` as keyword-only arguments — never read from the agent's entry. `expires_at` is stored ISO-Z at the sink (Invariant #16; an unparseable value is kept verbatim so the #2915 fingerprint never reads it as a rewrite). +**The ledger.** Written in the SAME compare-and-set UPDATE that flips `status`, so a writer that loses the race records nothing: `disposition` (`answered | cancelled | dismissed | expired` — `dismissed`, trinity-enterprise#748, rides `status='cancelled'`), `disposed_at`, `disposed_by` (`person | timeout`, plus `platform` for a platform alert the platform itself ended — #3246), `disposed_by_email` (NULL for timeout), `disposition_reason` (the operator's optional cancel reason, ≤ 500), `batch_id` (one uuid per bulk sweep). Nullable, no backfill — a row that ended before the ledger reads from `status`, and no surface presents its `created_at` as an ending time. The same migration (`operator_queue_ask_object` / Alembic `0076_operator_queue_ask_object`) adds `raised_by`, `channel`, `to_role`, `resolved_to`, `proposal`, `supersedes_expired` for PR B (agent-raised asks); the file poller already stamps `channel='file', raised_by='agent'` as keyword-only arguments — never read from the agent's entry. `expires_at` is stored ISO-Z at the sink (Invariant #16; an unparseable value is kept verbatim so the #2915 fingerprint never reads it as a rewrite). **The writers** (`db/operator_queue.py`): `respond_to_item` (CAS adds `expires_at IS NULL OR expires_at > now` — a late answer returns `_status_conflict` with status still `pending`); `cancel_item(item_id, *, disposed_by_email, reason, disposition='cancelled')` (CAS; a lost race returns `_status_conflict`; `disposition='dismissed'` for the addressee's Dismiss, anything else is a `ValueError`); `bulk_cancel_items(…, *, disposed_by_email, reason) → {batch_id, rows}` (one UPDATE stamps the sweep's own `batch_id`, the re-select `id IN (:ids) AND batch_id = :b` is exactly the winners — dialect-agnostic, no RETURNING); `mark_expired() → rows` (bounded candidate select, then per-id CAS — overlapping leaders end each row once). -**The sink** (`services/ask_service.py`): `answer(item, …)` (runs the #2376 offered-option check first — with the #3242 reserved `(something else)` branch and the `not_off_menu` refusal on a platform-minted approval — the one writer of an answer), `cancel`, `bulk_cancel`, `expire`. Each: CAS writer → one audit row per transition (`answered` · `cancelled` (+`has_reason`) · `bulk_cancel` (+`batch_id`, the cancelled ids) · `expired` (`source=system`); ids and enums only) → ONE thin trigger (`operator_queue_responded` / `operator_queue_cancelled` agent-keyed; `operator_queue_cleared` count-only for a sweep; expiry sends none — the poll cycle's one `operator_queue_sync` covers it, its reset now sits above expiry) → the ending observers (`register_ending_observer`), handed only the CAS-won rows. Synchronous: the portal's answer route is a plain `def` on a worker thread, so the audit and broadcast hop to the loop through `operator_resume_service.spawn_on_loop` and are never awaited. `_broadcast_payload` names each trigger so the ent#467 `/ws` guard reads it. +**The sink** (`services/ask_service.py`): `answer(item, …)` (runs the #2376 offered-option check first — with the #3242 reserved `(something else)` branch and the `not_off_menu` refusal on a platform-minted approval — the one writer of an answer), `cancel`, `bulk_cancel`, `expire`. Each: CAS writer → one audit row per transition (`answered` · `cancelled` (+`has_reason`) · `bulk_cancel` (+`batch_id`, the cancelled ids) · `expired` (`source=system`); ids and enums only) → ONE thin trigger (`operator_queue_responded` / `operator_queue_cancelled` agent-keyed; `operator_queue_cleared` count-only for a sweep; expiry sends none — the poll cycle's one `operator_queue_sync` covers it, its reset now sits above expiry) → the ending observers (`register_ending_observer`), handed only the CAS-won rows. Synchronous: the portal's answer route is a plain `def` on a worker thread, so the audit and broadcast hop to the loop through `operator_resume_service.spawn_on_loop` and are never awaited. That hop works from EVERY thread origin (#3246): the loop thread, an anyio worker (Starlette's threadpool), and — since the headroom sweep and the skills reconcile end platform alerts through `asyncio.to_thread`, the loop's default executor, which anyio does not own — any other thread while the host loop runs, via `call_soon_threadsafe` on the loop `main.py::lifespan` captures first (`remember_host_loop`) and every on-loop spawn refreshes. With no loop to hop to it still raises, never a silent no-op (pinned by `test_3246_platform_alerts_thread_origin.py`). `_broadcast_payload` names each trigger so the ent#467 `/ws` guard reads it. **Only a person ends an ask.** `dependencies.reject_non_person_principal` (an allowlist over `mcp_scope ∈ {None, "user"}`, no `agent_name`/`connector_agent`/`portal_delegate`; a principal with no scope attribute fails closed) guards respond, cancel and bulk-cancel before anything is read → 403 `person_required`. The Workspace answer (`client_portal/asks/router.py`) holds its platform principals to the same rule: `get_portal_principal` sets `PortalPrincipal.is_person` from `is_person_principal`, so a system-scoped key keeps its Workspace read breadth (#2198) but its answer is refused with the same 403 before the row is read. @@ -746,9 +746,17 @@ The lease reaper (`services/lease_reaper_service.py`) creates operator-queue ite Agent creates -> pending -> responded (by a person) -> acknowledged (by agent) -> cancelled (by a person — single cancel or bulk Clear All) -> expired (by the platform, once expires_at passed) + -> cancelled by the PLATFORM (platform alerts only, #3246: + condition_cleared, or superseded by the upgrade sweep) ``` -Since trinity-enterprise#611 the same UPDATE writes the endings ledger: `disposition` answered | cancelled | expired, `disposed_by` person | timeout, `disposed_at`, and for a cancel the optional `disposition_reason` (+ `batch_id` for a sweep). `acknowledged` is a delivery fact, not an ending — its row keeps `disposition = answered`. +### Platform alerts — one row per subject (#3246) + +A platform alert is a **condition**, not a message (requirements §26.14). `services/platform_alerts.py::observe(agent, kind, key, …)` files at most one pending row per subject `:` — the partial unique index `uq_operator_queue_pending_subject` enforces it — and a newer reading updates that row in place: `last_seen_at`, `context.seen_count` + 1, `expires_at` moved to the last reading + the kind's lifetime (14 days, `OPERATOR_PLATFORM_ALERT_LIFETIME_DAYS`; a fixed 30-day net for edge-triggered kinds). It no longer stays pending until someone acknowledges it: `clear` / `reconcile` end it `cancelled` with `disposed_by = 'platform'`, `disposition_reason = 'condition_cleared'`, and otherwise it expires. After a person ends one, the same reading files nothing for 7 days (`OPERATOR_PLATFORM_ALERT_SNOOZE_DAYS`) unless the priority rises or a material key changes. The surfaces read a platform ending as "Ended by the platform — the condition cleared" and show "seen N times · last seen " on a pending alert seen more than once (QueueCard, QueueItemDetail, `/m`). + +**Only four emitter families are on the seam in this PR** — subscription headroom (`_sub-headroom`), legacy skills-library adoption (`_skills-sync`), the system agent's stale base image / start failure, the agent circuit breaker's dormant alarm. Every other platform emitter still files one row per minted id and is listed in `PREFIXES_NOT_YET_ON_SEAM`; they follow in PR 2. The upgrade sweep already collapses the backlog for every kind whose subject it can derive. + +Since trinity-enterprise#611 the same UPDATE writes the endings ledger: `disposition` answered | cancelled | expired, `disposed_by` person | timeout | platform (platform alerts only, #3246), `disposed_at`, and for a cancel the optional `disposition_reason` (+ `batch_id` for a sweep). `acknowledged` is a delivery fact, not an ending — its row keeps `disposition = answered`. Cancellations and expirations are propagated back to the agent on the next sync cycle (#1017): still-`pending` entries in the agent's file are flipped in place to their terminal DB status (`cancelled` or `expired`) so the agent stops waiting on them (entries missing from the file are never appended — there is nothing to deliver). @@ -1080,6 +1088,7 @@ Clear All on **Resolved** instead calls `clearResolved()` -> `POST /api/operator | Date | Change | |------|--------| +| 2026-10-05 | #3246 — platform alerts are conditions: one pending row per subject (updated in place, seen count, last seen), ended by the platform when the condition clears, a 14-day lifetime and a 7-day snooze. Four emitter families on the seam (headroom, skills adoption, system agent, circuit breaker); the rest follow in PR 2. Ending labels and the seen line on the queue surfaces. | | 2026-10-05 | #3243 — atomic asks: hard caps on agent asks (≤5 options excluding `(something else)`, ≤60 characters each, chip lookalikes refused, agent titles ≤120), refused by name on the native raise after the replay (`too_many_options`, `option_too_long`, `invalid_options`, `title_too_long`) and held on the file path with the offending ids on `platform.ingestion` (no flood alert); five authoring rules in the `ask_operator` description, three prompt lines pointing at it; no schema change | | 2026-10-05 | #3242 — approvals get one sanctioned off-menu answer: the reserved `(something else)` decision with the person's instruction in `response_text`, accepted by the sink on any approval and refused by name elsewhere (`instruction_required`, `reserved_value`, `not_off_menu`, `invalid_options` at raise); a Something else chip on the desktop card and detail, `/m` and the Workspace, hidden on gate approvals on every surface (`decided_by_options` on the Workspace projection); resolved views read "Something else"; the resume frame and the contract text name the value; no schema change | | 2026-09-29 | trinity-enterprise#715 — the reads stop handing a person's email to a machine key: the get/list routes give non-person principals the `_MACHINE_ROW_FIELDS` allowlist (no `responded_by_*`, `addressed_to_email` on any row, `disposed_by_email`, `resolved_to`); the platform's heads-ups about a person (ent#499 problem report, ent#308 inbox collision) are skipped for machines (404 on the item route) and `portal-inbox-collision-` is a reserved prefix, so the collision alert never reaches the agent's file or its resume turn; the file write-back no longer writes `responded_by` (both paths; the prompt now says `response`, `response_text` and `responded_at`); the clear-resolved `/ws` trigger is the count only. Tests: `tests/unit/test_ent715_queue_person_fields.py`. | diff --git a/docs/memory/learnings/2026-10-05-an-alert-is-a-condition-not-a-message.md b/docs/memory/learnings/2026-10-05-an-alert-is-a-condition-not-a-message.md new file mode 100644 index 0000000000..5d54cc993c --- /dev/null +++ b/docs/memory/learnings/2026-10-05-an-alert-is-a-condition-not-a-message.md @@ -0,0 +1,3 @@ +## 2026-10-05 — pattern — An alert is a condition, not a message: dedup on the subject, not the id +**Context**: #3246. Every platform emitter minted a `request_id` per reading (a timestamp, a reset day, a tier) and relied on `UNIQUE(agent_name, request_id)` ON CONFLICT DO NOTHING for dedup. That made the id the state machine: a persisting condition filed a new row whenever the id rolled over, the same condition read by two workers became two rows, and a condition that cleared left its rows pending until a person acknowledged them — no emitter knew which row was "its" row to end. +**Lesson**: Key an alert on the thing it is ABOUT (`:`), enforce "at most one pending per subject" in the database (a partial unique index that every terminal transition releases), update that row in place on each reading (`last_seen_at`, a seen count), and give the emitter a `clear` it calls where the code knows the condition ended — recorded as the platform's ending, never a person's answer or a timeout. Pair it with a lifetime (no hook is perfect) and a snooze (a person's dismissal must hold). When migrating emitters in batches, a ratchet listing the not-yet-migrated prefixes keeps "the rest follow" honest. diff --git a/docs/memory/requirements/security.md b/docs/memory/requirements/security.md index c6ebaefc5d..646ef215d9 100644 --- a/docs/memory/requirements/security.md +++ b/docs/memory/requirements/security.md @@ -253,7 +253,7 @@ - **Description**: 20.5a made the weekly reading real and 20.5b made it durable; neither tells the operator anything *before* the limit is reached. This adds the alert: a per-subscription warning when the 7-day window crosses a configurable threshold, and a fleet escalation when every registered subscription is saturated at once. - **The window is FIXED-with-reset, not rolling — measured, not assumed.** Querying 20.5b's own history table settled it: `seven_day_resets_at` held CONSTANT at one midnight-UTC instant across five days of probes while utilization climbed 36→90, then STEPPED exactly +7 days. Under a rolling window with continuous usage it would have advanced continuously. (`docs/memory/feature-flows/dashboard-grid-view.md` described these as rolling; that claim is corrected.) Two consequences, both of which REMOVE specified work: 1. Utilization is monotonic non-decreasing inside a window, so the specified **hysteresis floor is dead code** — the only real re-arm is the reset. - 2. `resets_at` IS the window's identity, so the deterministic alert id `sub-headroom-{sid}-{reset-day}-{tier}` becomes the **entire state machine**: `create_item` maps `item["id"]` onto `request_id` under `UNIQUE(agent_name, request_id)` with ON CONFLICT DO NOTHING, so the same window re-emits into the same row (edge-trigger), a reset mints a new id (re-arm), and cross-worker/cross-restart dedup is free. **No durable memo** — nothing to leak, race, or clean up on `delete_subscription`. The id is quantised to the DAY as a belt: were some plan to behave as rolling, it degrades to one alert/day rather than one/probe. + 2. `resets_at` IS the window's identity, so the deterministic alert id `sub-headroom-{sid}-{reset-day}-{tier}` became the **entire state machine** (*superseded by #3246, §26.14: the alert is now one row per subscription keyed by the subject `subscription_headroom:`, updated in place by each reading, ended by the platform when the reading falls back under the threshold, and expiring 14 days after the last reading; `tier` is a material key, so warn → critical escalates in place*): `create_item` maps `item["id"]` onto `request_id` under `UNIQUE(agent_name, request_id)` with ON CONFLICT DO NOTHING, so the same window re-emits into the same row (edge-trigger), a reset mints a new id (re-arm), and cross-worker/cross-restart dedup is free. **No durable memo** — nothing to leak, race, or clean up on `delete_subscription`. The id is quantised to the DAY as a belt: were some plan to behave as rolling, it degrades to one alert/day rather than one/probe. - **The threshold fires; the projection ranks.** There is deliberately no configurable "alert if the reset is less than N hours away" — no such constant is right for every operator. Window length is known and `resets_at` says how much is left, so `projected_end = utilization_pct / fraction_of_window_elapsed` derives the operator's own pace: 75% at day 3 projects to 175%, 75% with hours left to ~77%. Per the operator ruling the alert is **never withheld** at the threshold; the projection sets `priority` (`low` under 100% projected, `high` at or over). An unknowable projection is treated as not-on-pace — a missing `resets_at` is not evidence of an emergency. - **Assessability is a THIRD state, not the existing binary predicate.** `classify_headroom` → `saturated | has_headroom | unassessable`. `_headroom_indicates_healthy` returns `False` for a stale snapshot, a rejected token, a transport error **and** a genuinely saturated subscription, so it cannot distinguish "no evidence" from "no headroom" — used as the gate, the fleet escalation would be blocked by exactly the condition it exists to report. (#2396's docstring named this feature as that predicate's consumer; that was wrong and is corrected. Its BODY is untouched — `resolve_rate_limited_now` consumes it, so a behavioural edit moves every `LIMIT` badge.) The classifier keys on **utilization, not window status**: `allowed_warning` is deliberately non-blocking per #2396, so a status-driven classifier would file a live 90% reading as `has_headroom`. Status remains the *stronger* signal — a blocking 7d status or a probe 429 is `saturated` with or without a number, counting toward the fleet claim but raising no percentage-crossing alert, because there is no percentage to name. Every could-not-tell path returns `unassessable`, including a 5h-only snapshot (`parse_unified_headers` admits that shape). - **Fleet escalation, and the claim it does NOT make**: requires ≥2 subscriptions (with one, "this is full" and "every one is full" are the same fact) and every member assessable — one `unassessable` blocks the claim and is named instead (the ent#100 positive-evidence rule). When it fires the per-subscription alerts are suppressed for that cycle. It states what was **measured** and deliberately does **not** say "auto-switch has no viable target": since abilityai/trinity#2409 the selector ranks its candidates over the SAME cached readings this sweep samples (and drops a fresh provider refusal), but its candidate set is also narrowed by the 2h failure filter the sweep never sees — so the claim stays about the measurement and never promises what the selector will do. @@ -681,6 +681,18 @@ - **Files**: `src/backend/services/skill_gate_service.py`, `src/backend/services/skill_gate_errors.py`, `src/backend/utils/skill_invocation.py`, `src/backend/db/skill_gate_requests.py`, `src/backend/services/ask_service.py` (`may_end`, gate budget), `src/backend/db/operator_queue.py`, `src/backend/services/task_execution_service.py` (backstop), `src/backend/services/dispatch_admission_service.py`, `src/backend/services/chat_execution_service.py`, `src/backend/error_handlers.py`, the callers named above, `src/mcp-server/src/client.ts` (`parseGateResult`), `src/mcp-server/src/tools/{chat,connector}.ts`; the hook (#752): `src/backend/routers/skill_gate.py`, `src/backend/dependencies.py` (`get_self_agent`), `src/backend/services/channel_completion_report.py`, `src/backend/services/agent_service/lifecycle.py` (marker sync), `docker/base-image/hooks/{skill-gate.py,_skill_gate.py}`, `docker/base-image/hooks/managed-settings.d/50-skill-gate.json`, `docker/base-image/agent_server/services/claude_code.py` (isolated turn), `docker/base-image/agent_server/routers/info.py` (`/health`) - **Tests**: `tests/unit/test_ent751_*.py`, `tests/unit/test_ent752_*.py`, `src/mcp-server/src/chat-gate.test.ts` +### 26.14 Platform Alerts — one row per subject, platform endings, a lifetime (OPS-001-PLATFORM-ALERTS) +- **Status**: 🟡 Partially implemented (abilityai/trinity#3246, PR 1 of 2) — the seam, the schema, the upgrade sweep and the surfaces ship now; **only four emitter families are on the seam in this PR** (subscription headroom, legacy skills-library adoption, the system agent's stale base image / start failure, the agent circuit breaker's dormant alarm). Every other platform emitter still files through its old path and is listed by prefix in the ratchet `PREFIXES_NOT_YET_ON_SEAM` (`tests/unit/test_3246_platform_alerts_registry.py`); they follow in PR 2. +- **Description**: A platform alert is a **condition**, not a message. Before #3246 each emitter minted a `request_id` per reading (a timestamp, a day, a tier), so a condition that persisted filed a new row per reading, and a condition that cleared left its rows pending until someone acknowledged them. The dedup key is now the **subject** (`:`, stored in `operator_queue.subject`), not the id. + - **At most one pending row per subject**, enforced by the database: partial unique index `uq_operator_queue_pending_subject (agent_name, subject) WHERE status='pending' AND subject IS NOT NULL`. A newer reading **updates that row in place** (title, question, context, priority; `last_seen_at`; `context.seen_count` + 1; `expires_at` moved) — it never files a second row. + - **The platform ends what it knows has cleared.** `platform_alerts.clear` / `reconcile` end the row `cancelled` with `disposed_by = 'platform'` and `disposition_reason = 'condition_cleared'` (the upgrade sweep's collapsed duplicates read `superseded`). A platform ending is never a person's answer and never a timeout: every surface reads it "Ended by the platform — the condition cleared" (`utils/operatorQueue.js::queueEnding`), and the Workspace projection's `ended_by` is `platform` (`client_portal/asks/service.py::_ending_of`). + - **A lifetime, not "until someone acknowledges it".** Each kind declares a lifetime class: 14 days from the last reading (env `OPERATOR_PLATFORM_ALERT_LIFETIME_DAYS`; subscription headroom and skills adoption), a fixed 30-day net for edge-triggered kinds with a clear hook (base image, circuit breaker), or none (a person must act). A row past its lifetime expires through the normal expiry path. + - **Snooze.** After a person ends an alert, a new reading of the same subject files nothing for 7 days (`DEFAULT_SNOOZE_DAYS`, env `OPERATOR_PLATFORM_ALERT_SNOOZE_DAYS`; a bad value falls back to the default with a WARNING) unless the reading raises the priority or changes one of the kind's `material_keys` (headroom: `tier`) — a warn→critical escalation still reaches a person. + - **Upgrade sweep** (SQLite `platform_alert_subjects` + its Alembic revision): per `(agent_name, derived subject)` keeps the newest pending row, ends the rest as one `superseded` batch, stamps survivors; idempotent; reads only platform-raised rows (`raised_by IS NULL` — an agent's own ask or a gate row under a reserved prefix is never stamped or ended, whatever its id looks like) and never touches a prefix the registry does not know. It collapses the backlog for every kind it can derive, including those whose emitters move in PR 2. Upgrade note: agent rows minted before this release under the four newly reserved prefixes (`skills-reconcile-`, `skills-fleet-reinject-`, `retention-guard-`, `ent615-git-token-scrub-`) become platform-minted on read, so they lose their ending wake and their mirror in the agent's queue file; the pending row itself is untouched. + - **Surfaces.** A pending alert seen more than once reads "seen N times · last seen " on QueueCard, QueueItemDetail and `/m`; an expired alert reads "Expired — nobody acted on it". The canary's L-03 orphan scan exempts the `_skills-sync` alarm host. +- **Files**: `src/backend/services/platform_alerts.py`, `src/backend/db/operator_queue.py`, `src/backend/services/ask_service.py`, the four emitters, `src/frontend/src/utils/operatorQueue.js` +- **Tests**: `tests/unit/test_3246_*.py`, `src/frontend/tests/unit/platformAlertSurfaces.spec.js` + --- ## 28. Agent Guardrails (GUARD-001) diff --git a/docs/memory/requirements/skills.md b/docs/memory/requirements/skills.md index 6ee8ae24a4..96ab8ccfcb 100644 --- a/docs/memory/requirements/skills.md +++ b/docs/memory/requirements/skills.md @@ -48,7 +48,7 @@ #### 21.1.3 Migration & Fresh Installs - **Existing single-repo installs migrate losslessly** (AC#6): a configured `skills_library_url` is adopted as a regular **custom** source (custom, not default — precedence must keep preferring the repo the operator actually chose). The DB row is written before the clone moves, so a crash between them leaves a source that simply re-clones; the reverse order would strand a checkout no row points at. Idempotent and fail-soft - **Fresh installs are pre-configured** (AC#3): the bundled community source is seeded as a **row** during the #1638 fresh-install window (`users` empty, before `_ensure_admin_user`), both dialects. A row, not a read-time code default — a default would resurrect a source the admin deleted and would silently hand an existing install a source it never configured. `TRINITY_DEFAULT_SKILL_SOURCE=""` disables the seed. Never raises: `init_database` runs at import, so a raising seed would crash-loop boot -- **A refusal to adopt is bounded by its own class** (#2744): the terminal *"this install already has skills sources, so it is past migration"* refusal is the designed resting state of a migrated install whose legacy key lingers — not a failure — and `sync_library` re-enters it on **every** sync, unattended under the ent#236 auto-sync loop. It is therefore `priority: "low"` + `logger.info` with a **stable** `request_id` derived from the refused URL (`skills-legacy-adoption-refused-{sha256(url)[:12]}`; the family prefix `skills-legacy-adoption-` is reserved in `operator_queue_service._RESERVED_ID_PREFIXES`, so an agent cannot pre-create the id and self-suppress the alarm through the sink's on-conflict dedup), so `create_item`'s `(agent_name, request_id)` ON CONFLICT collapses every repeat to **one** row per refused URL and an operator's dismissal sticks. A *different* refused URL still raises its own item. The two genuine failure branches — the URL failed validation, or adoption threw — keep `priority: "high"` and timestamped, repeat-visible ids **by product decision** (a repeat carries information there), not because they are bounded. The URL echoed in the log line and in `context.url` is `strip_url_credentials`-scrubbed: the validation-reject branch is exactly the one an `EmbeddedCredentialError` reaches, so the raw value there may carry a PAT (Invariant #12) +- **A refusal to adopt is bounded by its own class** (#2744): the terminal *"this install already has skills sources, so it is past migration"* refusal is the designed resting state of a migrated install whose legacy key lingers — not a failure — and `sync_library` re-enters it on **every** sync, unattended under the ent#236 auto-sync loop. It is therefore `priority: "low"` + `logger.info` with a **stable** `request_id` derived from the refused URL (`skills-legacy-adoption-refused-{sha256(url)[:12]}`; the family prefix `skills-legacy-adoption-` is reserved in `operator_queue_service._RESERVED_ID_PREFIXES`, so an agent cannot pre-create the id and self-suppress the alarm through the sink's on-conflict dedup), so `create_item`'s `(agent_name, request_id)` ON CONFLICT collapses every repeat to **one** row per refused URL and an operator's dismissal sticks. *(Since #3246, §26.14: both branches report through `platform_alerts.observe` under ONE subject per URL (`skills_legacy_adoption:`, alarm host `_skills-sync`); repeats update that row and bump its seen count, the platform ends it when the URL is adopted or no longer configured, it expires 14 days after the last reading, and a dismissal holds for the 7-day snooze unless the priority rises.)* A *different* refused URL still raises its own item. The two genuine failure branches — the URL failed validation, or adoption threw — keep `priority: "high"` and timestamped, repeat-visible ids **by product decision** (a repeat carries information there), not because they are bounded. The URL echoed in the log line and in `context.url` is `strip_url_credentials`-scrubbed: the validation-reject branch is exactly the one an `EmbeddedCredentialError` reaches, so the raw value there may carry a PAT (Invariant #12) #### 21.1.4 Vendor-Neutral Layout — Per-Source Skills Root (ent#332) - **Why**: the SKILL.md format is harness-portable — only the *discovery path* is Claude-specific. The public community catalog (`abilityai/trinity-skills`, ent#296) is legible on its GitHub front page and consumable by non-Claude runtimes: top-level `skills//` plus a root `catalog.yaml`. The platform must consume that layout without breaking any existing `.claude/skills/` source diff --git a/src/backend/canary/snapshot.py b/src/backend/canary/snapshot.py index f642a5dbe6..4d7c35daf8 100644 --- a/src/backend/canary/snapshot.py +++ b/src/backend/canary/snapshot.py @@ -153,8 +153,13 @@ _LOG_ARCHIVE_AGENT = "_log-archive" # ent#434: the weekly-headroom alarm host. _SUB_HEADROOM_AGENT = "_sub-headroom" +# #3246: the skills-library alarm host (mirrors +# services.skill_service.RECONCILE_ALARM_AGENT_NAME) — its rows now arrive +# through the platform-alert seam and would otherwise read as L-03 orphans. +_SKILLS_SYNC_AGENT = "_skills-sync" _PLATFORM_ALARM_SENTINELS = (_RETENTION_GUARD_AGENT, _DB_BACKUP_AGENT, - _LOG_ARCHIVE_AGENT, _SUB_HEADROOM_AGENT) + _LOG_ARCHIVE_AGENT, _SUB_HEADROOM_AGENT, + _SKILLS_SYNC_AGENT) _SENTINEL_SQL_LIST = ", ".join(f"'{name}'" for name in _PLATFORM_ALARM_SENTINELS) ORPHAN_SCAN_TABLES = [ diff --git a/src/backend/client_portal/asks/models.py b/src/backend/client_portal/asks/models.py index 6a0a3a18d1..ebbb5bad39 100644 --- a/src/backend/client_portal/asks/models.py +++ b/src/backend/client_portal/asks/models.py @@ -35,7 +35,7 @@ class WorkspaceAsk(BaseModel): # (trinity-enterprise#611); the ANSWER response projects the row it recorded. status: str # How the ask ended, COARSE on purpose (trinity-enterprise#611): `ended_by` - # is `you` | `operator` | `timeout`, never an email, and the operator's + # is `you` | `operator` | `platform` (#3246) | `timeout`, never an email, and the operator's # cancel reason never crosses. `ended_at` is when it ended — None when the # platform does not know (a row that ended before the ledger), never the # time the ask was filed. diff --git a/src/backend/client_portal/asks/service.py b/src/backend/client_portal/asks/service.py index 7b0eb9b410..25eaa633b5 100644 --- a/src/backend/client_portal/asks/service.py +++ b/src/backend/client_portal/asks/service.py @@ -108,7 +108,9 @@ def _ending_of(item: dict, viewer_email: Optional[str]) -> tuple: """`(ended_at, ended_by)` for a client — COARSE on purpose. `ended_by` is `you` (the viewer answered), `operator` (another person - answered or cancelled) or `timeout`; never an email and never the cancel + answered or cancelled), `platform` (the platform ended the row itself — + `disposed_by = 'platform'`, #3246: never a person's answer, never a + timeout) or `timeout`; never an email and never the cancel reason (both are the operator's, not the client's). `ended_at` is the ledger's time, or a legacy answer's time — never `created_at`, which is when the ask was filed, not when it ended. @@ -118,6 +120,8 @@ def _ending_of(item: dict, viewer_email: Optional[str]) -> tuple: return None, None if status == "expired": return item.get("disposed_at"), "timeout" + if item.get("disposed_by") == "platform": + return item.get("disposed_at"), "platform" by = item.get("disposed_by_email") or (item.get("responded_by_email") if status == "answered" else None) who = "you" if by and viewer_email and by.lower() == viewer_email.lower() else "operator" at = item.get("disposed_at") or (item.get("responded_at") if status == "answered" else None) diff --git a/src/backend/database.py b/src/backend/database.py index 9439a34cbb..a72391e99c 100644 --- a/src/backend/database.py +++ b/src/backend/database.py @@ -3901,6 +3901,24 @@ def mark_operator_queue_acknowledged(self, agent_name, item_id): # #1631: agent-scoped — item_id is the agent's request_id, not the uuid. return self._operator_queue_ops.mark_acknowledged(agent_name, item_id) + # #3246 — platform alerts: one pending row per (agent, subject) + def find_pending_operator_queue_by_subject(self, agent_name, subject): + return self._operator_queue_ops.find_pending_by_subject(agent_name, subject) + + def find_person_ended_operator_queue_by_subject(self, agent_name, subject, since): + return self._operator_queue_ops.find_person_ended_by_subject(agent_name, subject, since) + + def create_platform_operator_queue_item(self, agent_name, item, *, subject, + max_pending_for_type=None): + # Returns {"outcome": created|updated|refused_at_budget, "row", "changed"}. + return self._operator_queue_ops.create_platform_item( + agent_name, item, subject=subject, max_pending_for_type=max_pending_for_type, + ) + + def end_operator_queue_items_by_platform(self, ids, *, reason, batch_id=None): + # Returns {"batch_id", "rows"} — the rows THIS call ended (CAS-won). + return self._operator_queue_ops.end_items_by_platform(ids, reason=reason, batch_id=batch_id) + def mark_operator_queue_expired(self): # trinity-enterprise#611: returns the rows this sweep ended (was a count). return self._operator_queue_ops.mark_expired() diff --git a/src/backend/db/migrations.py b/src/backend/db/migrations.py index 4cad464811..efa4f09643 100644 --- a/src/backend/db/migrations.py +++ b/src/backend/db/migrations.py @@ -5178,6 +5178,169 @@ def _migrate_supersede_queue_flood_backlog(cursor, conn): conn.commit() +# #3246: the shared sweep — both tracks run this function over their own +# connection (the Alembic revision imports it), so the survivor rule cannot +# drift between them. `run(sql, params, fetch=False)` is the one driver seam: +# sqlite3 and SQLAlchemy `text()` both bind `:name` parameters. +# `raised_by IS NULL` is the platform-row discriminator: the agent file seam +# writes 'agent' and gates write 'gate', and an agent can mint an id under a +# reserved prefix on its own name (the git-token-scrub prefix was unreserved +# before #3246), so the id shape alone must never pick a survivor. +_PLATFORM_ALERT_SWEEP_SELECT = """ +SELECT id, agent_name, request_id, status, created_at, expires_at, context, + subject, disposed_by, disposed_at, raised_by +FROM operator_queue +WHERE raised_by IS NULL + AND (status = 'pending' + OR (disposed_by = 'person' AND subject IS NULL AND disposed_at >= :since)) +""" + +_PLATFORM_ALERT_SWEEP_END = """ +UPDATE operator_queue +SET status = 'cancelled', + disposition = 'cancelled', + disposed_at = :now, + disposed_by = 'platform', + disposed_by_email = NULL, + disposition_reason = 'superseded', + batch_id = :batch_id +WHERE status = 'pending' AND raised_by IS NULL AND id IN ({ids}) +""" + +_SWEEP_CHUNK = 400 # well under every driver's bound-parameter limit + + +def run_platform_alert_sweep(run, *, now=None, batch_id=None) -> dict: + """#3246: collapse the platform-alert backlog to one pending row per subject. + + Reads every pending row plus the person-ended rows inside the snooze + window, hands them to the leaf's pure ``plan_sweep`` (the ONE source of + per-kind id parsing — imported function-locally so neither migration + track pulls the service graph into ``init_database()``), then writes: + + * survivors: ``subject``, ``last_seen_at``, ``context.seen_count = 1`` and + ``expires_at`` only where it was NULL; + * every other pending row of the subject: ended as ONE batch — + ``cancelled`` / ``platform`` / ``superseded`` / NULL email / ``batch_id`` + — the ent#611 vocabulary, compare-and-set on ``status = 'pending'`` so a + row a person ends first is never overwritten; + * known kinds with no derivable subject: ``expires_at`` only (the row + leaves on its own; nothing is merged on a guess); + * person-ended rows inside the snooze window: ``subject`` only. + + Agent-raised rows and gate rows (``raised_by IS NOT NULL``) are excluded + by the SELECT and skipped again by the planner; external prefixes never + derive. None of them is ever stamped or ended. Idempotent: a second run finds the survivors stamped + and plans nothing. Returns the write counts for the boot log. + """ + import json as _json + import uuid as _uuid + from datetime import datetime as _dt, timezone as _tz + + from services.platform_alerts import plan_sweep, snooze_window + + now = now or _dt.now(_tz.utc) + now_iso = now.strftime("%Y-%m-%dT%H:%M:%S.%fZ") + since_iso = (now - snooze_window()).strftime("%Y-%m-%dT%H:%M:%S.%fZ") + cols = ("id", "agent_name", "request_id", "status", "created_at", "expires_at", + "context", "subject", "disposed_by", "disposed_at", "raised_by") + rows = [dict(zip(cols, r)) for r in run(_PLATFORM_ALERT_SWEEP_SELECT, {"since": since_iso}, fetch=True)] + plan = plan_sweep(rows, now=now) + counts = {"survivors": 0, "ended": 0, "lifetime_only": 0, "snoozed": 0} + if plan.is_empty(): + return counts + + for stamp in plan.survivor_stamps: + # `expires_at` is only in the statement when the planner set one: an + # untyped NULL bound into COALESCE is driver-dependent on PostgreSQL, + # and a stamp that carries none has nothing to write there. + params = {"id": stamp.id, "subject": stamp.subject, "last_seen_at": stamp.last_seen_at, + "context": _json.dumps(stamp.context) if stamp.context is not None else None} + expiry_sql = "" + if stamp.expires_at is not None: + expiry_sql = ", expires_at = COALESCE(expires_at, :expires_at)" + params["expires_at"] = stamp.expires_at + run( + "UPDATE operator_queue SET subject = :subject, last_seen_at = :last_seen_at, " + f"context = :context{expiry_sql} " + "WHERE id = :id AND status = 'pending' AND raised_by IS NULL", + params, + ) + counts["survivors"] += 1 + + batch = batch_id or _uuid.uuid4().hex + ended = list(plan.ended_ids) + for start in range(0, len(ended), _SWEEP_CHUNK): + chunk = ended[start:start + _SWEEP_CHUNK] + params = {f"e{i}": id_ for i, id_ in enumerate(chunk)} + placeholders = ", ".join(f":{k}" for k in params) + params.update({"now": now_iso, "batch_id": batch}) + run(_PLATFORM_ALERT_SWEEP_END.format(ids=placeholders), params) + counts["ended"] += len(chunk) + + for stamp in plan.lifetime_stamps: + run("UPDATE operator_queue SET expires_at = :expires_at " + "WHERE id = :id AND status = 'pending' AND expires_at IS NULL AND raised_by IS NULL", + {"id": stamp.id, "expires_at": stamp.expires_at}) + counts["lifetime_only"] += 1 + + for stamp in plan.snooze_stamps: + run("UPDATE operator_queue SET subject = :subject " + "WHERE id = :id AND subject IS NULL AND raised_by IS NULL", + {"id": stamp.id, "subject": stamp.subject}) + counts["snoozed"] += 1 + return counts + + +# Both tracks create the same two indexes, by name (schema.py declares them too). +PLATFORM_ALERT_INDEX_DDL = ( + "CREATE UNIQUE INDEX IF NOT EXISTS uq_operator_queue_pending_subject " + "ON operator_queue(agent_name, subject) WHERE status = 'pending' AND subject IS NOT NULL", + "CREATE INDEX IF NOT EXISTS idx_operator_queue_agent_subject ON operator_queue(agent_name, subject)", +) + + +def _migrate_platform_alert_subjects(cursor, conn): + """#3246 — platform alerts: one pending row per subject. + + Ordered columns → sweep → index, in ONE migration: the partial unique index + can only be created once the sweep has collapsed the duplicates an + installed backlog holds, so the three steps cannot be split across + migrations without breaking the upgrade of exactly the installs this is + for. Adds `subject TEXT` and `last_seen_at TEXT` (nullable, PRAGMA-guarded), + runs `run_platform_alert_sweep` (the leaf's `plan_sweep` decides; see its + docstring for the rules), then creates `uq_operator_queue_pending_subject` + (partial, pending + subject only) and `idx_operator_queue_agent_subject`. + + No wake and no file write follow: these are platform-minted rows + (`is_platform_minted`), which take no part in either. Idempotent. + + Mirrored by the Alembic revision 0090_platform_alert_subjects. + """ + for column in ("subject", "last_seen_at"): + _safe_add_column( + cursor, + "operator_queue", + column, + f"ALTER TABLE operator_queue ADD COLUMN {column} TEXT", + log_msg=f"Adding {column} to operator_queue for platform alerts (#3246)", + ) + conn.commit() + + def _run(sql, params, fetch=False): + cursor.execute(sql, params) + return cursor.fetchall() if fetch else None + + counts = run_platform_alert_sweep(_run) + conn.commit() + if any(counts.values()): + logger.info("Platform-alert backlog sweep (#3246): %s", counts) + + for ddl in PLATFORM_ALERT_INDEX_DDL: + cursor.execute(ddl) + conn.commit() + + MIGRATIONS = [ ("agent_sharing", _migrate_agent_sharing_table), ("schedule_executions_observability", _migrate_schedule_executions_observability), @@ -5337,4 +5500,5 @@ def _migrate_supersede_queue_flood_backlog(cursor, conn): ("pull_sync", _migrate_pull_sync), ("skill_gate_requests_table", _migrate_skill_gate_requests_table), ("supersede_queue_flood_backlog", _migrate_supersede_queue_flood_backlog), + ("platform_alert_subjects", _migrate_platform_alert_subjects), ] diff --git a/src/backend/db/operator_queue.py b/src/backend/db/operator_queue.py index e7fad0ce54..3f489572fc 100644 --- a/src/backend/db/operator_queue.py +++ b/src/backend/db/operator_queue.py @@ -19,6 +19,7 @@ from datetime import datetime from sqlalchemy import select, update, func, and_, or_, case, delete +from sqlalchemy.exc import IntegrityError from .engine import get_engine, make_insert from .tables import operator_queue @@ -172,6 +173,9 @@ def _row_to_item(row) -> Dict: "resolved_to": json.loads(row["resolved_to"]) if row["resolved_to"] else None, "proposal": json.loads(row["proposal"]) if row["proposal"] else None, "supersedes_expired": row["supersedes_expired"], + # #3246 — the platform alert's condition key and its latest reading + "subject": row["subject"], + "last_seen_at": row["last_seen_at"], } # Columns selected for a full queue-item record, in the canonical order. @@ -217,6 +221,8 @@ def _row_to_item(row) -> Dict: operator_queue.c.resolved_to, operator_queue.c.proposal, operator_queue.c.supersedes_expired, + operator_queue.c.subject, # #3246 — platform alerts: kind:key + operator_queue.c.last_seen_at, ) def create_item( @@ -391,6 +397,8 @@ def _insert_values( *, channel: Optional[str], raised_by: Optional[str], + subject: Optional[str] = None, + last_seen_at: Optional[str] = None, ) -> Tuple[str, Dict]: """The DB-sink belts and the column values of a new row, shared by every create so the belts cannot drift between the file and native paths.""" @@ -477,9 +485,208 @@ def _insert_values( # it does not decide it, and it must never derive it from `context` # (which is agent-authored). addressed_to_email=item.get("addressed_to_email"), + # #3246: platform-owned, keyword-only — never read from `item`. + subject=subject, + last_seen_at=last_seen_at, ) return request_id, values + # ------------------------------------------------------------------ + # #3246 — platform alerts: one pending row per (agent, subject) + # ------------------------------------------------------------------ + _RENDERED_FIELDS = ("title", "question", "priority") + + def find_pending_by_subject(self, agent_name: str, subject: str) -> Optional[Dict]: + """The one pending row for `(agent, subject)`, or None (#3246).""" + with get_engine().connect() as conn: + row = self._find_pending_by_subject(conn, agent_name, subject) + return self._row_to_item(row) if row else None + + def _find_pending_by_subject(self, conn, agent_name: str, subject: str): + return conn.execute( + select(*self._SELECT_COLS).where(and_( + operator_queue.c.agent_name == agent_name, + operator_queue.c.subject == subject, + operator_queue.c.status == "pending", + )) + ).mappings().first() + + def find_person_ended_by_subject( + self, agent_name: str, subject: str, since: str, + ) -> Optional[Dict]: + """The newest row for `(agent, subject)` a PERSON ended at or after + `since` (ISO-Z text, Invariant #16), any terminal status — the seam's + snooze read (#3246). Served by `idx_operator_queue_agent_subject`.""" + stmt = ( + select(*self._SELECT_COLS) + .where(and_( + operator_queue.c.agent_name == agent_name, + operator_queue.c.subject == subject, + operator_queue.c.status != "pending", + operator_queue.c.disposed_by == "person", + operator_queue.c.disposed_at.isnot(None), + operator_queue.c.disposed_at >= since, + )) + .order_by(operator_queue.c.disposed_at.desc(), operator_queue.c.id.desc()) + .limit(1) + ) + with get_engine().connect() as conn: + row = conn.execute(stmt).mappings().first() + return self._row_to_item(row) if row else None + + def create_platform_item( + self, + agent_name: str, + item: Dict, + *, + subject: Optional[str], + max_pending_for_type: Optional[int] = None, + ) -> Dict: + """Record a platform reading: find → touch → count → insert, in ONE + locked transaction (#3246). + + `subject` is the condition key (`kind:key`); None is an event, which is + always a new row. With a subject, the pending row for `(agent, subject)` + is updated in place — title / question / priority / context / + `last_seen_at` / `expires_at`, `context.seen_count` + 1 — by a + compare-and-set on `status = 'pending'`: a row a person ended first is + never overwritten (the lock covers creates, not `respond_to_item`), and + a lost CAS falls through to a fresh row. The find runs BEFORE the + #1677 per-type count, so a would-be update is never refused at budget; + `max_pending_for_type` is the cap for a budgeted kind (None: unbudgeted). + + Returns `{"outcome": "created" | "updated" | "refused_at_budget", + "row", "changed"}`; `changed` is whether a RENDERED field (title / + question / priority / context minus `seen_count`) moved, so a bare + repeat reading sends no broadcast. The partial unique index + `uq_operator_queue_pending_subject` is the backstop for a lock that + failed open: its `IntegrityError` is caught and the call re-finds and + touches the row that won. + """ + now = utc_now_iso() + for attempt in (1, 2): + try: + with get_engine().begin() as conn: + self._lock_agent_for_create(conn, agent_name) + if subject is not None: + current = self._find_pending_by_subject(conn, agent_name, subject) + if current is not None: + touched = self._touch(conn, current, item, now=now) + if touched is not None: + return touched + if attempt == 2: + # The index said a pending row existed and it is gone + # (ended between the two transactions): fall through. + pass + if max_pending_for_type is not None: + pending = conn.execute( + select(func.count()).where(and_( + *_own_pending_conds(agent_name), + operator_queue.c.type == item.get("type", "question"), + )) + ).scalar() or 0 + if pending >= max_pending_for_type: + return {"outcome": "refused_at_budget", "row": None, "changed": False} + context = item.get("context") + context = dict(context) if isinstance(context, dict) else {} + if subject is not None: + context["seen_count"] = 1 + values_item = {**item, "context": context or None} + request_id, values = self._insert_values( + agent_name, values_item, channel=None, raised_by=None, + subject=subject, last_seen_at=now, + ) + conn.execute( + make_insert(operator_queue).values(**values).on_conflict_do_nothing( + index_elements=["agent_name", "request_id"]) + ) + row = conn.execute(select(*self._SELECT_COLS).where(and_( + operator_queue.c.agent_name == agent_name, + operator_queue.c.request_id == request_id, + ))).mappings().first() + return {"outcome": "created", "row": self._row_to_item(row), "changed": True} + except IntegrityError: + if subject is None or attempt == 2: + raise + # the partial unique index won the race the lock did not cover: + # a pending row for the subject now exists — re-find and touch it. + continue + + def _touch(self, conn, current, item: Dict, *, now: str) -> Optional[Dict]: + """Update the pending row `current` in place; None when the CAS lost.""" + old_context = json.loads(current["context"]) if current["context"] else {} + if not isinstance(old_context, dict): + old_context = {} + new_context = item.get("context") + new_context = dict(new_context) if isinstance(new_context, dict) else {} + seen = old_context.get("seen_count") + new_context["seen_count"] = (seen if isinstance(seen, int) and seen > 0 else 1) + 1 + changed = any( + (item.get(f) if item.get(f) is not None else current[f]) != current[f] + for f in self._RENDERED_FIELDS + ) or {k: v for k, v in old_context.items() if k != "seen_count"} != {k: v for k, v in new_context.items() if k != "seen_count"} + context_json = json.dumps(new_context) + if len(context_json.encode("utf-8")) > _DB_BELT_CONTEXT_MAX_BYTES: + raise ValueError(f"operator-queue 'context' exceeds {_DB_BELT_CONTEXT_MAX_BYTES} bytes") + result = conn.execute( + update(operator_queue) + .where(and_(operator_queue.c.id == current["id"], operator_queue.c.status == "pending")) + .values( + title=item.get("title") or current["title"], + question=item.get("question") or current["question"], + priority=item.get("priority") or current["priority"], + context=context_json, + last_seen_at=now, + expires_at=_iso_z_deadline(item.get("expires_at")), + ) + ) + if not result.rowcount: + return None + row = conn.execute( + select(*self._SELECT_COLS).where(operator_queue.c.id == current["id"]) + ).mappings().first() + return {"outcome": "updated", "row": self._row_to_item(row), "changed": bool(changed)} + + def end_items_by_platform( + self, ids: List[str], *, reason: str, batch_id: Optional[str] = None, + ) -> Dict: + """End the listed pending rows as the PLATFORM (#3246): the + `bulk_cancel_items` shape — one compare-and-set UPDATE on + `status = 'pending'` stamping the ent#611 ledger with + `disposed_by = 'platform'`, no email — re-selected by `batch_id`, so + the rows returned are exactly the ones THIS call ended; a row a person + ended first is skipped, never re-ended. Returns `{"batch_id", "rows"}`; + `batch_id` is None when nothing was ended.""" + empty = {"batch_id": None, "rows": []} + ids = list(dict.fromkeys(ids or ())) + if not ids: + return empty + batch_id = batch_id or uuid.uuid4().hex + now = utc_now_iso() + with get_engine().begin() as conn: + result = conn.execute( + update(operator_queue) + .where(and_(operator_queue.c.status == "pending", operator_queue.c.id.in_(ids))) + .values( + status="cancelled", + disposition="cancelled", + disposed_at=now, + disposed_by="platform", + disposed_by_email=None, + disposition_reason=reason, + batch_id=batch_id, + ) + ) + if result.rowcount == 0: + return empty + rows = conn.execute( + select(*self._SELECT_COLS).where(and_( + operator_queue.c.id.in_(ids), + operator_queue.c.batch_id == batch_id, + )) + ).mappings().all() + return {"batch_id": batch_id, "rows": [self._row_to_item(r) for r in rows]} + def get_item(self, item_id: str) -> Optional[Dict]: """Get a single queue item by ID.""" stmt = select(*self._SELECT_COLS).where(operator_queue.c.id == item_id) @@ -1218,6 +1425,11 @@ def mark_expired(self) -> List[Dict]: and_( operator_queue.c.id == item_id, operator_queue.c.status == "pending", + # #3246: a platform alert refreshed between the + # candidate select and this CAS moved its deadline; + # it is not expired, and the refresh wins. + operator_queue.c.expires_at.isnot(None), + operator_queue.c.expires_at < now, ) ) .values( diff --git a/src/backend/db/schema.py b/src/backend/db/schema.py index 68951aa670..726719b0ae 100644 --- a/src/backend/db/schema.py +++ b/src/backend/db/schema.py @@ -1744,6 +1744,8 @@ resolved_to TEXT, proposal TEXT, supersedes_expired TEXT, + subject TEXT, + last_seen_at TEXT, FOREIGN KEY (responded_by_id) REFERENCES users(id) ) """, @@ -2271,6 +2273,15 @@ # platform-minted `id` (uuid) is the global handle; `request_id` carries the # agent's string, so two agents can reuse the same id without collision. "CREATE UNIQUE INDEX IF NOT EXISTS idx_operator_queue_agent_request ON operator_queue(agent_name, request_id)", + # #3246: a platform alert is a condition, not a message — at most ONE + # pending row per (agent, subject). Partial: ended rows and subject-less + # rows (agent-raised asks, gates, not-yet-migrated emitters) stay outside + # it. Created by the migration only AFTER the backlog sweep collapsed + # duplicates, so an install holding them upgrades cleanly. + "CREATE UNIQUE INDEX IF NOT EXISTS uq_operator_queue_pending_subject ON operator_queue(agent_name, subject) WHERE status = 'pending' AND subject IS NOT NULL", + # The seam's snooze lookup reads person-ended rows by subject — a partial + # index over pending rows cannot serve it. + "CREATE INDEX IF NOT EXISTS idx_operator_queue_agent_subject ON operator_queue(agent_name, subject)", # trinity-enterprise#751 — the gate's caps count pending rows per executor # and per requester; the sweep reads by state. "CREATE INDEX IF NOT EXISTS idx_skill_gate_requests_agent_state ON skill_gate_requests(agent_name, state)", diff --git a/src/backend/db/tables.py b/src/backend/db/tables.py index f5a71590f7..14f48eda8b 100644 --- a/src/backend/db/tables.py +++ b/src/backend/db/tables.py @@ -1508,6 +1508,20 @@ def process_bind_param(self, value, dialect): Column("resolved_to", Text), # JSON list of person refs Column("proposal", Text), # JSON — the frozen action Column("supersedes_expired", Text), # the predecessor row's uuid + # #3246: a platform alert is a condition, not a message. `subject` is + # `kind:key`, the dedup key the seam keeps at most one pending row for; + # `last_seen_at` moves on every reading. Both nullable, no backfill beyond + # the upgrade sweep: agent-raised asks and gates never carry a subject. + Column("subject", Text), + Column("last_seen_at", Text), + Index( + "uq_operator_queue_pending_subject", + "agent_name", "subject", + unique=True, + sqlite_where=text("status = 'pending' AND subject IS NOT NULL"), + postgresql_where=text("status = 'pending' AND subject IS NOT NULL"), + ), + Index("idx_operator_queue_agent_subject", "agent_name", "subject"), ) skill_gate_requests = Table( diff --git a/src/backend/dependencies.py b/src/backend/dependencies.py index 5fd2b8af19..178feb611d 100644 --- a/src/backend/dependencies.py +++ b/src/backend/dependencies.py @@ -1162,8 +1162,9 @@ def is_interactive_principal(current_user: User) -> bool: def is_person_principal(current_user: User) -> bool: """Is this caller a PERSON — the only author an ask's ending may record? - The endings ledger records `disposed_by` as an enum of two, `person` or - `timeout` (trinity-enterprise#611). An agent-scoped key resolves to its OWNER + The endings ledger records `disposed_by` as `person`, `timeout` or — for a + platform alert the platform itself ended (#3130/#3246) — `platform`; only + `person` is a principal (trinity-enterprise#611). An agent-scoped key resolves to its OWNER carrying the owner's role, so before this an agent could answer or cancel any ask its owner could reach — its own approval included — and the row recorded the owner. Recording that as `person` would make the ledger lie. diff --git a/src/backend/main.py b/src/backend/main.py index a7efcbfe43..0012c8a18c 100644 --- a/src/backend/main.py +++ b/src/backend/main.py @@ -1177,6 +1177,11 @@ async def lifespan(app: FastAPI): A thin orchestrator over the phase helpers above (#1028). Read the phase list as the boot sequence; each name links to the block it used to inline. """ + # #3246: background work spawned from a thread anyio does not own + # (`asyncio.to_thread` — the headroom sweep, the skills reconcile) hops back + # to THIS loop; capture it before any phase can spawn from a thread. + from services import operator_resume_service as _ors + _ors.remember_host_loop() await _init_logging_and_first_run_notice() await _start_event_bus() await _log_startup_environment() diff --git a/src/backend/migrations/versions/0090_platform_alert_subjects.py b/src/backend/migrations/versions/0090_platform_alert_subjects.py new file mode 100644 index 0000000000..1190497da1 --- /dev/null +++ b/src/backend/migrations/versions/0090_platform_alert_subjects.py @@ -0,0 +1,58 @@ +"""platform_alert_subjects — one pending platform alert per subject (#3246). + +PostgreSQL half of the dual-track pair; the SQLite half is +``db/migrations.py::platform_alert_subjects``. Ordered columns → sweep → +index in ONE revision: the partial unique index can only be created once the +sweep has collapsed the duplicates an installed backlog holds. + +Adds ``subject TEXT`` and ``last_seen_at TEXT`` (nullable, no backfill beyond +the sweep), runs the shared ``run_platform_alert_sweep`` (both tracks call the +same function over their own connection, so the survivor rule cannot drift: +per (agent, subject) the newest pending row — by ``created_at``, then ``id`` — +stays stamped; the rest end ``cancelled`` / ``disposed_by = 'platform'`` / +``superseded`` as one batch; rows whose subject cannot be derived are never +merged on a guess), then creates ``uq_operator_queue_pending_subject`` +(partial: ``status = 'pending' AND subject IS NOT NULL``) and +``idx_operator_queue_agent_subject``. + +Revision ID: 0090_platform_alert_subjects +Revises: 0089_supersede_queue_flood_backlog +""" +from alembic import op +from sqlalchemy import text + + +revision = "0090_platform_alert_subjects" +down_revision = "0089_supersede_queue_flood_backlog" +branch_labels = None +depends_on = None + +_COLUMNS = ("subject", "last_seen_at") +_INDEXES = ("uq_operator_queue_pending_subject", "idx_operator_queue_agent_subject") + + +def upgrade() -> None: + from db.migrations import PLATFORM_ALERT_INDEX_DDL, run_platform_alert_sweep + + for column in _COLUMNS: + op.execute(f"ALTER TABLE operator_queue ADD COLUMN IF NOT EXISTS {column} TEXT") + + bind = op.get_bind() + + def _run(sql, params, fetch=False): + result = bind.execute(text(sql), params) + return result.fetchall() if fetch else None + + run_platform_alert_sweep(_run) + + for ddl in PLATFORM_ALERT_INDEX_DDL: + op.execute(ddl) + + +def downgrade() -> None: + for name in reversed(_INDEXES): + op.execute(f"DROP INDEX IF EXISTS {name}") + for column in reversed(_COLUMNS): + op.execute(f"ALTER TABLE operator_queue DROP COLUMN IF EXISTS {column}") + # The sweep's endings are deliberately not reverted: the inverse is + # "re-open hundreds of duplicate platform alarms", which no rollback wants. diff --git a/src/backend/routers/operator_queue.py b/src/backend/routers/operator_queue.py index be9fc45846..060c2e5f6c 100644 --- a/src/backend/routers/operator_queue.py +++ b/src/backend/routers/operator_queue.py @@ -95,6 +95,7 @@ def _assert_agent_accessible(agent_name: str, accessible: Optional[Set[str]]) -> "disposition", "disposed_at", "disposed_by", "disposition_reason", "batch_id", "raised_by", "channel", "to_role", "proposal", "supersedes_expired", "aging", "aged_since", + "subject", "last_seen_at", # #3246: a platform alert's condition key + latest reading ) diff --git a/src/backend/services/agent_client/circuit.py b/src/backend/services/agent_client/circuit.py index 8f89ddb12a..5cbe1bf761 100644 --- a/src/backend/services/agent_client/circuit.py +++ b/src/backend/services/agent_client/circuit.py @@ -209,55 +209,57 @@ def is_circuit_failure(exc: BaseException) -> bool: def _emit_dormant_alert(agent_name: str) -> None: - """Insert a circuit_breaker_dormant operator-queue entry. + """Report the circuit_breaker_dormant condition through the platform + alert seam (#3246): one pending row per agent, ended by the platform when + the circuit closes again (`_clear_dormant_alert`). Called from CircuitState.record_failure on the closed/open → dormant transition. The Lua atomicity of _RECORD_FAILURE_LUA guarantees exactly - one worker observes the transition, so this fires at most once per - distinct dormant entry — no de-dupe layer required. + one worker observes the transition; the seam keeps it one row anyway. """ try: - from database import db - from utils.helpers import utc_now_iso - now = utc_now_iso() - # Use the generic 'alert' type so the existing Operating Room UI - # (QueueCard.vue / QueueItemDetail.vue branch on 'approval|question|alert') - # renders an Acknowledge control. The narrower discriminator goes in - # context.alert_type for callers that want to filter — same pattern - # the existing `sync_failing` work should adopt when its UI is touched. - item = { - "id": f"cb-dormant-{agent_name}-{now}", - "agent_name": agent_name, - "type": "alert", - "status": "pending", - "priority": "high", - "title": "Agent circuit breaker DORMANT", - "question": ( + from services import platform_alerts + # The seam files type 'alert', so the Operating Room UI (QueueCard.vue / + # QueueItemDetail.vue branch on 'approval|question|alert') renders an + # Acknowledge control. The narrower discriminator goes in + # context.alert_type for callers that want to filter. + outcome = platform_alerts.observe( + agent_name, "circuit_dormant", agent_name, + priority="high", + title="Agent circuit breaker DORMANT", + question=( f"{agent_name}'s circuit breaker entered DORMANT after " f"{CIRCUIT_DORMANT_AFTER_OPEN_PROBES} consecutive failed probes. " f"Scheduled tasks fast-fail until the agent recovers via the " f"~{int(CIRCUIT_DORMANT_COOLDOWN_SECONDS / 60)} min cooldown " f"probe or an admin reset." ), - "context": { + context={ "agent_name": agent_name, "alert_type": "circuit_breaker_dormant", "transition": "dormant", "dormant_after_open_probes": CIRCUIT_DORMANT_AFTER_OPEN_PROBES, "dormant_cooldown_seconds": CIRCUIT_DORMANT_COOLDOWN_SECONDS, }, - "created_at": now, - } - db.create_operator_queue_item(agent_name, item) + ) logger.warning( - "[CB] circuit_breaker_dormant operator-queue entry emitted for %s", - agent_name, + "[CB] circuit_breaker_dormant operator-queue entry for %s: %s", + agent_name, outcome, ) except Exception: # Don't let alert-delivery failure mask the breaker transition. logger.exception("[CB] failed to emit dormant alert for %s", agent_name) +def _clear_dormant_alert(agent_name: str) -> None: + """The circuit closed: the platform ends the dormant alert (#3246).""" + try: + from services import platform_alerts + platform_alerts.clear(agent_name, "circuit_dormant", agent_name) + except Exception: + logger.exception("[CB] failed to clear dormant alert for %s", agent_name) + + class CircuitState: """Per-agent circuit breaker, Redis-backed (#631). @@ -362,6 +364,8 @@ def record_success(self) -> None: "Circuit CLOSED for agent %s (recovered from %s)", self.agent_name, prior, ) + if prior == "dormant": + _clear_dormant_alert(self.agent_name) except Exception as e: logger.warning("Circuit record_success swallowed (%s)", e) _reset_circuit_redis_client() @@ -486,5 +490,7 @@ def reset_circuit(agent_name: str) -> None: logger.info("Circuit reset to CLOSED for %s", agent_name) except Exception as e: logger.warning("reset_circuit(%s) swallowed: %s", agent_name, e) + return + _clear_dormant_alert(agent_name) diff --git a/src/backend/services/ask_service.py b/src/backend/services/ask_service.py index 4979fa89fd..ff4cb53734 100644 --- a/src/backend/services/ask_service.py +++ b/src/backend/services/ask_service.py @@ -347,6 +347,58 @@ def expire() -> Ending: return _ended(EndingEvent(EXPIRED, tuple(rows), None), audit, None) +# #3246: the platform ends its own alerts for exactly these reasons — DATA on +# the ledger (`disposition_reason`), rendered by name on the card. A person's +# cancel reason is free text; the platform's is a closed vocabulary. +CONDITION_CLEARED = "condition_cleared" +SUPERSEDED = "superseded" +PLATFORM_ENDING_REASONS = (CONDITION_CLEARED, SUPERSEDED) + + +def clear_platform(ids: Iterable[str], *, reason: str, batch_id: Optional[str] = None) -> Ending: + """The PLATFORM ended its own alerts: the condition cleared, or a newer + reading superseded them (#3246). Mirrors `expire`: no Actor, the ending is + `cancelled` / `disposed_by = 'platform'` / NULL email, and only the rows + this call won the compare-and-set for are returned and handed to the + observers — a row a person ended first is skipped, never re-ended. One + audit row (`platform_cleared`) and one thin `operator_queue_cancelled` + trigger per agent; listeners refetch through the access-controlled list. + """ + if reason not in PLATFORM_ENDING_REASONS: + raise ValueError(f"platform ending reason must be one of {PLATFORM_ENDING_REASONS}, got {reason!r}") + ids = list(dict.fromkeys(ids)) + out = db.end_operator_queue_items_by_platform(ids, reason=reason, batch_id=batch_id) + rows, batch_id = out["rows"], out["batch_id"] + if not rows: + return Ending(rows=[], batch_id=None) + audit = [{ + "event_action": "platform_cleared", + "source": "system", + "target_type": "operator_queue", + "details": { + "batch_id": batch_id, + "reason": reason, + "cancelled": len(rows), + "skipped": len(ids) - len(rows), + "ids": [r["id"] for r in rows], + "agent_names": sorted({r["agent_name"] for r in rows}), + }, + }] + agents = list(dict.fromkeys(r["agent_name"] for r in rows)) + triggers = [_broadcast_payload({"type": "operator_queue_cancelled", + "data": {"agent_name": agent, "batch_id": batch_id}}) + for agent in agents] + for extra in triggers[1:]: + try: + operator_resume_service.spawn_on_loop(lambda extra=extra: _announce([], extra)) + except Exception: # noqa: BLE001 — the ending is committed; it must stand + logger.warning("[AskService] could not schedule a platform-clear trigger", exc_info=True) + return _ended( + EndingEvent(CANCELLED, tuple(rows), None, reason=reason, batch_id=batch_id), + audit, triggers[0], batch_id=batch_id, + ) + + # --------------------------------------------------------------------------- # The one way an ask begins (PR B): an agent — or, later, a gate — raises it # --------------------------------------------------------------------------- diff --git a/src/backend/services/operator_queue_service.py b/src/backend/services/operator_queue_service.py index 3cae742857..6b896224b7 100644 --- a/src/backend/services/operator_queue_service.py +++ b/src/backend/services/operator_queue_service.py @@ -244,6 +244,14 @@ def _floored_env_cap(name: str, default: int, floor: int) -> int: # `~/.trinity/operator-queue.json`. Correct: the sweep alarm is a platform # alarm ABOUT the agent, not a loop the agent opened and is waiting on. "gitignore-untracked-", + # #3246: four platform families that were never reserved. Reserving them + # also makes `is_platform_minted` true for their rows, which keeps them out + # of the agent's own file and its answer wake — intended: each is an alarm + # ABOUT the agent (or the install), not a loop the agent opened. + "skills-reconcile-", # skill_service reconcile refusal (ent#236) + "skills-fleet-reinject-", # skills_sync_service fleet re-inject failures (ent#236) + "retention-guard-", # retention_guard prune refusal (#1644) + "ent615-git-token-scrub-", # git_service/token_scrub refusal + unreadable (ent#615) ROLE_DRIFT_ALERT_PREFIX, # role-assignment drift (trinity-enterprise#500) — # the role file lives in the AGENT'S OWN workspace, so # an unreserved prefix would let it pre-create the id diff --git a/src/backend/services/operator_resume_service.py b/src/backend/services/operator_resume_service.py index 096a4cb055..52d8b1349a 100644 --- a/src/backend/services/operator_resume_service.py +++ b/src/backend/services/operator_resume_service.py @@ -546,6 +546,39 @@ def _schedule_on_loop(coro_factory: Callable[[], Awaitable[Any]]) -> None: task.add_done_callback(_inflight.discard) +# The loop that hosts the app — captured at lifespan start (`main.py`) and +# refreshed by every on-loop spawn, so a thread anyio does NOT own can still hop +# back to it (#3246). Per process, like `_inflight`; a worker never shares it. +_host_loop: Optional[asyncio.AbstractEventLoop] = None + + +_CAPTURE_RUNNING = object() + + +def remember_host_loop(loop: Any = _CAPTURE_RUNNING) -> None: + """Record the loop background work hops onto. With no argument the RUNNING + loop is captured (nothing changes when called off the loop thread); + ``None`` forgets it.""" + global _host_loop + if loop is _CAPTURE_RUNNING: + try: + loop = asyncio.get_running_loop() + except RuntimeError: + return + _host_loop = loop + + +def _live_host_loop() -> Optional[asyncio.AbstractEventLoop]: + """The captured loop only while it can actually run the callback: a closed + loop raises on `call_soon_threadsafe`, and one that is not running (a test's + private loop between `run_until_complete` calls, a loop that already exited) + would queue the task forever — the `different loop` footgun.""" + loop = _host_loop + if loop is None or loop.is_closed() or not loop.is_running(): + return None + return loop + + def spawn_on_loop(coro_factory: Callable[[], Awaitable[Any]]) -> None: """Run ``coro_factory()`` as a background task on the event loop. @@ -554,9 +587,10 @@ def spawn_on_loop(coro_factory: Callable[[], Awaitable[Any]]) -> None: broadcast. A factory, not a coroutine: the coroutine object has to be created ON the loop thread that awaits it. - WORKS FROM BOTH CALLER SHAPES, and that is not defensive padding — it is the - ent#430 defect. `asyncio.create_task` needs a RUNNING loop, and gets one only - when the caller is `async def`. The operator route is; the Workspace ask route + WORKS FROM EVERY CALLER SHAPE (the loop thread, an anyio worker thread, and + since #3246 any other thread while the host loop runs), and that is not + defensive padding — it is the ent#430 defect. `asyncio.create_task` needs a + RUNNING loop, and gets one only when the caller is `async def`. The operator route is; the Workspace ask route is a plain `def`, which FastAPI runs through `run_in_threadpool` — a worker thread with no loop — so `create_task` raised `RuntimeError: no running event loop`, the caller's `except` swallowed it, and every client answer recorded @@ -580,17 +614,31 @@ def spawn_on_loop(coro_factory: Callable[[], Awaitable[Any]]) -> None: try: _run_sync_in_loop(_schedule_on_loop, coro_factory) + return except RuntimeError as e: - # Reached only from a thread anyio does not own — not a shape any - # production caller has (Starlette's threadpool IS anyio's), but it - # must not degrade into the silent no-op this whole fix removes. - # Re-raised with the cause named so the caller's `except` logs - # something actionable and reports `resume_requested: false`. - raise RuntimeError( - "background work could not be scheduled: called from a thread " - f"with neither a running loop nor an anyio portal ({e})" - ) from e + # A thread anyio does not own. Since #3246 that IS a production + # shape: `asyncio.to_thread` runs on the loop's DEFAULT executor, + # and the headroom sweep (`subscription_recovery_service`) and the + # skills reconcile (`skills_sync_service`, `routers/skills.py`) end + # platform alerts from exactly there — the ending committed with no + # audit row and no trigger, swallowed into a WARNING by + # `platform_alerts._spawn` / `ask_service._ended`. The host loop is + # captured at lifespan start and by every on-loop spawn, so hop to + # it with `call_soon_threadsafe`: the task is created ON the loop + # thread, as `_schedule_on_loop` requires, and the caller does not + # wait for the work. + loop = _live_host_loop() + if loop is None: + # Nothing to hop to. Must not degrade into the silent no-op this + # whole fix removes — re-raised with the cause named so the + # caller's `except` logs something actionable. + raise RuntimeError( + "background work could not be scheduled: called from a thread " + f"with neither a running loop, an anyio portal, nor a live host loop ({e})" + ) from e + loop.call_soon_threadsafe(_schedule_on_loop, coro_factory) return + remember_host_loop() _schedule_on_loop(coro_factory) diff --git a/src/backend/services/platform_alerts.py b/src/backend/services/platform_alerts.py new file mode 100644 index 0000000000..dfbfac92bd --- /dev/null +++ b/src/backend/services/platform_alerts.py @@ -0,0 +1,689 @@ +"""Platform alerts: one pending row per subject (#3246). + +A platform-raised alert describes a CONDITION ("this subscription is near its +limit", "this agent's circuit is dormant"). The operator queue used to store +each reading of it as a separate MESSAGE, so a new reading filed a new pending +row and nothing ended the old ones. This module is the one place that knows +what a platform alert is about: its SUBJECT, `":"`, which the +database can see and which carries at most one pending row. + +This file is the **leaf**: the kind registry, the key belt, the derivation of a +subject from a pre-#3246 row's `request_id` (+ context), and the pure upgrade +sweep planner. It is **stdlib-only at module level** — both migration tracks +(`db/migrations.py`, loaded by file path in `test_schema_parity.py`, and the +Alembic revision) import it, and so does `operator_queue_service`, which +`ask_service` imports. A module-level import of the service graph here would +close that ring (#3246 plan, E1/E2). Anything that needs the database imports +it inside the function. + +Lifetimes (T4): a kind declares one of three classes — + * ``LIFETIME_DEFAULT`` — 14 days from the last reading, env + ``OPERATOR_PLATFORM_ALERT_LIFETIME_DAYS``; + * ``LIFETIME_NET`` — a fixed 30 days, for edge-triggered kinds that have a + clear hook, where expiry is only a net against a clear lost to a crash; + * ``None`` — a person must act; the row never expires on its own. + +Snooze (T5, operator ruling 2026-10-05): after a person ends an alert, a new +reading of the same subject files nothing for 7 days (env +``OPERATOR_PLATFORM_ALERT_SNOOZE_DAYS``) unless it raises the priority or +changes one of the kind's ``material_keys``. The window is read here so the +sweep and the seam agree on it. +""" +import hashlib +import json +import logging +import os +import re +from dataclasses import dataclass, field +from datetime import datetime, timedelta, timezone +from typing import Any, Callable, Dict, Iterable, Mapping, Optional, Tuple + +logger = logging.getLogger(__name__) + +# --------------------------------------------------------------------------- +# Lifetime and snooze +# --------------------------------------------------------------------------- + +LIFETIME_DEFAULT = "default" +LIFETIME_NET = "net" +_LIFETIME_CLASSES = (LIFETIME_DEFAULT, LIFETIME_NET, None) + +DEFAULT_LIFETIME_DAYS = 14 +LIFETIME_ENV = "OPERATOR_PLATFORM_ALERT_LIFETIME_DAYS" +NET_LIFETIME_DAYS = 30 + +DEFAULT_SNOOZE_DAYS = 7 +SNOOZE_ENV = "OPERATOR_PLATFORM_ALERT_SNOOZE_DAYS" + + +def _positive_int_env(name: str, default: int) -> int: + raw = os.getenv(name) + if raw is None: + return default + try: + value = int(raw.strip()) + except (TypeError, ValueError): + value = 0 + if value < 1: + logger.warning( + "[platform_alerts] %s=%r is not a positive whole number of days; " + "using the default %d", name, raw, default, + ) + return default + return value + + +def default_lifetime_days() -> int: + """The default lifetime in days (env knob; bad value → default + WARNING).""" + return _positive_int_env(LIFETIME_ENV, DEFAULT_LIFETIME_DAYS) + + +def snooze_window() -> timedelta: + """How long a person's ending of an alert silences new readings of it.""" + return timedelta(days=_positive_int_env(SNOOZE_ENV, DEFAULT_SNOOZE_DAYS)) + + +# --------------------------------------------------------------------------- +# Registry +# --------------------------------------------------------------------------- + +CLEARS_HOOK = "hook" # the emitter ends the row when the condition clears +CLEARS_LIFETIME_ONLY = "lifetime_only" # nothing observes the clear; the lifetime ends it +CLEARS_PERSON = "person" # a person must act on it +_CLEARS = (CLEARS_HOOK, CLEARS_LIFETIME_ONLY, CLEARS_PERSON) + + +@dataclass(frozen=True) +class Kind: + """One kind of platform alert. + + ``prefix`` is the request-id prefix its rows carry; it must fall under a + platform-reserved prefix (`operator_queue_service._RESERVED_ID_PREFIXES`) + so an agent cannot mint a row into it. Kinds may share a prefix (two + conditions under one family) — they then share a lifetime class. ``event`` + kinds have no subject: every observation is its own row. + """ + + name: str + prefix: str + lifetime: Optional[str] + clears: str + budgeted_type: Optional[str] = None + material_keys: Tuple[str, ...] = () + event: bool = False + + def __post_init__(self): + if not re.fullmatch(r"[a-z][a-z0-9_]*", self.name or ""): + raise ValueError(f"kind name must be lower_snake: {self.name!r}") + if not self.prefix or not self.prefix.endswith(("-", "_")): + raise ValueError(f"kind {self.name}: prefix must end in '-' or '_'") + if self.lifetime not in _LIFETIME_CLASSES: + raise ValueError(f"kind {self.name}: unknown lifetime class {self.lifetime!r}") + if self.clears not in _CLEARS: + raise ValueError(f"kind {self.name}: unknown clears {self.clears!r}") + + +# Prefixes whose emitter lives outside this repository (the enterprise +# submodule). The sweep never touches them; the emitter registers its kind +# through `register(...)` from its own package. +EXTERNAL_PREFIXES = frozenset({"role-drift-"}) + +_D, _N = LIFETIME_DEFAULT, LIFETIME_NET + +_BUILTIN_KINDS = ( + # -- the four families #3246 names ------------------------------------ + Kind("subscription_headroom", "sub-headroom-", _D, CLEARS_HOOK, material_keys=("tier",)), + Kind("skills_legacy_adoption", "skills-legacy-adoption-", _D, CLEARS_HOOK), + Kind("base_image_stale", "base-image-stale-", _N, CLEARS_HOOK), + Kind("system_agent_start_failed", "base-image-stale-start-", _N, CLEARS_HOOK), + Kind("circuit_dormant", "cb-dormant-", _N, CLEARS_HOOK), + # -- every other in-repo platform emitter ----------------------------- + Kind("skills_reconcile_refused", "skills-reconcile-", _N, CLEARS_HOOK), + Kind("skills_fleet_reinject", "skills-fleet-reinject-", _D, CLEARS_HOOK), + Kind("sync_failing", "sync-failing-", _N, CLEARS_HOOK), + Kind("sync_diverged", "sync-diverged-", _N, CLEARS_HOOK), + Kind("git_bloat_size", "git-bloat-", _N, CLEARS_HOOK), + Kind("git_bloat_maintenance", "git-bloat-", _N, CLEARS_HOOK), + Kind("db_backup_failure", "db-backup-", _D, CLEARS_HOOK), + Kind("db_backup_stale", "db-backup-", _D, CLEARS_HOOK), + Kind("log_archive_unwritable", "log-archive-", _D, CLEARS_HOOK), + Kind("retention_refused", "retention-guard-", _D, CLEARS_HOOK, material_keys=("reason",)), + Kind("system_seed", "system-seed-", _N, CLEARS_HOOK), + Kind("git_token_scrub_refused", "ent615-git-token-scrub-", _D, CLEARS_HOOK), + Kind("git_token_scrub_unreadable", "ent615-git-token-scrub-unreadable-", _D, CLEARS_HOOK), + Kind("queue_flood", "queue-flood-", _D, CLEARS_HOOK, budgeted_type="queue_flood"), + Kind("alert_budget", "alert-budget-", _D, CLEARS_HOOK), + Kind("skill_not_found", "skill-not-found-", _D, CLEARS_HOOK, budgeted_type="skill_not_found"), + Kind("effect_unguarded", "effect-unguarded-", _D, CLEARS_LIFETIME_ONLY, + budgeted_type="effect_unguarded"), + Kind("validation_failed", "val_", _D, CLEARS_LIFETIME_ONLY), + # -- a person must act: no lifetime ---------------------------------- + Kind("poison", "poison-", None, CLEARS_PERSON), + Kind("portal_inbox_collision", "portal-inbox-collision-", None, CLEARS_PERSON), + Kind("gitignore_untracked", "gitignore-untracked-", None, CLEARS_PERSON, + budgeted_type="gitignore_untracked", event=True), + Kind("workspace_problem_report", "workspace-problem-", None, CLEARS_PERSON, + budgeted_type="workspace_problem_report", event=True), +) + +KINDS: Dict[str, Kind] = {k.name: k for k in _BUILTIN_KINDS} + + +def register(kind: Kind) -> None: + """Register a kind defined outside this module (the enterprise emitters). + + A duplicate name is refused: two emitters silently sharing a kind would + share subjects, and one's clear would end the other's alert. + """ + if kind.name in KINDS: + raise ValueError(f"platform alert kind {kind.name!r} is already registered") + KINDS[kind.name] = kind + + +def _kind(name: str) -> Kind: + try: + return KINDS[name] + except KeyError: + raise KeyError(f"unregistered platform alert kind {name!r}") from None + + +def lifetime_for(kind_name: str) -> Optional[timedelta]: + """The kind's lifetime, measured from its last reading; None = never expires.""" + lifetime = _kind(kind_name).lifetime + if lifetime is None: + return None + if lifetime == LIFETIME_NET: + return timedelta(days=NET_LIFETIME_DAYS) + return timedelta(days=default_lifetime_days()) + + +# --------------------------------------------------------------------------- +# Key belt / subject +# --------------------------------------------------------------------------- + +_KEY_RE = re.compile(r"^[A-Za-z0-9._:-]{1,128}$") + + +def _belt_key(key: Any) -> str: + if isinstance(key, bool) or not isinstance(key, (str, int)): + raise ValueError(f"platform alert key must be a string, got {type(key).__name__}") + text = str(key).strip() + if not text: + raise ValueError("platform alert key is empty") + if _KEY_RE.match(text): + return text + return hashlib.sha256(text.encode("utf-8")).hexdigest()[:16] + + +def subject_for(kind_name: str, key: Any) -> Optional[str]: + """``":"`` with the key belted: id-shaped and ≤128 chars stays, + anything else becomes ``sha256(key)[:16]``. An event kind has no subject + (None); a condition kind without a usable key raises ValueError.""" + kind = _kind(kind_name) + if kind.event: + return None + return f"{kind.name}:{_belt_key(key)}" + + +# --------------------------------------------------------------------------- +# Legacy subject derivation (the upgrade sweep's one source of id parsing) +# --------------------------------------------------------------------------- + +@dataclass(frozen=True) +class LegacyMatch: + """A pre-#3246 row read back: its kind, and its subject when the id (+ one + named context key) identifies the condition. ``subject is None`` means + "this kind, but the row does not say which condition" — never guessed.""" + + kind: str + subject: Optional[str] + + +_TS = r"\d{4}-\d{2}-\d{2}T[0-9:.]+(?:Z|[+-]\d{2}:?\d{2})?" +_DAY = r"\d{4}-\d{2}-\d{2}" + + +def _rx(pattern: str) -> Callable[[str], Optional[str]]: + compiled = re.compile(pattern) + + def parse(rest: str, _ctx: Mapping) -> Optional[str]: + m = compiled.fullmatch(rest) + return m.group(1) if m else None + + return parse + + +def _ctx_str(ctx: Mapping, name: str) -> Optional[str]: + value = ctx.get(name) + return value if isinstance(value, str) and value.strip() else None + + +def _headroom(rest: str, _ctx: Mapping) -> Optional[str]: + # sub-headroom-{safe_sid}-{episode}-{tier} | sub-headroom-fleet-{episode}; + # episode is a day or `unknown-`. Non-greedy sid so `unknown-` is + # read as the episode, not swallowed into a sid. + m = re.fullmatch(rf"(.+?)-(?:unknown-)?{_DAY}-(?:warn|crit)", rest) + if m: + return m.group(1) + if re.fullmatch(rf"fleet-(?:unknown-)?{_DAY}", rest): + return "fleet" + return None + + +def _legacy_adoption(rest: str, ctx: Mapping) -> Optional[str]: + # Steady state carries sha256(raw url)[:12]; the failure branch is + # timestamped and carries only the SCRUBBED url in context. For a URL with + # no embedded credentials the two hashes agree; for one with credentials + # the failure rows land on a second subject — two survivors, never a wrong + # merge. + m = re.fullmatch(r"refused-([0-9a-f]{12})", rest) + if m: + return m.group(1) + url = _ctx_str(ctx, "url") + if url and re.fullmatch(_TS, rest): + return hashlib.sha256(url.strip().encode("utf-8")).hexdigest()[:12] + return None + + +def _effect_unguarded(_rest: str, ctx: Mapping) -> Optional[str]: + effect, reason = _ctx_str(ctx, "effect_type"), _ctx_str(ctx, "reason") + return f"{effect}:{reason}" if effect and reason else None + + +def _alert_budget(rest: str, ctx: Mapping) -> Optional[str]: + return _ctx_str(ctx, "alert_type") or _rx(r".+-([a-z0-9_]+)-b\d+")(rest, ctx) + + +def _context_key(name: str) -> Callable[[str, Mapping], Optional[str]]: + return lambda _rest, ctx: _ctx_str(ctx, name) + + +def _constant(value: str) -> Callable[[str, Mapping], Optional[str]]: + return lambda _rest, _ctx: value + + +def _whole(rest: str, _ctx: Mapping) -> Optional[str]: + return rest or None + + +def _no_subject(_rest: str, _ctx: Mapping) -> Optional[str]: + return None + + +def _db_backup(rest: str, ctx: Mapping) -> Tuple[str, Optional[str]]: + alert_type = ctx.get("alert_type") + if alert_type in ("db_backup_failure", "db_backup_stale"): + return alert_type, _ctx_str(ctx, "backup_dir") + return "db_backup_failure", None + + +# (id prefix, kind, key parser) — matched longest prefix first. A parser +# returns the natural key or None. `db-backup-` picks its kind from context. +_LEGACY_PARSERS: Tuple[Tuple[str, str, Callable], ...] = ( + ("sub-headroom-", "subscription_headroom", _headroom), + ("skills-legacy-adoption-", "skills_legacy_adoption", _legacy_adoption), + ("base-image-stale-start-", "system_agent_start_failed", _rx(r"(.+)-\d+")), + ("base-image-stale-", "base_image_stale", _rx(rf"(.+)-{_TS}")), + ("cb-dormant-", "circuit_dormant", _rx(rf"(.+)-{_TS}")), + ("skills-reconcile-", "skills_reconcile_refused", _rx(r"(.+)-\d+")), + ("skills-fleet-reinject-", "skills_fleet_reinject", _constant("fleet")), + ("sync-failing-", "sync_failing", _rx(rf"(.+)-{_TS}")), + ("sync-diverged-", "sync_diverged", _rx(rf"(.+)-{_TS}")), + # Two conditions (size, maintenance) share this prefix and are told apart + # only in prose: kind known for the lifetime, subject never guessed. + ("git-bloat-", "git_bloat_size", _no_subject), + ("db-backup-", "db_backup_failure", _db_backup), + ("log-archive-", "log_archive_unwritable", _context_key("path")), + ("retention-guard-", "retention_refused", _rx(r"(.+)-\d+")), + ("system-seed-", "system_seed", _whole), + ("ent615-git-token-scrub-unreadable-", "git_token_scrub_unreadable", _rx(rf"(.+)-{_DAY}")), + ("ent615-git-token-scrub-", "git_token_scrub_refused", _rx(rf"(.+)-{_DAY}")), + ("queue-flood-", "queue_flood", _context_key("reason")), + ("alert-budget-", "alert_budget", _alert_budget), + ("skill-not-found-", "skill_not_found", _context_key("command")), + ("effect-unguarded-", "effect_unguarded", _effect_unguarded), + ("val_", "validation_failed", _rx(r"(.+)_\d{14}")), + ("poison-", "poison", _whole), + ("portal-inbox-collision-", "portal_inbox_collision", _whole), + ("gitignore-untracked-", "gitignore_untracked", _no_subject), + ("workspace-problem-", "workspace_problem_report", _no_subject), +) +_LEGACY_PARSERS = tuple(sorted(_LEGACY_PARSERS, key=lambda p: -len(p[0]))) + + +def _as_context(context: Any) -> Dict: + if isinstance(context, dict): + return context + if isinstance(context, (str, bytes)) and context: + try: + parsed = json.loads(context) + except (TypeError, ValueError): + return {} + return parsed if isinstance(parsed, dict) else {} + return {} + + +def derive_legacy_subject(request_id: Any, context: Any) -> Optional[LegacyMatch]: + """Read a pre-#3246 platform row back into (kind, subject). + + None when the prefix is not an in-repo platform kind (an agent's own ask, + a gate row, an enterprise prefix) — the sweep must leave those alone. + """ + if not isinstance(request_id, str) or not request_id: + return None + ctx = _as_context(context) + for prefix, kind_name, parse in _LEGACY_PARSERS: + if not request_id.startswith(prefix) or kind_name not in KINDS: + continue + rest = request_id[len(prefix):] + result = parse(rest, ctx) + if isinstance(result, tuple): + kind_name, key = result + else: + key = result + subject = None + if key is not None: + try: + subject = subject_for(kind_name, key) + except ValueError: + subject = None + return LegacyMatch(kind_name, subject) + return None + + +# --------------------------------------------------------------------------- +# Upgrade sweep planner (pure) +# --------------------------------------------------------------------------- + +@dataclass(frozen=True) +class Stamp: + """One row write the sweep plans. A field left None is not written.""" + + id: Any + subject: Optional[str] = None + last_seen_at: Optional[str] = None + expires_at: Optional[str] = None + context: Optional[Dict] = None + + +@dataclass(frozen=True) +class SweepPlan: + # survivors: subject + last_seen_at + context.seen_count, expires_at when it was NULL + survivor_stamps: Tuple[Stamp, ...] = () + # pending duplicates to end as one batch: cancelled / platform / superseded + ended_ids: Tuple[Any, ...] = () + # known kind, no derivable subject: expires_at only (so it leaves on its own) + lifetime_stamps: Tuple[Stamp, ...] = () + # person-ended rows inside the snooze window: subject only (read by the seam) + snooze_stamps: Tuple[Stamp, ...] = field(default=()) + + def is_empty(self) -> bool: + return not (self.survivor_stamps or self.ended_ids + or self.lifetime_stamps or self.snooze_stamps) + + +def _iso_z(dt: datetime) -> str: + # Same shape as utils.helpers.utc_now_iso (Invariant #16). + return dt.astimezone(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ") + + +def _parse_iso(value: Any) -> Optional[datetime]: + if not isinstance(value, str) or not value: + return None + try: + dt = datetime.fromisoformat(value.replace("Z", "+00:00")) + except ValueError: + return None + return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) + + +def _age_key(row: Mapping) -> Tuple: + # Newest wins: created_at, then id (the #3130 rule); NULL created_at is oldest. + created = row.get("created_at") + return (created is not None, created or "", str(row.get("id"))) + + +def plan_sweep(rows: Iterable[Mapping], *, now: datetime) -> SweepPlan: + """Plan the one-shot upgrade sweep over fetched queue rows. + + ``rows``: mappings with ``id, agent_name, request_id, status, created_at, + expires_at, context, subject, disposed_by, disposed_at, raised_by``. Rules: + + * Only platform-raised rows (``raised_by`` NULL) whose request id derives + to an in-repo kind are considered; agent-raised and gate rows, unknown + and external prefixes are never touched. + * Pending rows are grouped by ``(agent_name, subject)``. A row that already + carries a subject holds the slot (it is the seam's live row); otherwise + the newest by ``created_at`` then ``id`` survives. Every other pending + row of the group is ended. + * Survivors are stamped with ``subject``, ``last_seen_at = created_at`` + (``now`` when NULL), ``context.seen_count = 1`` and — only when + ``expires_at`` is NULL and the kind has a lifetime — ``now + lifetime``. + * Pending rows of a known kind with no derivable subject get the lifetime + stamp only. + * Person-ended rows whose ending falls inside the snooze window get + ``subject`` only, so the seam's snooze can see them. + * Planning over the plan's own result yields an empty plan. + """ + if now.tzinfo is None: + raise ValueError("plan_sweep needs an aware `now`") + snooze_since = now - snooze_window() + + groups: Dict[Tuple[Any, str], list] = {} + lifetime_stamps, snooze_stamps = [], [] + for row in rows: + if row.get("raised_by") is not None: + # An agent's own ask or a gate row: never the platform's, whatever + # its id looks like (the prefix was unreserved when it was minted). + continue + match = derive_legacy_subject(row.get("request_id"), row.get("context")) + if match is None: + continue + stored_subject = row.get("subject") + status = row.get("status") + if status == "pending": + subject = stored_subject or match.subject + if subject is not None: + groups.setdefault((row.get("agent_name"), subject), []).append(row) + continue + lifetime = lifetime_for(match.kind) + if lifetime is not None and row.get("expires_at") is None: + lifetime_stamps.append(Stamp(id=row.get("id"), expires_at=_iso_z(now + lifetime))) + continue + if (stored_subject is None and match.subject is not None + and row.get("disposed_by") == "person"): + disposed = _parse_iso(row.get("disposed_at")) + if disposed is not None and disposed >= snooze_since: + snooze_stamps.append(Stamp(id=row.get("id"), subject=match.subject)) + + survivor_stamps, ended_ids = [], [] + for (_agent, subject), members in groups.items(): + holders = [r for r in members if r.get("subject")] + survivor = max(holders or members, key=_age_key) + ended_ids.extend(r.get("id") for r in members if r is not survivor) + if survivor.get("subject"): + continue + match = derive_legacy_subject(survivor.get("request_id"), survivor.get("context")) + lifetime = lifetime_for(match.kind) + expires_at = None + if lifetime is not None and survivor.get("expires_at") is None: + expires_at = _iso_z(now + lifetime) + context = dict(_as_context(survivor.get("context"))) + context["seen_count"] = 1 + survivor_stamps.append(Stamp( + id=survivor.get("id"), + subject=subject, + last_seen_at=survivor.get("created_at") or _iso_z(now), + expires_at=expires_at, + context=context, + )) + + return SweepPlan( + survivor_stamps=tuple(survivor_stamps), + ended_ids=tuple(ended_ids), + lifetime_stamps=tuple(lifetime_stamps), + snooze_stamps=tuple(snooze_stamps), + ) + + +# --------------------------------------------------------------------------- +# The seam: observe / clear / reconcile (#3246 C4) +# +# Every import below is function-local: this module stays a stdlib-only leaf +# at import time (the migration tracks import `plan_sweep` from it, and +# `operator_queue_service` calls back into it — a module-level import of the +# service graph would close the cycle E1 names). +# --------------------------------------------------------------------------- + +OBSERVED_CREATED = "created" +OBSERVED_UPDATED = "updated" +OBSERVED_SNOOZED = "snoozed" +OBSERVED_REFUSED_UNREGISTERED = "refused_unregistered" +OBSERVED_REFUSED_AT_BUDGET = "refused_at_budget" +OBSERVED_FAILED = "failed" + +_PRIORITY_RANK = {"low": 0, "medium": 1, "high": 2, "critical": 3} + + +def _snoozed_by(kind: Kind, ended: Optional[Mapping], priority: str, context: Mapping) -> bool: + """A person ended this subject's row inside the snooze window, and the new + reading is neither a priority increase nor a material change.""" + if not ended: + return False + if _PRIORITY_RANK.get(priority, 0) > _PRIORITY_RANK.get(ended.get("priority"), 0): + return False + old = _as_context(ended.get("context")) + return all(context.get(k) == old.get(k) for k in kind.material_keys) + + +def _spawn(factory: Callable[[], Any]) -> None: + try: + from services import operator_resume_service + operator_resume_service.spawn_on_loop(factory) + except Exception: # noqa: BLE001 — the row is committed; a lost trigger is not fatal + logger.warning("[platform-alerts] could not schedule a follow-up", exc_info=True) + + +async def _broadcast_sync(agent_name: str) -> None: + """Thin trigger (#918: identifiers only); listeners refetch the access-controlled list. + + Keyed by the row's host agent, like the create path's `operator_queue_new`, + so the scope filter delivers it only to clients who may see that agent's + queue (ent#467). The poller's fleet-level `operator_queue_sync` is one + trigger per cycle spanning agents; this one is always about one row. + """ + from services import operator_queue_service as oqs + manager = oqs._websocket_manager + if manager: + await manager.broadcast(json.dumps({ + "type": "operator_queue_sync", + "data": {"agent_name": agent_name}, + })) + + +def observe( + agent_name: str, + kind: str, + key: Any = None, + *, + title: str, + question: str, + priority: str = "high", + context: Optional[Mapping] = None, +) -> str: + """Report a new reading of a platform condition (#3246). Never raises. + + The subject's pending row is updated in place (one row per subject, the + latest reading); with no pending row, a reading a person ended inside the + snooze window (`OPERATOR_PLATFORM_ALERT_SNOOZE_DAYS`, default 7) files + nothing unless its priority rose or a `material_keys` value changed; else + a fresh row is filed with the kind's lifetime as `expires_at`, measured + from this reading. Returns one of the ``OBSERVED_*`` outcomes. + """ + try: + k = KINDS.get(kind) + if k is None: + logger.error("[platform-alerts] refusing an alert of unregistered kind %r for %s", + kind, agent_name) + return OBSERVED_REFUSED_UNREGISTERED + from database import db + from utils.helpers import utc_now_iso + + subject = subject_for(kind, key) + ctx = dict(context) if isinstance(context, Mapping) else {} + now = datetime.now(timezone.utc) + if subject is not None and db.find_pending_operator_queue_by_subject(agent_name, subject) is None: + since = _iso_z(now - snooze_window()) + ended = db.find_person_ended_operator_queue_by_subject(agent_name, subject, since) + if _snoozed_by(k, ended, priority, ctx): + return OBSERVED_SNOOZED + lifetime = lifetime_for(kind) + stem = _belt_key(key) + "-" if subject is not None else "" + item = { + "id": f"{k.prefix}{stem}{utc_now_iso()}", + "type": k.budgeted_type or "alert", + "status": "pending", + "priority": priority, + "title": title, + "question": question, + "context": ctx, + "created_at": utc_now_iso(), + "expires_at": _iso_z(now + lifetime) if lifetime is not None else None, + } + cap = None + if k.budgeted_type: + from services import operator_queue_service as oqs + cap = oqs.OPERATOR_ALERT_MAX_PENDING_PER_TYPE + out = db.create_platform_operator_queue_item( + agent_name, item, subject=subject, max_pending_for_type=cap) + if out["outcome"] == OBSERVED_REFUSED_AT_BUDGET: + from services import operator_queue_service as oqs + _spawn(lambda: oqs._maybe_emit_alert_budget_episode( + agent_name, k.budgeted_type, ctx.get("triggered_by"))) + return OBSERVED_REFUSED_AT_BUDGET + if out.get("changed"): + _spawn(lambda: _broadcast_sync(agent_name)) + return out["outcome"] + except Exception: # noqa: BLE001 — an alert must never break its emitter + logger.error("[platform-alerts] observe failed for %s/%s", agent_name, kind, exc_info=True) + return OBSERVED_FAILED + + +def _end(rows: Iterable[Mapping], reason: Optional[str]) -> int: + ids = [r["id"] for r in rows if r] + if not ids: + return 0 + from services import ask_service + return len(ask_service.clear_platform(ids, reason=reason or ask_service.CONDITION_CLEARED).rows) + + +def clear(agent_name: str, kind: str, key: Any = None, *, reason: Optional[str] = None) -> int: + """The condition behind `(agent, kind, key)` cleared: the platform ends its + pending row (ent#611 ledger, `disposed_by = 'platform'`). A row a person + ended first is untouched. Returns the number ended; never raises.""" + try: + subject = subject_for(kind, key) + if subject is None: + return 0 + from database import db + return _end([db.find_pending_operator_queue_by_subject(agent_name, subject)], reason) + except Exception: # noqa: BLE001 + logger.error("[platform-alerts] clear failed for %s/%s", agent_name, kind, exc_info=True) + return 0 + + +def reconcile(agent_name: str, kind: str, live_keys: Iterable[Any]) -> int: + """End every pending row of `kind` for `agent_name` whose key is not in + `live_keys` — for an emitter that knows the full current set rather than + each clear. Returns the number ended; never raises.""" + try: + _kind(kind) + live = {subject_for(kind, key) for key in live_keys} + from database import db + rows = db.list_operator_queue_items(agent_name=agent_name, status="pending", limit=1000) + stale = [r for r in rows + if (r.get("subject") or "").startswith(f"{kind}:") and r["subject"] not in live] + return _end(stale, None) + except Exception: # noqa: BLE001 + logger.error("[platform-alerts] reconcile failed for %s/%s", agent_name, kind, exc_info=True) + return 0 diff --git a/src/backend/services/skill_service.py b/src/backend/services/skill_service.py index c56b397af5..91ff2e7618 100644 --- a/src/backend/services/skill_service.py +++ b/src/backend/services/skill_service.py @@ -797,6 +797,9 @@ def _adopt_legacy_clone(self) -> Optional[str]: url = get_skills_library_url() except Exception: # noqa: BLE001 return None + # #3246: a refusal row is live only while the setting still names its + # URL — a removed or changed setting ends the rows it no longer backs. + self._reconcile_adoption_alerts(url) if not url: return None @@ -855,6 +858,7 @@ def _adopt_legacy_clone(self) -> Optional[str]: existing = [s for s in db.list_skill_sources() if _same_skills_repo(s.url, normalized_url)] if existing: + self._clear_adoption_alert(url) return existing[0].id if db.count_skill_sources() > 0: @@ -916,9 +920,38 @@ def _adopt_legacy_clone(self) -> Optional[str]: source.id, e, ) + self._clear_adoption_alert(url) logger.info("migrated skills_library_url into skill source %s", source.id) return source.id + # #3246: the legacy adoption refusal is one platform condition per URL. + + @staticmethod + def _adoption_alert_key(url: Any) -> str: + """`sha256(url)[:12]` over the RAW value — never the URL itself (a raw + URL is a conflict key that could carry a PAT). Raises on a non-str + setting value; every caller is inside a try.""" + return hashlib.sha256(url.strip().encode()).hexdigest()[:12] + + def _clear_adoption_alert(self, url: Any) -> None: + """Adoption succeeded (or the URL already names a source): end the + URL's refusal row as the platform. Never raises.""" + from services import platform_alerts + try: + platform_alerts.clear(RECONCILE_ALARM_AGENT_NAME, "skills_legacy_adoption", + self._adoption_alert_key(url)) + except Exception as e: # noqa: BLE001 + logger.warning(f"could not clear skills adoption alarm: {e}") + + def _reconcile_adoption_alerts(self, url: Any) -> None: + """End every refusal row whose URL is no longer the setting. Never raises.""" + from services import platform_alerts + try: + live = [self._adoption_alert_key(url)] if url else [] + platform_alerts.reconcile(RECONCILE_ALARM_AGENT_NAME, "skills_legacy_adoption", live) + except Exception as e: # noqa: BLE001 + logger.warning(f"could not reconcile skills adoption alarms: {e}") + # ------------------------------------------------------------------------- # Durable sync status (ent#236) # ------------------------------------------------------------------------- @@ -2511,43 +2544,32 @@ def _record_adoption_failure( silently is how the refusal gets rediscovered as "skills stopped working". - A NEW CALL SITE MUST CHOOSE `steady_state` EXPLICITLY. The default is - the unbounded timestamped/`high` shape, which is correct only for a - genuine transient failure — nothing enforces that choice (the #1677 AST - guard scans call sites of `create_operator_queue_item`, not of this - method), so the docstring is the enforcement. + A NEW CALL SITE MUST CHOOSE `steady_state` EXPLICITLY — it picks the + severity of the reading. - Two classes of refusal share this emitter (#2744): + Two classes of refusal share this emitter (#2744), and since #3246 one + subject — `skills_legacy_adoption:` on the + `_skills-sync` host — so a URL has at most one pending row: * A genuine FAILURE — the URL failed validation, or adoption threw. - `priority: "high"` and a TIMESTAMPED id, so a repeat is visible AS a - repeat rather than deduped away by `create_item`'s ON CONFLICT DO - NOTHING. Repeat-visible here is a PRODUCT DECISION, not a claim that - these branches are bounded: a permanently invalid `skills_library_url` - re-enters the validation branch on every sync just as forever. + `priority: "high"`, `logger.error`. A repeat is visible AS a repeat + through the row's seen count and last-seen time, not as a new row. * The STEADY STATE — "this install already has sources, so it is past migration": the designed resting state of every migrated install whose - legacy key lingers, not a failure. `priority: "low"`, `logger.info`, - and a STABLE id derived from the refused URL, so the `(agent_name, - request_id)` ON CONFLICT collapses every repeat to ONE row and an - operator's dismissal sticks. The family prefix is registered in - `operator_queue_service._RESERVED_ID_PREFIXES`, because a stable id - derived from an admin-visible URL is guessable and an agent could - otherwise pre-create it and silence the alarm (the #1632 C2 class); - that tuple also drives `is_platform_minted`. - - The id is the sha256 of the URL, never the URL itself: `request_id` is a - conflict key, not a display field, and a raw URL would break the - id-shape (`^[A-Za-z0-9._:-]+$`) the reserved-prefix machinery assumes. - - **The bound this emitter's #1677 platform-only classification rests on** - is: the only input is the `skills_library_url` setting, which - `routers/settings.py` blocks on the generic settings PUT + legacy key lingers, not a failure. `priority: "low"`, `logger.info`. + A person's dismissal holds through the seam's snooze window. + + The kind's id prefix stays in + `operator_queue_service._RESERVED_ID_PREFIXES`, so an agent cannot + pre-create one of these rows (the #1632 C2 class); that tuple also + drives `is_platform_minted`. The key is the sha256 of the URL, never + the URL itself: a raw URL is not id-shaped and could carry a PAT. + + The volume bound: the only input is the `skills_library_url` setting, + which `routers/settings.py` blocks on the generic settings PUT (`LEGACY_SKILLS_LIBRARY_KEYS`, 422) and no other writer reaches — so no - agent can drive the volume. It is NOT "admin-driven cadence": ent#236's - auto-sync calls the same `sync_library()` unattended on a 300s-86400s - timer. Keep this comment and the `_ALLOWED_CALLERS` justification in - `tests/unit/test_1677_operator_alert_emitters.py` in step. + agent can drive it. ent#236's auto-sync calls the same `sync_library()` + unattended on a 300s-86400s timer, which is why one row per URL matters. The echoed URL is SCRUBBED (#2744). Naming what was refused is the whole point of the alarm, but `EmbeddedCredentialError` is a `ValueError` @@ -2564,38 +2586,25 @@ def _record_adoption_failure( else: logger.error(f"[ent#346] {message} (url={safe_url})") try: - from utils.helpers import utc_now_iso - - context = { - "alert_type": "skills_legacy_adoption_refused", - "url": safe_url, - } - if steady_state: - # Computed INSIDE the try: `url.strip().encode()` raises - # AttributeError on a non-str setting value, and out here that - # degrades to a warning and no alarm — the fail-soft guarantee. - # Hoisting it above the try turns a decorative alarm into a - # raiser. - request_id = ( - "skills-legacy-adoption-refused-" - f"{hashlib.sha256(url.strip().encode()).hexdigest()[:12]}" - ) - else: - # Timestamped, so a repeated failure is visible as repeated - # rather than silently deduped by the ON CONFLICT DO NOTHING - # in `create_item`. - request_id = f"skills-legacy-adoption-{utc_now_iso()}" - - db.create_operator_queue_item( - RECONCILE_ALARM_AGENT_NAME, - { - "id": request_id, - "type": "alert", - "priority": "low" if steady_state else "high", - "title": "Legacy skills-library adoption refused", - "question": message, - "context": context, - "expires_at": None, + from services import platform_alerts + + # #3246: one platform condition per URL — `sha256(raw url)[:12]` + # (computed INSIDE the try: a non-str setting value raises here and + # degrades to a warning, the fail-soft guarantee). Both branches + # share the subject, so a repeat — steady-state or a genuine + # failure — updates the one pending row (seen count, latest + # message and priority) instead of filing a second, and the row is + # ended by the platform when adoption succeeds or the setting is + # removed (`_clear_adoption_alert` / `_reconcile_adoption_alerts`). + platform_alerts.observe( + RECONCILE_ALARM_AGENT_NAME, "skills_legacy_adoption", + self._adoption_alert_key(url), + priority="low" if steady_state else "high", + title="Legacy skills-library adoption refused", + question=message, + context={ + "alert_type": "skills_legacy_adoption_refused", + "url": safe_url, }, ) except Exception as e: # noqa: BLE001 — the alarm is decorative diff --git a/src/backend/services/subscription_headroom_alerts.py b/src/backend/services/subscription_headroom_alerts.py index fce3653f8d..3a4ce0c520 100644 --- a/src/backend/services/subscription_headroom_alerts.py +++ b/src/backend/services/subscription_headroom_alerts.py @@ -23,30 +23,19 @@ 1. A hysteresis floor is dead code — utilization does not fall inside a window, so the only real re-arm is the reset. -2. `resets_at` therefore IS the window's identity, and putting it in the alert - id makes the id the entire state machine: - - sub-headroom-{sid}-{reset-day}-{tier} - - `db.create_operator_queue_item` maps `item["id"]` onto `request_id`, which - is `UNIQUE(agent_name, request_id)` with ON CONFLICT DO NOTHING. So the - same window re-emits into the same row (no duplicate), a reset mints a new - id (re-armed), and the escalation carries its own tier suffix. Cross-worker - and cross-restart dedup fall out for free, with no lock and no memo. - -The id is quantised to the DAY rather than the exact instant. Under the -measured fixed-window semantics that is exactly one episode per window; if a -different provider plan ever did behave as rolling, it degrades to at most one -alert per day instead of one per probe. The cheap belt is worth more than the -precision. - -## Residual, stated rather than hidden - -`create_item` has no UPDATE path, so a warning row keeps the number it was -raised with — a 75% alert still reads 75% when the subscription later sits at -92%. Same residual `retention_guard` documents for its own alarm. The -escalation is a SEPARATE id with a self-contained body, so the newer figure -does arrive; it just arrives as a second item rather than an edit. +2. `resets_at` therefore IS the window's identity. It travels on the row's + context as `window` (quantised to the day — a provider that did behave as + rolling degrades to one window per day). + +## One row per subscription (#3246) + +Each reading is reported through `services.platform_alerts.observe` with the +subject `subscription_headroom:` (or `:fleet`), so a subscription +has at most one pending row: a later reading updates it in place, a critical +reading replaces the warning (`tier` is the material key the snooze compares), +and the evaluation pass ends the row (`clear_recovered`) once the subscription +is measured back under its threshold — which, under the fixed window, is the +reset. Cross-worker dedup is the seam's partial unique index on the subject. """ import logging @@ -64,7 +53,6 @@ SATURATED, UNASSESSABLE, ) -from utils.helpers import utc_now_iso logger = logging.getLogger(__name__) @@ -239,13 +227,14 @@ def episode_key(resets_at: Optional[str], *, now: Optional[datetime] = None) -> return f"unknown-{current.date().isoformat()}" -def alert_id(subscription_id: str, episode: str, tier: str) -> str: - safe_sid = _SID_SAFE.sub("-", str(subscription_id))[:64] - return f"{ALARM_ID_PREFIX}{safe_sid}-{episode}-{tier}" +# #3246: the subject key of the fleet-wide alert. +FLEET_KEY = "fleet" -def fleet_alert_id(episode: str) -> str: - return f"{ALARM_ID_PREFIX}fleet-{episode}" +def subject_key(subscription_id: str) -> str: + """The cleaned subscription id the legacy `sub-headroom-{sid}-…` ids + carried, so a row the upgrade sweep re-subjected matches a new reading.""" + return _SID_SAFE.sub("-", str(subscription_id))[:64] def decide_tier( @@ -331,42 +320,33 @@ def _fmt_reset(resets_at: Optional[str]) -> str: return dt.strftime("%Y-%m-%d %H:%M UTC") if dt else "reset unknown" -def _emit(item_id: str, *, title: str, question: str, priority: str, +def _emit(key: str, *, title: str, question: str, priority: str, context: Dict[str, Any]) -> bool: - """One platform create on the sentinel host. Never raises. - - Platform-only by construction: an agent cannot drive the VOLUME. The - cadence is the sweep's, the id is deterministic per (subscription, window, - tier) so a re-emit is an on-conflict no-op, and the per-cycle cap bounds a - whole fleet crossing at once. That is what makes it a direct create with an - allowlist entry in `tests/unit/test_1677_operator_alert_emitters.py` rather - than a `create_bounded_alert` caller. - - Note the precise claim: agent-chosen *names* do reach the body (an agent may - spawn children and name them), but they arrive sanitized, are capped at five - per alert, and already appear on every operator surface. It is the volume, - not the absence of agent-derived text, that justifies the exemption. + """Report one reading through the platform alert seam (#3246). Never raises. + + One pending row per subject — a subscription's cleaned id, or `fleet` — + on the sentinel host. A newer reading updates that row in place, so a + critical reading replaces the warning (`tier` is the kind's material key) + and the figure on the row is always the latest. The row is ended by the + platform when the evaluation pass finds the subscription back under its + threshold (`clear_recovered`). Returns True when a row was filed or + updated. """ - item = { - "id": item_id, - "type": "alert", - "status": "pending", - "priority": priority, - "title": title, - "question": question, - "context": context, - "created_at": utc_now_iso(), - # Must stay None: `mark_operator_queue_expired` flips any pending row - # past `expires_at` to expired fleet-wide every 5s. - "expires_at": None, - } - try: - db.create_operator_queue_item(ALARM_AGENT_NAME, item) - logger.warning("[headroom-alert] %s", title) - return True - except Exception: # noqa: BLE001 - logger.exception("[headroom-alert] failed to emit %s", item_id) - return False + from services import platform_alerts + outcome = platform_alerts.observe( + ALARM_AGENT_NAME, "subscription_headroom", key, + title=title, question=question, priority=priority, context=context, + ) + logger.warning("[headroom-alert] %s (%s)", title, outcome) + return outcome in (platform_alerts.OBSERVED_CREATED, platform_alerts.OBSERVED_UPDATED) + + +def clear_recovered(live_keys: Sequence[str]) -> int: + """#3246: end every pending headroom row whose subject is not in + `live_keys` — the subscriptions (and `fleet`) the evaluation pass could + not show to be back under the threshold. Never raises.""" + from services import platform_alerts + return platform_alerts.reconcile(ALARM_AGENT_NAME, "subscription_headroom", live_keys) def emit_subscription_alert( @@ -383,9 +363,8 @@ def emit_subscription_alert( ) -> bool: """One subscription crossed its weekly threshold. - The body is self-contained. A warning row cannot be edited later (see the - module docstring), so the escalation must stand on its own rather than - assume the operator reads it beside the earlier item. + The body is self-contained: the escalation replaces the warning on the + subscription's one row (#3246), so it must stand on its own. """ label = subscription_name or subscription_id episode = episode_key(resets_at, now=now) @@ -421,7 +400,7 @@ def emit_subscription_alert( lines.append("No agents are currently assigned to this subscription.") return _emit( - alert_id(subscription_id, episode, tier), + subject_key(subscription_id), title=headline, question="\n".join(lines), priority=priority_for(tier, projected_end), @@ -433,6 +412,7 @@ def emit_subscription_alert( "projected_end_pct": projected_end, "threshold_pct": threshold_pct, "resets_at": resets_at, + "window": episode, "agent_count": len(agents), "on_pace_to_exhaust": on_pace, }, @@ -470,7 +450,7 @@ def emit_fleet_alert( "There is no subscription with meaningful headroom left to move agents onto.", ] return _emit( - fleet_alert_id(episode), + FLEET_KEY, title=f"All {len(saturated)} subscriptions are near their weekly limit", priority="high", question="\n".join(lines), @@ -479,5 +459,6 @@ def emit_fleet_alert( "subscription_count": len(saturated), "threshold_pct": threshold_pct, "earliest_reset_at": earliest_reset, + "window": episode, }, ) diff --git a/src/backend/services/subscription_recovery_service.py b/src/backend/services/subscription_recovery_service.py index 5a44bd067f..ae4fb2ea40 100644 --- a/src/backend/services/subscription_recovery_service.py +++ b/src/backend/services/subscription_recovery_service.py @@ -404,6 +404,19 @@ async def _evaluate_alerts( } verdict = alerts.fleet_verdict(classifications) + # #3246: the evaluation pass is the one place that knows a subscription + # is back under its threshold, so it ends the rows it no longer backs. + # Only a MEASURED `HAS_HEADROOM` clears — an unassessable member keeps + # its row (no evidence is not recovery, the ent#100 rule) — and the + # fleet row survives until some member is measured with room. + live_keys = [ + alerts.subject_key(sid) for sid, c in classifications.items() + if c != alerts.HAS_HEADROOM + ] + if not verdict["has_headroom_ids"]: + live_keys.append(alerts.FLEET_KEY) + await asyncio.to_thread(alerts.clear_recovered, live_keys) + # The fleet alert names every saturated subscription, so emitting the # individual ones beside it would be N+1 operator items for one event. if verdict["saturated"]: diff --git a/src/backend/services/system_agent_service.py b/src/backend/services/system_agent_service.py index 9d593f04d2..daa8f5328a 100644 --- a/src/backend/services/system_agent_service.py +++ b/src/backend/services/system_agent_service.py @@ -216,6 +216,7 @@ async def ensure_deployed(self) -> dict: else: result["message"] = "System agent already running" logger.info("System agent already running on a current base image") + self._clear_alert("base_image_stale") return result # #1816 — STOPPED: the cold boundary. Delegate to the shared @@ -262,6 +263,7 @@ async def ensure_deployed(self) -> dict: "(pre-flight failed — see the preceding warning)" ) logger.info("System agent started (adoption skipped by pre-flight)") + self._clear_alert("system_agent_start_failed") return result except Exception as e: result["action"] = "start_failed" @@ -285,6 +287,9 @@ async def ensure_deployed(self) -> dict: "System agent started (recreated=%s, reason=%s)", result["recreated"], result["recreate_reason"], ) + self._clear_alert("system_agent_start_failed") + if start_result.get("recreate_reason") == "image_drift": + self._clear_alert("base_image_stale") return result except Exception as e: result["action"] = "start_failed" @@ -625,14 +630,15 @@ def _emit_base_image_stale_alert(self) -> None: SystemAgentService._last_base_image_alert_at = now ts = utc_now_iso() - item = { - "id": f"{BASE_IMAGE_STALE_ALERT_PREFIX}{SYSTEM_AGENT_NAME}-{ts}", - "agent_name": SYSTEM_AGENT_NAME, - "type": "alert", - "status": "pending", - "priority": "high", - "title": "System agent is running a stale base image", - "question": ( + # #3246: through the platform alert seam — one pending row for the + # system agent's image, ended by `_clear_alert` once a boot finds the + # image current or a start adopts the rebuilt one. + from services import platform_alerts + outcome = platform_alerts.observe( + SYSTEM_AGENT_NAME, "base_image_stale", SYSTEM_AGENT_NAME, + priority="high", + title="System agent is running a stale base image", + question=( f"{SYSTEM_AGENT_NAME} is still running the base image it was " f"created from, and a newer trinity-agent-base has been built. " f"It is never recreated while running. Restart it " @@ -640,14 +646,19 @@ def _emit_base_image_stale_alert(self) -> None: f"NOTE: adoption replaces the container, so anything installed " f"outside /home/developer is lost." ), - "context": {"agent_name": SYSTEM_AGENT_NAME, "detected_at": ts}, - "created_at": ts, - } - try: - db.create_operator_queue_item(SYSTEM_AGENT_NAME, item) - logger.warning("base-image staleness alarm emitted for %s", SYSTEM_AGENT_NAME) - except Exception: - logger.exception("failed to emit base-image staleness alarm") + context={"agent_name": SYSTEM_AGENT_NAME, "detected_at": ts}, + ) + logger.warning("base-image staleness alarm for %s: %s", SYSTEM_AGENT_NAME, outcome) + + @staticmethod + def _clear_alert(kind: str) -> None: + """#3246: the condition behind a system-agent alert cleared — the + platform ends its pending row. Never raises.""" + from services import platform_alerts + if kind == "base_image_stale": + platform_alerts.clear(SYSTEM_AGENT_NAME, "base_image_stale", SYSTEM_AGENT_NAME) + elif kind == "system_agent_start_failed": + platform_alerts.clear(SYSTEM_AGENT_NAME, "system_agent_start_failed", SYSTEM_AGENT_NAME) def _emit_start_failed_alert(self, reason: str) -> None: """#1816 R1: the platform may now have NO system agent, and nobody is @@ -666,26 +677,21 @@ def _emit_start_failed_alert(self, reason: str) -> None: `summary_or_error`).""" ts = utc_now_iso() reason = sanitize_text(reason) - bucket = int(time.time()) // START_FAILED_ALERT_BUCKET_SECONDS - item = { - "id": f"{BASE_IMAGE_STALE_ALERT_PREFIX}start-{SYSTEM_AGENT_NAME}-{bucket}", - "agent_name": SYSTEM_AGENT_NAME, - "type": "alert", - "status": "pending", - "priority": "critical", - "title": "System agent could not be started", - "question": ( + # #3246: through the platform alert seam — one pending row per system + # agent (a newer failure updates it in place), ended by `_clear_alert` + # when a start succeeds. + from services import platform_alerts + outcome = platform_alerts.observe( + SYSTEM_AGENT_NAME, "system_agent_start_failed", SYSTEM_AGENT_NAME, + priority="critical", + title="System agent could not be started", + question=( f"{SYSTEM_AGENT_NAME} failed to start: {reason}. The platform " f"orchestrator may be unavailable." ), - "context": {"agent_name": SYSTEM_AGENT_NAME, "detected_at": ts}, - "created_at": ts, - } - try: - db.create_operator_queue_item(SYSTEM_AGENT_NAME, item) - logger.warning("system-agent start-failure alarm emitted: %s", reason) - except Exception: - logger.exception("failed to emit system-agent start-failure alarm") + context={"agent_name": SYSTEM_AGENT_NAME, "detected_at": ts}, + ) + logger.warning("system-agent start-failure alarm (%s): %s", outcome, reason) def _set_system_scope(self, key_id: str): """Update MCP key to have system scope (bypasses permissions).""" diff --git a/src/frontend/src/components/operator/QueueCard.vue b/src/frontend/src/components/operator/QueueCard.vue index f96181f215..6fe2d8503e 100644 --- a/src/frontend/src/components/operator/QueueCard.vue +++ b/src/frontend/src/components/operator/QueueCard.vue @@ -30,6 +30,10 @@

+ +

{{ seenLine }}

@@ -233,7 +237,7 @@ import { useOperatorQueueStore } from '../../stores/operatorQueue' import { useAgentsStore } from '../../stores/agents' import { agentNameTooltip } from '../../utils/agentName' import { - queueTypeLabel, queueResponseKind, queueSyncBadge, queueReaskBadges, + queueTypeLabel, queueResponseKind, queueSyncBadge, queueReaskBadges, queueSeenLine, SOMETHING_ELSE, SOMETHING_ELSE_LABEL, offeredChips, decidedByOptions, buildQueueResponse, } from '../../utils/operatorQueue' import AgentAvatar from '../AgentAvatar.vue' @@ -257,6 +261,7 @@ const agentsStore = useAgentsStore() const responseKind = computed(() => queueResponseKind(props.item)) // #2915: sync/delivery/aging badge and the refused-response notice. const syncBadge = computed(() => queueSyncBadge(props.item)) +const seenLine = computed(() => queueSeenLine(props.item)) // #3246 const reaskBadges = computed(() => queueReaskBadges(props.item, store.items)) // trinity-enterprise#611 const diverged = computed(() => store.divergedItemId === props.item.id) const notAddressee = computed(() => store.notAddresseeItemId === props.item.id) diff --git a/src/frontend/src/components/operator/QueueItemDetail.vue b/src/frontend/src/components/operator/QueueItemDetail.vue index a044bc94d6..26e02e88e4 100644 --- a/src/frontend/src/components/operator/QueueItemDetail.vue +++ b/src/frontend/src/components/operator/QueueItemDetail.vue @@ -40,6 +40,8 @@

{{ item.title }}

+ +

{{ seenLine }}

@@ -98,11 +100,18 @@

Response

-

{{ decisionLabel(item.response) }}

-

{{ item.response_text }}

-

- by {{ item.responded_by_email }} · {{ formatDate(item.responded_at) }} + +

+ {{ endingText }}

+
@@ -208,7 +217,7 @@ import { useOperatorQueueStore } from '../../stores/operatorQueue' import { useAgentsStore } from '../../stores/agents' import { agentNameTooltip } from '../../utils/agentName' import { - queueResponseKind, queueTypeLabel, + queueResponseKind, queueTypeLabel, queueEnding, queueEndingText, queueSeenLine, SOMETHING_ELSE, SOMETHING_ELSE_LABEL, offeredChips, decidedByOptions, decisionLabel, buildQueueResponse, } from '../../utils/operatorQueue' @@ -224,6 +233,10 @@ const item = computed(() => store.selectedItem) // closed from the queue at all — and a budgeted alert type whose items // cannot be closed jams its own pending cap permanently. const responseKind = computed(() => queueResponseKind(item.value)) +const ending = computed(() => queueEnding(item.value)) +const endedByPlatform = computed(() => ending.value?.who === 'the platform') +const endingText = computed(() => queueEndingText(ending.value)) +const seenLine = computed(() => queueSeenLine(item.value)) const selectedOption = ref(null) const responseText = ref('') diff --git a/src/frontend/src/components/operator/ResolvedCard.vue b/src/frontend/src/components/operator/ResolvedCard.vue index 67e67fcea1..c7ab9d989d 100644 --- a/src/frontend/src/components/operator/ResolvedCard.vue +++ b/src/frontend/src/components/operator/ResolvedCard.vue @@ -105,7 +105,10 @@ const isTerminalWithoutResponse = computed(() => ) const ending = computed(() => queueEnding(props.item)) // trinity-enterprise#611 const endingText = computed(() => queueEndingText(ending.value)) -const note = computed(() => props.item.response_text || props.item.disposition_reason || '') +// #3246: a platform ending's reason is a token the ending text already words +// (`queueEndingText`), never printed raw as though an operator wrote it. +const note = computed(() => props.item.response_text + || (ending.value && ending.value.who === 'the platform' ? '' : props.item.disposition_reason) || '') const endedAtAbsolute = computed(() => (ending.value?.when ? formatLocalDateTime(ending.value.when) : '')) // An answer's "who" rides beside its time; a cancel / expiry names it in its label. const endingMeta = computed(() => { diff --git a/src/frontend/src/utils/operatorQueue.js b/src/frontend/src/utils/operatorQueue.js index 2764d2c468..5be3d34700 100644 --- a/src/frontend/src/utils/operatorQueue.js +++ b/src/frontend/src/utils/operatorQueue.js @@ -383,8 +383,15 @@ export const QUEUE_RESPONSE_NOT_ADDRESSEE = * Workspace projection's own `status` (`answered`) reads the same way. * - `who` — the person, for the Operating Room (`disposed_by_email`, or a * legacy answer's `responded_by_email`); the Workspace projection's coarse - * `ended_by` (`you` / `the operator`); `timeout` for an expiry; `null` when - * the platform does not know. + * `ended_by` (`you` / `the operator`); `timeout` for an expiry; `the + * platform` for a row the platform itself ended (#3246: `disposed_by = + * 'platform'`, or the projection's `ended_by = 'platform'`); `null` when the + * platform does not know. + * - `reason` — #3246: the platform's ending in words (`condition_cleared`, + * `superseded`); `null` for every other ending. A platform ending is never a + * person's answer and never a timeout. + * - `alert` — the row is an alert: its expiry reads "nobody acted on it", + * because nobody was asked anything. * - `when` — the ledger's `disposed_at`, the projection's `ended_at`, or a * legacy answer's `responded_at`. NEVER `created_at`: that is when the ask * was filed, and showing it as the ending time is the defect this replaces. @@ -396,6 +403,13 @@ export const ENDING_LABELS = Object.freeze({ expired: 'Expired', }) +/** #3246 — why the platform ended a row, in words. An unknown reason token + * falls back to nothing rather than being printed raw. */ +export const PLATFORM_ENDING_REASONS = Object.freeze({ + condition_cleared: 'the condition cleared', + superseded: 'superseded by a newer reading', +}) + function endingKind(item) { const d = item.disposition if (d === 'answered' || d === 'dismissed' || d === 'cancelled' || d === 'expired') return d @@ -411,8 +425,12 @@ export function queueEnding(item) { if (!kind) return null const when = item.disposed_at || item.ended_at || (kind === 'answered' ? item.responded_at : null) || null let who = null + let reason = null if (kind === 'expired') { who = 'timeout' + } else if (item.disposed_by === 'platform' || item.ended_by === 'platform') { + who = 'the platform' + reason = PLATFORM_ENDING_REASONS[item.disposition_reason] || null } else if (item.ended_by === 'you') { who = 'you' } else if (item.ended_by === 'operator') { @@ -420,17 +438,49 @@ export function queueEnding(item) { } else { who = item.disposed_by_email || (kind === 'answered' ? item.responded_by_email : null) || null } - return { kind, label: ENDING_LABELS[kind], who, when } + return { kind, label: ENDING_LABELS[kind], who, when, reason, alert: item.type === 'alert' } } /** The ending in words: "Cancelled by op@…", "Answered by you", - * "Expired — nobody answered in time", or the bare label when nobody is known. */ + * "Ended by the platform — the condition cleared", "Expired — nobody answered + * in time" ("nobody acted on it" for an alert), or the bare label when nobody + * is known. */ export function queueEndingText(ending) { if (!ending) return '' - if (ending.kind === 'expired') return `${ending.label} — nobody answered in time` + if (ending.kind === 'expired') { + return `${ending.label} — ${ending.alert ? 'nobody acted on it' : 'nobody answered in time'}` + } + if (ending.who === 'the platform') { + return ending.reason ? `Ended by the platform — ${ending.reason}` : 'Ended by the platform' + } return ending.who ? `${ending.label} by ${ending.who}` : ending.label } +function relativeAgo(iso, now) { + const then = new Date(iso).getTime() + if (Number.isNaN(then)) return null + const diffMin = Math.floor((now - then) / 60000) + if (diffMin < 1) return 'just now' + if (diffMin < 60) return `${diffMin}m ago` + const diffHr = Math.floor(diffMin / 60) + if (diffHr < 24) return `${diffHr}h ago` + return `${Math.floor(diffHr / 24)}d ago` +} + +/** + * #3246 — a pending platform alert is ONE row per condition, updated in place by + * every new reading, so a row seen many times must not read as a fresh alert. + * "seen N times · last seen 5m ago" when `context.seen_count > 1`; `''` for an + * ended row, a row seen once, and anything that is not an alert. + */ +export function queueSeenLine(item, now = Date.now()) { + if (!item || item.status !== 'pending' || item.type !== 'alert') return '' + const n = Number(item.context && item.context.seen_count) + if (!Number.isInteger(n) || n < 2) return '' + const ago = item.last_seen_at ? relativeAgo(item.last_seen_at, now) : null + return ago ? `seen ${n} times · last seen ${ago}` : `seen ${n} times` +} + /** What the resolved feed sorts by (#627 AC6): when the item ended; a legacy * answer's time; else — the only timestamp such a row has — when it was filed. */ export function queueEndingSortTime(item) { diff --git a/src/frontend/src/views/MobileAdmin.vue b/src/frontend/src/views/MobileAdmin.vue index c7da38c544..16af5de3bc 100644 --- a/src/frontend/src/views/MobileAdmin.vue +++ b/src/frontend/src/views/MobileAdmin.vue @@ -258,6 +258,8 @@
{{ queueTypeLabel(item.type) }}

+ +

{{ queueSeenLine(item) }}