From 424db1555de2d335574f66c89f84ea82a1005c95 Mon Sep 17 00:00:00 2001
From: Joseph Doherty <dohejw01@gmail.com>
Date: Tue, 10 Nov 2026 16:46:42 -0500
Subject: [PATCH] fix: outbox KPI per-status legs, composite event-log indexes,
 AuditLog ingest FORCESEEK (#122, #123, #124)

#122: the Notification Outbox KPI snapshots ORed every status in one
aggregation, which scanned the clustered index once the table held ~1 M
Parked rows (0.7-2.6 s; Health dashboard up to 5 s). Each snapshot now runs
three status-equality legs that seek their own index (~0.12 s); a cold Health
load on the rig is 0.12-0.52 s.

#123: site_events event_type/severity indexes are now composites with
timestamp, so a filtered page is read in order and stops at the limit
(severity=Error over 1 GB: 2,287 -> 15 ms cold, 115 -> 1.6 ms warm). Costs
454 -> 530 B/row, ~14 % fewer rows under the same cap. The schema apply drops
the old indexes in place; the cap size model samples timestamp three times
and re-scores inside its band with unchanged constants.

#124: the set-based AuditLog insert's NOT EXISTS probe scanned the EventId
index and concurrent writers deadlocked on S/IX page locks. WITH (FORCESEEK):
4 writers 775 rows/s + 76 deadlocks in 30 s -> 5,214 rows/s + 0. New
concurrent-writer regression test fails 4/4 without the hint.
---
 CLAUDE.md                                     |   5 +-
 docs/components/SiteEventLogging.md           |   8 +-
 ...7-site-event-log-cap-semantics-decision.md |  18 ++
 .../Repositories/AuditLogRepository.cs        |  13 +-
 .../NotificationOutboxRepository.cs           | 242 +++++++++---------
 .../EventLogPurgeService.cs                   |  14 +-
 .../SiteEventLogSchema.cs                     |  18 +-
 ...ationOutboxRepositoryKpiQueryShapeTests.cs |  66 ++++-
 .../Repositories/AuditLogRepositoryTests.cs   |  59 +++++
 .../SchemaIndexTests.cs                       |  72 +++++-
 .../SiteEventLogSchemaTests.cs                |   2 +-
 .../SiteEventsSchemaCalibrationTests.cs       |  27 +-
 12 files changed, 386 insertions(+), 158 deletions(-)

diff --git a/CLAUDE.md b/CLAUDE.md
index 7ef69773..0cd506dc 100644
--- a/CLAUDE.md
+++ b/CLAUDE.md
@@ -182,6 +182,7 @@ spec for each is `docs/requirements/Component-<Name>.md`, and `README.md` carrie
 - Audit-write failure NEVER aborts the user-facing action — audit is best-effort, the action's own success/failure path is authoritative.
 - 365-day central retention with monthly partition-switch purge; per-channel retention overrides (`AuditLog:PerChannelRetentionDays`) expire rows earlier than the global window via a bounded, batched row DELETE on the purge actor's maintenance path — values must be shorter than the global window (M5.5 T3); 7-day site SQLite retention with a hard `ForwardState` invariant (no row purged until forwarded or reconciled).
 - Append-only enforced via DB roles (writer role has INSERT only, no UPDATE/DELETE).
+- **The set-based ingest probe is `WITH (FORCESEEK)` (#124).** Left alone, `InsertManyIfNotExistsAsync`'s `NOT EXISTS` anti-semi-join scanned the EventId index, and concurrent writers (ingest + reconciliation singletons, dual-write, direct writes) deadlocked on S/IX page locks — 4 writers lost ~45 % of packets to 1205. With the hint: 0 deadlocks, 8 writers 8.6 k rows/s. Reproduce with `ResidualBench audit-contention`.
 - **Clustered key `(OccurredAtUtc, EventId)` + aligned unique `UX_AuditLog_EventId_Occurred (EventId, OccurredAtUtc)` (migration `FlipAuditLogClusteredKey`, #111, B2 of the #78 benchmark).** The Audit Log page's newest-first page/keyset now reads an ordered range instead of the whole table (300k rows: 90,532 → 34 reads). Offline whole-table rebuild — script `docs/plans/sql/FlipAuditLogClusteredKey.sql` (+ `.rollback.sql`), both centrals stopped: the partition-switch staging table now reads the live key order, but binaries older than the flip hard-code the old one. DEV/QA not applied (sizing + window in #111).
 - **Parquet cold archive (T2, ScadaBridge#42) — default OFF (`AuditLog:Archive:Enabled` + absolute `Directory`, shared by both centrals).** The purge actor writes each expired partition to `{Directory}/{YYYY}/auditlog-{YYYY-MM}-{stamp}Z.parquet` (10 stored columns + `ChainSeq`/`RowHash`, tick-exact `OccurredAtUtcTicks`) plus `.tombstones.parquet` and a `.manifest.json` (written last; chain head at archive time, SHA-256 per file), read under the chain lock SHARED; then the switch runs with a guard (under the EXCLUSIVE lock: partition row count + each month's `LastSeq` must still equal the archive's, else THROW 51002/51003 and nothing is purged). Fail closed: a partition that cannot be archived is not purged. `audit archive list|verify` = `ListAuditArchives`/`VerifyAuditArchive` (verify-chain's gate): file hashes, chain recomputed from the archive, chain end vs the DB purge anchor (`Matches`/`Superseded`/`NotPurged`/`Mismatch`), CLI exit 3. `audit export --format parquet` streams the 24 export columns (was 501). No migration. **Owner decisions (#109):** destination stays a directory/UNC (no object-storage client; share + permissions named at enablement), parquet-tools-only reading (no re-import), unsigned like the chain. **Retention** `AuditLog:Archive:RetentionDays` (default 0 = never; else > `AuditLog:RetentionDays`, counted from the month's end) and **superseded stale-archive cleanup** (`DeleteSupersededStaleArchives`, default true: a retried purge's `Archived` archive is deleted only when a later `Purged`, verifying archive provably holds every row — same anchor + chain-end hash, unchained rows by EventId; else `KeptStale`) run after every purge tick; deletion touches only the archive name's own three files, manifest first; `audit archive cleanup [--dry-run]` = `CleanupAuditArchives` (Administrator).
 - **Hash-chain tamper evidence (T1, ScadaBridge#41) — default OFF (`AuditLog:HashChain:Enabled`).** One SHA-256 chain per UTC calendar month: `ChainSeq` + `RowHash = SHA-256(prev || canonical(row))` on `AuditLog` (migration `AddAuditLogHashChain`), linked by the central `AuditLogHashChainService` (both nodes, serialised by `sp_getapplock`) — never by the ingest path. `AuditLogChainHead` catches tail deletion and carries the partition-switch checkpoint anchor; the per-channel DELETE writes an `AuditLogChainTombstone` per chained row in the same transaction. `audit verify-chain --month` = management command `VerifyAuditChain` (OperationalAudit, unscoped callers only), first break with partition/ChainSeq/EventId, CLI exit 3. A chained row's `SourceNode` is frozen (backfill skips it). **Unsigned by owner decision (#107); external anchor = off-box head export** (`AuditLog:HashChain:HeadExport`, default off: both centrals append changed per-month heads — month, `LastSeq`/`LastHash`, purge anchor, row + tombstone counts — to append-only JSONL files `{Directory}/{yyyy}/audit-chain-heads-{date}-{node}.jsonl` or RFC 5424 syslog UDP/TCP); `audit verify-heads --path` = `VerifyAuditChainHeads` checks them against the DB (a re-hashed forgery passes verify-chain, fails verify-heads; exit 3). **Two-login hardening (#107 item 3):** `AuditLog:MaintenanceConnectionString` (`${secret:}`-capable, unset = runtime login) runs the chain job, verify/head export, partition switch-out + archive and per-channel purge on `AuditMaintenanceContext` (keyed `audit-maintenance` repository/archive); that login gets `scadabridge_audit_chainer` + `scadabridge_audit_purger`, the runtime login `scadabridge_audit_writer` (whose DENYs block chaining/purging) — pinned by `AuditMaintenanceLoginTests`. Since #113 the monthly roll-forward and `backfill-source-node` use it too (migration `AuditMaintenanceGrants`: purger gets `ALTER ANY DATASPACE` + `UPDATE (SourceNode)`). Head-export failures (#115) bump `scadabridge.audit.chain_head_export.failures` and Degrade `/health/ready` entry `audit-chain-head-export`; archive cleanup (#114) also deletes writer-named orphan temp / manifest-less files older than 24 h (`DeletedOrphan`).
@@ -215,8 +216,8 @@ spec for each is `docs/requirements/Component-<Name>.md`, and `README.md` carrie
 - **Flow Debugger (2026-09-05, Deployment → `/deployment/flow-debugger` + CLI `flow-debug run-method|run-script|run-value-change|replay`).** Runs one entry point (an inbound API method, or a site script) and draws every step as a Central / Site / Tags / **External** swimlane; *Live* really runs it, *Replay* rebuilds a past execution. **v1.1 (2026-09-06, `docs/plans/2026-09-06-flow-debugger-outbound-calls.md`):** every `ExternalSystem.*` / `Database.*` / `Notify.Send` a traced run makes is an `ApiCall` / `DbCall` / `Notify` node on the External lane — the site routes those already-existing `ApiOutbound` / `DbOutbound` / `Notification` rows through the same inline trace collector, the replay query widens to those channels, and `FlowGraphBuilder` folds a cached operation's rows by `(kind, CorrelationId)` into ONE node with a `Lifecycle` (sync calls in-chain, cached/notify on dashed `Async` edges, trigger cascades on `Spawn` edges, the Response linked from the inbound execution's last step). Later lifecycle rows are not inline, so after a Live run the page polls the Replay command (5 s, ≤ 2 min) while any node is `Pending`, plus a *Refresh outcomes* button. **A refresh MERGES, never replaces (2026-09-06, found live in the browser):** the page unions the re-read rows into the run's own rows by `EventId` and rebuilds the graph, because a Live run carries its site rows inline while the same rows reach central's audit store by telemetry ~5–30 s later — a Replay read taken in between lacks them, and the old wholesale swap made an already-drawn node (a resolved cached DB write) vanish until the next manual refresh. The inline rows ARE the audited rows (`FlowTraceCollector` sinks the same `AuditEvent` objects the site writer persists), so the union is exact. Central method-script `Database.*` calls (`InboundDatabaseHelper`) are now audited (`DbOutbound`/`DbWrite`) — they were not. Rig demo: `docker/seed-flow-outbound-demo.sh` + `verify-delmia-notifier.sh` step I. **Trace steps ARE audit events** — `AuditChannel.ScriptTrace` with kinds `RouteCall` (central, **always-on** per `Route.To().Call`, joins the two halves of a Replay) and `ScriptRun` / `AttributeWrite` / `AttributeWait` (site, **gated** by a `bool Trace` riding beside `ParentExecutionId` on `RouteToCallRequest` / `ScriptCallRequest` / `SetStaticAttributeCommand` / `AttributeValueChanged`; proto fields additive, `site_command.proto` imports `sitestream.proto`). A Live run gets the site rows **inline** on `RouteToCallResponse.TraceEvents` (+ `ChildExecutionId`) because the 5 s telemetry cadence is too slow; Replay reads them from `AuditLog` (`GetTraceEventsAsync`). A value-change on a data-sourced attribute is a REAL tag write; the Instance Actor's **trace arm** (`SiteRuntime:TraceArmWindowSeconds`, 30) stamps the device echo with the writer's execution id so the triggered run is parented and traced, and central polls the site's new `PullTraceEvents` RPC (point read, not reconciliation) for it. Execution path is four management commands on the **per-node `ManagementActor`** (`FlowDebug*`; the three run commands are Deployer-only, Replay is wider) — a standby refuses the three run commands with **`STANDBY_NODE`** (shown verbatim); `ApiMethod.TraceAllCalls` (migration `AddApiMethodTraceAllCalls`, CLI `--trace-all-calls`) traces every real call to a method for later Replay (7–10 rows/call). Prediction before a run is a Roslyn syntax walk (`FlowShapePredictor`, REQ-SA-7, DB-free); the graph (`FlowStep`/`FlowGraphBuilder`, Commons) is a central projection, never a wire type; the renderer is custom SVG (`FlowGraphView`). Rig: `AuditLog:PerChannelRetentionDays:ScriptTrace=30` on both centrals; `verify-delmia-notifier.sh` step H proves live + replay. **Mixed-version caveat:** an older site ignores the flag → central `RouteCall` only, page says *site not traced*. **v1.2 (2026-09-06, `docs/plans/2026-09-06-flow-debugger-followups.md`):** a `Parked` External node offers **Retry / Discard in place** (same site-scope check, confirm and relay as the report pages, via `IFlowDebugService.RetryOutboundAsync`/`DiscardOutboundAsync`, then the page re-polls Replay until the node leaves Parked) and every folded node deep-links to `/site-calls/report?correlationId=` / `/notifications/report?correlationId=` (both pages accept it through additive `TrackedOperationIdFilter` / `NotificationIdFilter` query params and auto-open the row); register row 111 closed. **Trap closed the same day:** `FlowDebugService.PredictMethodAsync` used to resolve route targets with `GetAwaiter().GetResult()` inside the synchronous predictor walk — on the Blazor renderer sync context that DEADLOCKED the whole circuit (page inert, healthy websocket, no error) whenever a method routing to a site script was chosen; now a resolve-and-rewalk loop, pinned by a single-thread-pump test. Never block on a task from a Blazor event/lifecycle path. Browser checks of the page go through Playwright (`FlowDebuggerTests`, fixture login + an ephemeral method), not the login-disabled override. **Arch-reviewed 2026-09-06** (`docs/plans/2026-09-06-flow-debugger-arch-review.md`, remediation `docs/plans/2026-09-06-flow-debugger-arch-review-remediation.md`): the inline Live trace copy is now redacted and capped by the shared `FlowTraceCollector` (500 events **and** 1 MB of `DetailsJson`, F1), a method-entry Live run enforces the operator's site scope at route time (`RouteHelper.WithSiteScope` → `SITE_SCOPE_DENIED` → `ManagementUnauthorized`, F2) — **and since Gitea #55 a site-scoped operator is refused a method Live run outright** (a method has no site, so its central-side effects can't be scoped; audited, page disables the entry; Replay + script entry unchanged), and the page/Replay opened to the `OperationalAudit` roles under the new `RequireFlowDebugger` policy while everything that executes stays Deployer-only (F3). Design `docs/plans/2026-09-05-flow-debugger-design.md`; plan `docs/plans/2026-09-05-flow-debugger.md`. **v1.3 (2026-09-08, `docs/plans/2026-09-08-flow-debugger-script-entry-design.md`):** the second entry point is site → instance → **script**; ValueChange/Conditional scripts are fired by writing the trigger's attribute (derived from the trigger config, not picked), every other kind is a direct traced `RouteToCall` with a `ParameterValueForm` from the script's `ParameterDefinitions` and one operator-attributed `RouteCall` root row (`FlowDebugRunScriptCommand`, Deployer-only, long-running); `run-value-change` stays as the attribute-level CLI escape hatch (fires Expression triggers). **Instance picker (2026-09-08, `docs/plans/2026-09-08-instance-picker-design.md`):** Flow Debugger and Debug View both replaced their flat instance `<select>` with the shared **`InstancePicker`** combobox — a text field whose dropdown is the site's area tree (`InstanceTreeBuilder`, all areas collapsed on open), typed text filters and expands to the matches; `Areas`/`Instances` are compared by reference, so callers must assign the loaded lists once per site change, never per render, or the panel closes on every re-render. **v1.4 (2026-09-09, `docs/plans/2026-09-09-flow-debugger-step-logs-design.md`):** scripts can log — a `Log.Debug/Info/Warn/Error(message, exception?)` global on all four surfaces (site `ScriptGlobals`, inbound `InboundScriptContext`, the `ScriptCompileSurface` deploy gate as `CompileLog`, and both Test Run sandboxes/editor hosts, where it appends to the console pane), always reaching the node `ILogger` and emitting a `ScriptTrace`/**`ScriptLog`** audit row (one per line) **only when the run is traced**, so an untraced timer script costs no audit volume. `FlowGraphBuilder` **folds those rows onto the step that produced them** (`FlowStep.Logs`) and never draws them as nodes — attribution is `Extra.stepEventId`, pre-minted by `SetAttribute`/`WriteBatchAndWait`/`WaitInternal`/sync `ExternalSystem.Call` through an `AsyncLocal` `TraceStepScope`, else the execution's anchor row, else its first step, else `FlowGraph.UnattachedLogLines`; the runtime's own Information+ entries are captured for free by a `TraceCapturingLogger` decorator instead of editing every `_logger.LogWarning` call site. The failing step now carries the **whole exception** (`errorDetail = ex.ToString()` on the site `ScriptRun` row and on the central inbound row — the latter never on the HTTP body, which stays `SCRIPT_ERROR`). **The two halves are NOT gated alike:** the central inbound row's `errorDetail` is stamped on **every failing method request, traced or not** (`EndpointExtensions` stashes it on `HttpContext.Items`, `AuditWriteMiddleware` consumes it), bounded only by the redactor's `ApiInbound` cap `AuditLog:InboundMaxBytes` (1 MiB default; the 64 KB `ErrorCapBytes` is what bounds every OTHER channel's error rows, the site `ScriptRun` row included) — tracing only decides whether the row is later readable as a Flow Debugger step; the site `ScriptRun` `errorDetail` is **traced-only**, because an untraced run emits no `ScriptTrace` row at all, rendered as a red `Type: message` subtitle plus a **Log (n)** tab; the node badge is `✕ n ! n ≡ n` and the strip counts `Log: n errors · n warnings · n info` (a `Log.Error` is a line, not a failed step — it never moves the state chip). Bounded: `ScadaBridge:SiteRuntime:MaxLogLinesPerRun` / `ScadaBridge:InboundApi:MaxLogLinesPerRun` (200, validated > 0), 2 000-char messages, one closing `warn`/`runtime` row naming the dropped count, and a **low-priority inline lane** (`FlowTraceCollector.Add(evt, lowPriority: true)`, log rows stop 50 events short of the count cap and at 75 % of the byte budget, reserving the rest for the step rows written at run end) whose overflow sets `LogsTruncated` → the page badge *inline log lines capped — refresh outcomes pulls the rest*, which clears once a refresh merges in more log rows. **"Never `await` an audit write on a script path" still holds** — every log row is enqueued fire-and-continue, exactly like the 2026-09-06 cached-call telemetry fix. **Trap:** build `ScriptLogger` over the UNDECORATED logger — `TraceCapturingLogger` refuses to wrap another `TraceCapturingLogger` (it would emit two rows per line). CLI `flow-debug … --show-logs` + a `logs` column; rig demo `docker/seed-flow-log-demo.sh` + `verify-delmia-notifier.sh` step J. Mixed-version: an OLD central projects a NEW site's `ScriptLog` rows as `Unknown` nodes — upgrade central first, as always. **Node log (2026-11-10, Gitea #56 + #120, step-logs design §12):** while a run is traced, other actors' node-log lines tagged with its `ExecutionId` (Instance Actor, DCL actors, adapters; central callees) are copied into its `ScriptLog` rows as source `node` by the host's `ExecutionLogCaptureSink`, with no `site_events` schema change. They share the run's line cap and redaction. Untraced runs cost one volatile read. They are listed per execution in a **Node log** card (`FlowGraph.NodeLogs`), never folded onto a step. Akka `ILoggingAdapter` lines now reach Serilog via `akka.loggers = [SerilogAkkaLogger]` (loglevel stays INFO). The DCL actors use `ExecutionAwareLogging.GetLogger(Context)` so their Akka lines keep the id across the logger-actor hop. **Traps:** the step loggers' own forwards run inside `ExecutionLogCapture.Suppress()`, which is per thread, so the sink must never sit behind an async wrapper. The capture registration is declared inside the launcher/executor `try` so it ends before the catch blocks log the outcome. **Method pane structured inputs (2026-09-10, Gitea #37, `docs/plans/2026-09-10-flow-debugger-method-params-design.md`):** the *Run inbound API method* pane now renders the same `ParameterValueForm` as the script pane (keyed on the method id) plus a **Form | Raw JSON** whole-payload escape hatch (Monaco, schema-validated on every edit); the page keeps ONE value tree and serializes it as-is, the old per-field text coercion is gone, and Run is disabled while the form or the raw document is invalid — so a required array-of-object parameter (MesMoveOut `WorkOrders`) can be entered, and a failed parse is never sent as a string. The per-method `localStorage` entry stores the value tree and is dropped on restore if it no longer validates (that also retires the pre-#37 flat string entries).
 - Health reports: 30s interval, 60s offline threshold, monotonic sequence numbers, raw error counts per interval.
 - Dead letter monitoring as a health metric.
-- Site Event Logging: 30-day retention, 1GB storage cap, daily purge, paginated queries with keyword search (keyword searches are clamped to `ScadaBridge:SiteEventLog:KeywordSearchMaxWindow`, 7 d; reads run off the writer lock on their own connection). **The cap measures `site_events` + its four indexes, NOT the consolidated LocalDb file** (register row 81, 2026-08-17): whole-file measurement made the trim unwinnable on a replicated node — each batch freed event pages while adding an equal count of oplog rows + tombstones to the same file, so it stopped 2.5x over a 64 MB cap and a node with a down peer could not enforce the cap at all. `dbstat` (the exact per-table measurement) is **NOT compiled into the bundled `SQLitePCLRaw.lib.e_sqlite3`** — probed and pinned by `SqliteCapabilityTests`, so don't reach for it — the footprint is modelled from sampled row width instead (−0.8%…+12.8%, biased high). Consequence: `MaxStorageMb` no longer bounds the data volume alone; LocalDb bookkeeping is separate overhead bounded by `MaxOplogRows`/`TombstoneRetention` and is deliberately not this component's job. The cap trim re-reads the footprint count on batch 1 then every `CapTrimCountRefreshBatches` (10th) batch and vacuums once per tick. Decision: `docs/plans/2026-08-17-site-event-log-cap-semantics-decision.md`.
-- Notification Outbox KPIs are central-computed point-in-time from the `Notifications` table (global + per-source-site): queue depth, stuck count, parked count, delivered-last-interval, oldest-pending age.
+- Site Event Logging: 30-day retention, 1GB storage cap, daily purge, paginated queries with keyword search (keyword searches are clamped to `ScadaBridge:SiteEventLog:KeywordSearchMaxWindow`, 7 d; reads run off the writer lock on their own connection). **The `event_type` and `severity` indexes are composites with `timestamp` (#123)** so a filtered page is read in order and stops at the limit (2.3 s cold → ~1 ms); the trade is ~69 B/row more index, i.e. ~15 % fewer events under the same cap. `SiteEventLogSchema.Apply` drops the old single-column indexes in place on first boot. **The cap measures `site_events` + its four indexes, NOT the consolidated LocalDb file** (register row 81, 2026-08-17): whole-file measurement made the trim unwinnable on a replicated node — each batch freed event pages while adding an equal count of oplog rows + tombstones to the same file, so it stopped 2.5x over a 64 MB cap and a node with a down peer could not enforce the cap at all. `dbstat` (the exact per-table measurement) is **NOT compiled into the bundled `SQLitePCLRaw.lib.e_sqlite3`** — probed and pinned by `SqliteCapabilityTests`, so don't reach for it — the footprint is modelled from sampled row width instead (−0.8%…+12.8%, biased high). Consequence: `MaxStorageMb` no longer bounds the data volume alone; LocalDb bookkeeping is separate overhead bounded by `MaxOplogRows`/`TombstoneRetention` and is deliberately not this component's job. The cap trim re-reads the footprint count on batch 1 then every `CapTrimCountRefreshBatches` (10th) batch and vacuums once per tick. Decision: `docs/plans/2026-08-17-site-event-log-cap-semantics-decision.md`.
+- Notification Outbox KPIs are central-computed point-in-time from the `Notifications` table (global + per-source-site): queue depth, stuck count, parked count, delivered-last-interval, oldest-pending age. Each snapshot runs as three status-equality legs (live `Pending`/`Retrying`, `Parked`, `Delivered` in the interval) — never one aggregation over an OR of statuses, which scanned the clustered index once the table held ~1 M Parked rows and pushed the Health dashboard past its 5 s tile timeout (#122: 0.7–2.6 s → ~0.12 s on the rig). Probe with `sqlcmd -I`: without QUOTED_IDENTIFIER the filtered `IX_Notifications_Delivered` is unusable and timings are wildly wrong.
 - Stuck = `Pending`/`Retrying` older than a configurable age threshold (default 10 min) — display-only (KPI count + row badge), no escalation/alerting.
 - Headline KPI tiles surface on the Health dashboard; a new Central UI Notification Outbox page offers a queryable list with Retry/Discard actions on parked notifications.
 - Site Call Audit KPIs are central-computed point-in-time from the `SiteCalls` table (global + per-site), mirroring the Notification Outbox KPI shape; tiles surface on the Health dashboard alongside a queryable Central UI Site Calls page with Retry/Discard on parked rows.
diff --git a/docs/components/SiteEventLogging.md b/docs/components/SiteEventLogging.md
index 0017f074..59529b68 100644
--- a/docs/components/SiteEventLogging.md
+++ b/docs/components/SiteEventLogging.md
@@ -57,12 +57,16 @@ cmd.CommandText = """
         details TEXT
     );
     CREATE INDEX IF NOT EXISTS idx_events_timestamp ON site_events(timestamp);
-    CREATE INDEX IF NOT EXISTS idx_events_type ON site_events(event_type);
     CREATE INDEX IF NOT EXISTS idx_events_instance ON site_events(instance_id);
-    CREATE INDEX IF NOT EXISTS idx_events_severity ON site_events(severity);
+    DROP INDEX IF EXISTS idx_events_type;
+    DROP INDEX IF EXISTS idx_events_severity;
+    CREATE INDEX IF NOT EXISTS idx_events_type_timestamp ON site_events(event_type, timestamp);
+    CREATE INDEX IF NOT EXISTS idx_events_severity_timestamp ON site_events(severity, timestamp);
     """;
 ```
 
+The `event_type` and `severity` indexes are composites with `timestamp` (#123). Every query orders by `(timestamp, id)` and returns one page, so with a single-column index an equality filter read and sorted every matching row: `severity = Error` over a full 1 GB log took 2.3 s cold and 115 ms warm. The composite index returns rows already in timestamp order and stops at the page limit, in about 1 ms. The cost is about 69 bytes more index per row, which the storage-cap model charges, so a 1 GB cap now holds roughly 15 % fewer events. The two `DROP INDEX` statements migrate a pre-#123 file in place on its first boot; rebuilding the two indexes on a full 1 GB table takes seconds and happens once. Indexes are not part of LocalDb's replicated-table digest, so this change does not stop a mixed-version pair from syncing.
+
 Keyword search (`KeywordFilter`) runs as `LIKE '%…%' ESCAPE '\'` on `message` and `source`. A leading-wildcard `LIKE` cannot use a B-tree index, so keyword-only queries full-scan the table. All other filters (`event_type`, `severity`, `instance_id`, `timestamp`) are covered by the indexes above.
 
 ### Connection lock
diff --git a/docs/plans/2026-08-17-site-event-log-cap-semantics-decision.md b/docs/plans/2026-08-17-site-event-log-cap-semantics-decision.md
index 357ebc6e..3e41c911 100644
--- a/docs/plans/2026-08-17-site-event-log-cap-semantics-decision.md
+++ b/docs/plans/2026-08-17-site-event-log-cap-semantics-decision.md
@@ -171,6 +171,24 @@ stream whose row width changes sharply *in the middle* of the history — neithe
 tail — is mis-sized until the trim reaches it. Self-correcting in practice, since the cap is
 re-evaluated every tick and the head sample advances as the oldest rows go.
 
+### 3.2 Recalibration for the composite indexes (#123, 2026-11-10)
+
+#123 replaced the single-column `event_type` and `severity` indexes with `(event_type, timestamp)`
+and `(severity, timestamp)`. There are still four secondary indexes, so `IndexedColumnCount = 4`
+stands. The sampler's `indexed_bytes` now counts `timestamp` three times: once for its own index
+and once for each composite. The other constants are unchanged.
+
+The model was re-scored with `EstimateSiteEventsSize_TracksTheRealFootprint`:
+
+| Workload | estimated | actual | error | before #123 |
+| --- | ---: | ---: | ---: | ---: |
+| narrow rows, no details | 3 859 626 | 3 551 232 | **+8.7 %** | +8.4 % |
+| typical rows, 200 B details | 4 520 096 | 4 005 888 | **+12.8 %** | +12.8 % |
+| wide rows, 2 KB details | 13 012 048 | 13 078 528 | **−0.5 %** | −0.8 % |
+
+The band and the high bias are unchanged. `SiteEventsSchemaCalibrationTests` pins the new index
+set and the columns each index covers.
+
 ## 4. The wipe-prevention guard, restated
 
 The register asked whether the "size did not decrease" guard should survive. It should, and it
diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/AuditLogRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/AuditLogRepository.cs
index e5d62487..96e6660b 100644
--- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/AuditLogRepository.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/AuditLogRepository.cs
@@ -250,7 +250,18 @@ VALUES
         }
 
         sql.Append("\n) AS v (").Append(CanonicalColumnList).Append(")\n");
-        sql.Append("WHERE NOT EXISTS (SELECT 1 FROM dbo.AuditLog x WHERE x.EventId = v.EventId);");
+        // FORCESEEK (#124): left to itself the optimizer answered the anti-semi-join for
+        // a 256-row packet by SCANNING the EventId index. Under READ COMMITTED a scan
+        // holds its current page's S lock while it requests the next page, so two
+        // concurrent inserts — each holding IX on the pages it had just written —
+        // deadlocked on each other's pages (error 1205; captured graph: two INSERTs,
+        // S/IX page locks on AuditLog). Measured with 4 concurrent writers on a fresh
+        // database (ResidualBench audit-contention): 76 victims in 30 s, 775 rows/s.
+        // One index seek per VALUES row locks only the key it reads, and only while
+        // reading it: 0 victims, 5,214 rows/s. A seek can still wait on another
+        // writer's uncommitted row with the SAME EventId — a genuine duplicate, which
+        // the duplicate-key fallback below already resolves.
+        sql.Append("WHERE NOT EXISTS (SELECT 1 FROM dbo.AuditLog x WITH (FORCESEEK) WHERE x.EventId = v.EventId);");
 
         var conn = _context.Database.GetDbConnection();
         var openedHere = false;
diff --git a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs
index 0bed52e2..5e921617 100644
--- a/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase/Repositories/NotificationOutboxRepository.cs
@@ -428,63 +428,50 @@ VALUES
         return total;
     }
 
+    // ── Outbox KPIs (#122) ────────────────────────────────────────────────────────────
+    //
+    // Every snapshot is computed as THREE status-equality legs instead of one
+    // aggregation over an OR'd predicate:
+    //   * live    — Status IN (Pending, Retrying): queue depth, stuck count, oldest age;
+    //   * parked  — Status = Parked: the parked count;
+    //   * delivered — Status = Delivered AND DeliveredAt >= since: the interval count.
+    // The OR'd pre-filter (the shape before #122) made SQL Server scan the clustered
+    // index: on the rig's 4.35 M-row table (945 k Parked, 3.4 M Delivered) the global
+    // snapshot took 0.7-2.6 s and 76 k logical reads. Split, each leg seeks its own index
+    // — (Status, NextAttemptAt) for live and parked, the filtered IX_Notifications_Delivered
+    // for the interval leg — and the global snapshot takes ~0.12 s (parked count 18 k
+    // reads, live 195, delivered 5). The grouped per-site/per-node snapshots still read
+    // every parked row to group it, but the live and delivered legs no longer ride along.
+
     /// <inheritdoc />
     public async Task<NotificationKpiSnapshot> ComputeKpisAsync(
         DateTimeOffset stuckCutoff, DateTimeOffset deliveredSince, CancellationToken cancellationToken = default)
     {
         var now = DateTimeOffset.UtcNow;
 
-        // One conditional-aggregation pass replaces four sequential COUNT round trips:
-        // each metric is a COUNT(CASE WHEN <predicate> THEN 1 END) over the same scan
-        // (arch-review 04). GroupBy(_ => 1) yields a single group (no rows → no group).
-        //
-        // The aggregation is PREDICATE-RESTRICTED instead of scanning the
-        // whole table. Every KPI here is about the live queue (Pending/Retrying), the
-        // parked backlog, or the last delivery interval; the overwhelming bulk of the
-        // Notifications table is historical Delivered and Discarded rows that
-        // contribute to NONE of them. The pre-filter below is the union of the
-        // metric-contributing predicates, which lets the optimizer seek the status
-        // index for the live/parked legs and the filtered
-        // (DeliveredAt) WHERE Status='Delivered' index for the interval leg,
-        // instead of paying a full scan whose cost grows with retained history.
-        // Mirrors the pre-filter ComputePerSiteKpisAsync/ComputePerNodeKpisAsync
-        // already use — the global snapshot was the odd one out.
-        var counts = await _context.Notifications
-            .Where(n => n.Status == NotificationStatus.Pending
-                || n.Status == NotificationStatus.Retrying
-                || n.Status == NotificationStatus.Parked
-                || (n.Status == NotificationStatus.Delivered
-                    && n.DeliveredAt != null && n.DeliveredAt >= deliveredSince))
+        // Live leg: counts and the oldest CreatedAt in one aggregation over the (small)
+        // non-terminal set. GroupBy(_ => 1) yields a single group, or none when the queue is empty.
+        var live = await LiveQueue()
             .GroupBy(_ => 1)
             .Select(g => new
             {
-                QueueDepth = g.Count(n => n.Status == NotificationStatus.Pending
-                    || n.Status == NotificationStatus.Retrying),
-                StuckCount = g.Count(n => (n.Status == NotificationStatus.Pending
-                        || n.Status == NotificationStatus.Retrying)
-                    && n.CreatedAt < stuckCutoff),
-                ParkedCount = g.Count(n => n.Status == NotificationStatus.Parked),
-                DeliveredLastInterval = g.Count(n => n.Status == NotificationStatus.Delivered
-                    && n.DeliveredAt != null && n.DeliveredAt >= deliveredSince),
+                QueueDepth = g.Count(),
+                StuckCount = g.Count(n => n.CreatedAt < stuckCutoff),
+                OldestCreatedAt = g.Min(n => (DateTimeOffset?)n.CreatedAt),
             })
             .FirstOrDefaultAsync(cancellationToken);
 
-        // Oldest non-terminal CreatedAt — TOP(1) via ordered FirstOrDefault, so the
-        // live queue is never materialized. The nullable cast lets FirstOrDefault
-        // return null (no non-terminal rows) instead of throwing.
-        var oldestCreatedAt = await _context.Notifications
-            .Where(n => n.Status == NotificationStatus.Pending
-                || n.Status == NotificationStatus.Retrying)
-            .OrderBy(n => n.CreatedAt)
-            .Select(n => (DateTimeOffset?)n.CreatedAt)
-            .FirstOrDefaultAsync(cancellationToken);
+        var parkedCount = await _context.Notifications
+            .CountAsync(n => n.Status == NotificationStatus.Parked, cancellationToken);
+
+        var deliveredCount = await DeliveredSince(deliveredSince).CountAsync(cancellationToken);
 
         return new NotificationKpiSnapshot(
-            QueueDepth: counts?.QueueDepth ?? 0,
-            StuckCount: counts?.StuckCount ?? 0,
-            ParkedCount: counts?.ParkedCount ?? 0,
-            DeliveredLastInterval: counts?.DeliveredLastInterval ?? 0,
-            OldestPendingAge: oldestCreatedAt is null ? null : now - oldestCreatedAt.Value);
+            QueueDepth: live?.QueueDepth ?? 0,
+            StuckCount: live?.StuckCount ?? 0,
+            ParkedCount: parkedCount,
+            DeliveredLastInterval: deliveredCount,
+            OldestPendingAge: live?.OldestCreatedAt is null ? null : now - live.OldestCreatedAt.Value);
     }
 
     /// <inheritdoc />
@@ -493,47 +480,31 @@ VALUES
     {
         var now = DateTimeOffset.UtcNow;
 
-        // One grouped conditional-aggregation query per scope: counts AND the oldest
-        // non-terminal CreatedAt (server-side MIN over a CASE, so the live queue is
-        // never materialized) computed in a single scan. Pre-filter to rows that
-        // contribute to at least one metric so a site with only fully-uncounted rows
-        // (e.g. Discarded, or out-of-window Delivered) does not surface as an
-        // all-zero row — preserving the prior union-of-metric-keys semantics.
-        var rows = await _context.Notifications
-            .Where(n => n.Status == NotificationStatus.Pending
-                || n.Status == NotificationStatus.Retrying
-                || n.Status == NotificationStatus.Parked
-                || (n.Status == NotificationStatus.Delivered
-                    && n.DeliveredAt != null && n.DeliveredAt >= deliveredSince))
+        var live = await LiveQueue()
             .GroupBy(n => n.SourceSiteId)
-            .Select(g => new
-            {
-                Site = g.Key,
-                QueueDepth = g.Count(n => n.Status == NotificationStatus.Pending
-                    || n.Status == NotificationStatus.Retrying),
-                StuckCount = g.Count(n => (n.Status == NotificationStatus.Pending
-                        || n.Status == NotificationStatus.Retrying)
-                    && n.CreatedAt < stuckCutoff),
-                ParkedCount = g.Count(n => n.Status == NotificationStatus.Parked),
-                DeliveredLastInterval = g.Count(n => n.Status == NotificationStatus.Delivered
-                    && n.DeliveredAt != null && n.DeliveredAt >= deliveredSince),
-                OldestCreatedAt = g.Min(n => (n.Status == NotificationStatus.Pending
-                        || n.Status == NotificationStatus.Retrying)
-                    ? (DateTimeOffset?)n.CreatedAt
-                    : null),
-            })
+            .Select(g => new LiveRow(
+                g.Key,
+                g.Count(),
+                g.Count(n => n.CreatedAt < stuckCutoff),
+                g.Min(n => (DateTimeOffset?)n.CreatedAt)))
             .ToListAsync(cancellationToken);
 
-        return rows
-            .OrderBy(r => r.Site, StringComparer.Ordinal)
-            .Select(r => new SiteNotificationKpiSnapshot(
-                SourceSiteId: r.Site,
-                QueueDepth: r.QueueDepth,
-                StuckCount: r.StuckCount,
-                ParkedCount: r.ParkedCount,
-                DeliveredLastInterval: r.DeliveredLastInterval,
-                OldestPendingAge: r.OldestCreatedAt is null ? null : now - r.OldestCreatedAt.Value))
-            .ToList();
+        var parked = await _context.Notifications
+            .Where(n => n.Status == NotificationStatus.Parked)
+            .GroupBy(n => n.SourceSiteId)
+            .Select(g => new CountRow(g.Key, g.Count()))
+            .ToListAsync(cancellationToken);
+
+        var delivered = await DeliveredSince(deliveredSince)
+            .GroupBy(n => n.SourceSiteId)
+            .Select(g => new CountRow(g.Key, g.Count()))
+            .ToListAsync(cancellationToken);
+
+        // A key appears only when it contributes to at least one leg, so a site with only
+        // uncounted rows (Discarded, out-of-window Delivered) never surfaces as an all-zero row.
+        return MergeLegs(live, parked, delivered, now,
+            (key, depth, stuck, parkedCount, deliveredCount, age) =>
+                new SiteNotificationKpiSnapshot(key, depth, stuck, parkedCount, deliveredCount, age));
     }
 
     /// <inheritdoc />
@@ -542,44 +513,83 @@ VALUES
     {
         var now = DateTimeOffset.UtcNow;
 
-        // Same single-scan grouped aggregation as ComputePerSiteKpisAsync, keyed by
-        // SourceNode. Rows with a NULL SourceNode (legacy / unstamped) are excluded —
-        // per-node KPIs are only meaningful when node identity is known.
-        var rows = await _context.Notifications
-            .Where(n => n.SourceNode != null
-                && (n.Status == NotificationStatus.Pending
-                    || n.Status == NotificationStatus.Retrying
-                    || n.Status == NotificationStatus.Parked
-                    || (n.Status == NotificationStatus.Delivered
-                        && n.DeliveredAt != null && n.DeliveredAt >= deliveredSince)))
+        // Same three legs as ComputePerSiteKpisAsync, keyed by SourceNode. Rows with a NULL
+        // SourceNode (legacy / unstamped) are excluded — per-node KPIs are only meaningful
+        // when node identity is known.
+        var live = await LiveQueue()
+            .Where(n => n.SourceNode != null)
             .GroupBy(n => n.SourceNode!)
-            .Select(g => new
-            {
-                Node = g.Key,
-                QueueDepth = g.Count(n => n.Status == NotificationStatus.Pending
-                    || n.Status == NotificationStatus.Retrying),
-                StuckCount = g.Count(n => (n.Status == NotificationStatus.Pending
-                        || n.Status == NotificationStatus.Retrying)
-                    && n.CreatedAt < stuckCutoff),
-                ParkedCount = g.Count(n => n.Status == NotificationStatus.Parked),
-                DeliveredLastInterval = g.Count(n => n.Status == NotificationStatus.Delivered
-                    && n.DeliveredAt != null && n.DeliveredAt >= deliveredSince),
-                OldestCreatedAt = g.Min(n => (n.Status == NotificationStatus.Pending
-                        || n.Status == NotificationStatus.Retrying)
-                    ? (DateTimeOffset?)n.CreatedAt
-                    : null),
-            })
+            .Select(g => new LiveRow(
+                g.Key,
+                g.Count(),
+                g.Count(n => n.CreatedAt < stuckCutoff),
+                g.Min(n => (DateTimeOffset?)n.CreatedAt)))
             .ToListAsync(cancellationToken);
 
-        return rows
-            .OrderBy(r => r.Node, StringComparer.Ordinal)
-            .Select(r => new NodeNotificationKpiSnapshot(
-                SourceNode: r.Node,
-                QueueDepth: r.QueueDepth,
-                StuckCount: r.StuckCount,
-                ParkedCount: r.ParkedCount,
-                DeliveredLastInterval: r.DeliveredLastInterval,
-                OldestPendingAge: r.OldestCreatedAt is null ? null : now - r.OldestCreatedAt.Value))
+        var parked = await _context.Notifications
+            .Where(n => n.Status == NotificationStatus.Parked && n.SourceNode != null)
+            .GroupBy(n => n.SourceNode!)
+            .Select(g => new CountRow(g.Key, g.Count()))
+            .ToListAsync(cancellationToken);
+
+        var delivered = await DeliveredSince(deliveredSince)
+            .Where(n => n.SourceNode != null)
+            .GroupBy(n => n.SourceNode!)
+            .Select(g => new CountRow(g.Key, g.Count()))
+            .ToListAsync(cancellationToken);
+
+        return MergeLegs(live, parked, delivered, now,
+            (key, depth, stuck, parkedCount, deliveredCount, age) =>
+                new NodeNotificationKpiSnapshot(key, depth, stuck, parkedCount, deliveredCount, age));
+    }
+
+    /// <summary>The non-terminal queue: <c>Status IN (Pending, Retrying)</c>, two index seek ranges.</summary>
+    private IQueryable<Notification> LiveQueue() =>
+        _context.Notifications.Where(n => n.Status == NotificationStatus.Pending
+            || n.Status == NotificationStatus.Retrying);
+
+    /// <summary>
+    /// Delivered rows inside the KPI interval. The literal <c>Status = 'Delivered'</c> plus the
+    /// <c>DeliveredAt</c> range matches the filtered <c>IX_Notifications_Delivered</c> index.
+    /// </summary>
+    private IQueryable<Notification> DeliveredSince(DateTimeOffset deliveredSince) =>
+        _context.Notifications.Where(n => n.Status == NotificationStatus.Delivered
+            && n.DeliveredAt != null && n.DeliveredAt >= deliveredSince);
+
+    private sealed record LiveRow(string Key, int QueueDepth, int StuckCount, DateTimeOffset? OldestCreatedAt);
+
+    private sealed record CountRow(string Key, int Count);
+
+    /// <summary>
+    /// Joins the three per-key legs into one snapshot per key, ordered ordinally by key.
+    /// </summary>
+    private static List<TSnapshot> MergeLegs<TSnapshot>(
+        IReadOnlyList<LiveRow> live,
+        IReadOnlyList<CountRow> parked,
+        IReadOnlyList<CountRow> delivered,
+        DateTimeOffset now,
+        Func<string, int, int, int, int, TimeSpan?, TSnapshot> create)
+    {
+        var liveByKey = live.ToDictionary(r => r.Key, StringComparer.Ordinal);
+        var parkedByKey = parked.ToDictionary(r => r.Key, r => r.Count, StringComparer.Ordinal);
+        var deliveredByKey = delivered.ToDictionary(r => r.Key, r => r.Count, StringComparer.Ordinal);
+
+        return liveByKey.Keys
+            .Concat(parkedByKey.Keys)
+            .Concat(deliveredByKey.Keys)
+            .Distinct(StringComparer.Ordinal)
+            .OrderBy(k => k, StringComparer.Ordinal)
+            .Select(key =>
+            {
+                liveByKey.TryGetValue(key, out var l);
+                return create(
+                    key,
+                    l?.QueueDepth ?? 0,
+                    l?.StuckCount ?? 0,
+                    parkedByKey.GetValueOrDefault(key),
+                    deliveredByKey.GetValueOrDefault(key),
+                    l?.OldestCreatedAt is null ? null : now - l.OldestCreatedAt.Value);
+            })
             .ToList();
     }
 
diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/EventLogPurgeService.cs b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/EventLogPurgeService.cs
index 66ae7cb1..a9d12d7a 100644
--- a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/EventLogPurgeService.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/EventLogPurgeService.cs
@@ -89,7 +89,11 @@ public class EventLogPurgeService : BackgroundService
     /// </summary>
     private const double TablePageFillFactor = 0.78;
 
-    /// <summary>Number of indexes on <c>site_events</c> (timestamp, event_type, instance_id, severity).</summary>
+    /// <summary>
+    /// Number of secondary indexes on <c>site_events</c>: timestamp, instance_id,
+    /// (event_type, timestamp) and (severity, timestamp). The sampled <c>indexed_bytes</c>
+    /// charges every covered column once per index, hence <c>3 * octet_length(timestamp)</c>.
+    /// </summary>
     private const int IndexedColumnCount = 4;
 
     /// <summary>Per-entry bookkeeping in an index B-tree leaf: key-length varint, rowid and cell pointer.</summary>
@@ -680,8 +684,8 @@ public class EventLogPurgeService : BackgroundService
     /// <para>
     /// <b>The model.</b> Rows are sampled from both ends of the table (<see cref="SizeSampleRows"/>
     /// from each — the head is what the oldest-first trim will actually delete, the tail catches a
-    /// workload whose event shape has drifted) to get an average payload width and an average
-    /// width across the four indexed columns. Table cost is then computed as whole pages —
+    /// workload whose event shape has drifted) to get an average payload width and the average
+    /// summed width of the columns each of the four indexes covers. Table cost is then computed as whole pages —
     /// <c>ceil(rows / rowsPerPage) × pageSize</c> — rather than as a flat bytes-per-row rate,
     /// because that is the term that dominates once rows approach half a page: rows too wide to
     /// pack two-per-page waste the remainder, which a linear model underestimates by ~40 %.
@@ -730,7 +734,7 @@ public class EventLogPurgeService : BackgroundService
                                     + octet_length(severity) + COALESCE(octet_length(instance_id), 0)
                                     + octet_length(source) + octet_length(message)
                                     + COALESCE(octet_length(details), 0) AS payload_bytes,
-                                octet_length(timestamp) + octet_length(event_type)
+                                (3 * octet_length(timestamp)) + octet_length(event_type)
                                     + octet_length(severity) + COALESCE(octet_length(instance_id), 0)
                                     AS indexed_bytes
                             FROM site_events LIMIT {SizeSampleRows}
@@ -742,7 +746,7 @@ public class EventLogPurgeService : BackgroundService
                                     + octet_length(severity) + COALESCE(octet_length(instance_id), 0)
                                     + octet_length(source) + octet_length(message)
                                     + COALESCE(octet_length(details), 0) AS payload_bytes,
-                                octet_length(timestamp) + octet_length(event_type)
+                                (3 * octet_length(timestamp)) + octet_length(event_type)
                                     + octet_length(severity) + COALESCE(octet_length(instance_id), 0)
                                     AS indexed_bytes
                             FROM site_events ORDER BY rowid DESC LIMIT {SizeSampleRows}
diff --git a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogSchema.cs b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogSchema.cs
index 0af60396..04581b72 100644
--- a/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogSchema.cs
+++ b/src/ZB.MOM.WW.ScadaBridge.SiteEventLogging/SiteEventLogSchema.cs
@@ -67,10 +67,24 @@ public static class SiteEventLogSchema
                 details TEXT
             );
             CREATE INDEX IF NOT EXISTS idx_events_timestamp ON site_events(timestamp);
-            CREATE INDEX IF NOT EXISTS idx_events_type ON site_events(event_type);
             CREATE INDEX IF NOT EXISTS idx_events_instance ON site_events(instance_id);
-            CREATE INDEX IF NOT EXISTS idx_events_severity ON site_events(severity);
+            DROP INDEX IF EXISTS idx_events_type;
+            DROP INDEX IF EXISTS idx_events_severity;
+            CREATE INDEX IF NOT EXISTS idx_events_type_timestamp ON site_events(event_type, timestamp);
+            CREATE INDEX IF NOT EXISTS idx_events_severity_timestamp ON site_events(severity, timestamp);
             """;
+        // The type and severity indexes are composites with timestamp (#123). The query
+        // service always orders by (timestamp, id) and returns one page, so a
+        // single-column index made an equality filter read EVERY matching row and sort
+        // it before returning 500 (severity=Error over a full 1 GB log: 2.3 s cold,
+        // 115 ms warm). The composite returns the rows already in timestamp order and
+        // stops at the page limit (~1 ms). The cost is ~69 bytes per row more index,
+        // which the storage-cap size model charges (EventLogPurgeService sampling SQL).
+        // The DROPs migrate a pre-#123 file in place on its first boot; dropping and
+        // building an index on a 1 GB table takes seconds and runs once. Indexes are not
+        // part of LocalDb's replicated-table digest, so a pair may upgrade one node at a
+        // time as far as this change is concerned.
+        //
         // The query service also supports keyword search via leading-wildcard
         // LIKE on message/source. A leading-wildcard LIKE cannot use a B-tree
         // index, so that path intentionally full-scans; severity/event_type/
diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/NotificationOutboxRepositoryKpiQueryShapeTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/NotificationOutboxRepositoryKpiQueryShapeTests.cs
index fd79c4f8..98dc500d 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/NotificationOutboxRepositoryKpiQueryShapeTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/NotificationOutboxRepositoryKpiQueryShapeTests.cs
@@ -8,9 +8,11 @@ namespace ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests;
 
 /// <summary>
 /// Query-shape coverage for the Notification Outbox KPI snapshots (arch-review 04).
-/// Each snapshot must be a single-query conditional aggregation with a
-/// bounded oldest-pending — the old shape issued 5-7 sequential COUNT round trips
-/// and materialized the whole live queue to find the oldest row.
+/// Each snapshot is at most three status-equality legs (live queue, parked, delivered
+/// in the interval) — the shape before arch-review 04 issued 5-7 sequential COUNT round
+/// trips and materialized the whole live queue to find the oldest row, and the single
+/// OR'd aggregation that replaced it scanned the clustered index once the table held
+/// ~1 M Parked rows (#122).
 /// </summary>
 public class NotificationOutboxRepositoryKpiQueryShapeTests
 {
@@ -25,10 +27,10 @@ public class NotificationOutboxRepositoryKpiQueryShapeTests
             SourceNode = sourceNode,
         };
 
-    /// <summary>Verifies that all three KPI snapshots issue at most two queries and bound the oldest-pending lookup.</summary>
+    /// <summary>Verifies that all three KPI snapshots issue at most three seekable legs and never OR the parked and live predicates.</summary>
     /// <returns>A task that represents the asynchronous operation.</returns>
     [Fact]
-    public async Task AllThreeKpiSnapshots_IssueAtMostTwoQueries_AndBoundTheOldestLookup()
+    public async Task AllThreeKpiSnapshots_IssueAtMostThreeStatusLegs_NeverOneOrdScan()
     {
         var counter = new SelectCountingInterceptor();
         await using var ctx = SqliteTestHelper.CreateInMemoryContext(counter);
@@ -45,13 +47,12 @@ public class NotificationOutboxRepositoryKpiQueryShapeTests
         var stuckCutoff = now.AddMinutes(-5);
         var deliveredSince = now.AddMinutes(-60);
 
-        // Global: one aggregation query + one bounded oldest query.
+        // Global: three legs — live (counts + MIN), parked count, delivered count.
         counter.Reset();
         var global = await repo.ComputeKpisAsync(stuckCutoff, deliveredSince);
-        Assert.True(counter.Count <= 2,
-            $"ComputeKpisAsync issued {counter.Count} queries against Notifications; expected <= 2");
-        // The oldest lookup must be bounded (LIMIT/TOP), never a full non-terminal SELECT.
-        Assert.Contains(counter.Commands, sql => sql.Contains("LIMIT", StringComparison.OrdinalIgnoreCase));
+        Assert.True(counter.Count <= 3,
+            $"ComputeKpisAsync issued {counter.Count} queries against Notifications; expected <= 3");
+        AssertNoLegMixesParkedWithLive(counter.Commands);
         // The aggregation must be PREDICATE-RESTRICTED, not an unrestricted
         // scan. Every KPI here concerns the live queue, the parked backlog or the
         // last delivery interval; historical Delivered/Discarded rows — which are
@@ -68,8 +69,9 @@ public class NotificationOutboxRepositoryKpiQueryShapeTests
         // Per-site: a single grouped conditional-aggregation query (counts + MIN).
         counter.Reset();
         var perSite = await repo.ComputePerSiteKpisAsync(stuckCutoff, deliveredSince);
-        Assert.True(counter.Count <= 2,
-            $"ComputePerSiteKpisAsync issued {counter.Count} queries against Notifications; expected <= 2");
+        Assert.True(counter.Count <= 3,
+            $"ComputePerSiteKpisAsync issued {counter.Count} queries against Notifications; expected <= 3");
+        AssertNoLegMixesParkedWithLive(counter.Commands);
         var siteA = Assert.Single(perSite);
         Assert.Equal(3, siteA.QueueDepth);
         Assert.NotNull(siteA.OldestPendingAge);
@@ -77,13 +79,49 @@ public class NotificationOutboxRepositoryKpiQueryShapeTests
         // Per-node: same single grouped aggregation, keyed by SourceNode.
         counter.Reset();
         var perNode = await repo.ComputePerNodeKpisAsync(stuckCutoff, deliveredSince);
-        Assert.True(counter.Count <= 2,
-            $"ComputePerNodeKpisAsync issued {counter.Count} queries against Notifications; expected <= 2");
+        Assert.True(counter.Count <= 3,
+            $"ComputePerNodeKpisAsync issued {counter.Count} queries against Notifications; expected <= 3");
+        AssertNoLegMixesParkedWithLive(counter.Commands);
         var nodeA = Assert.Single(perNode);
         Assert.Equal(3, nodeA.QueueDepth);
         Assert.NotNull(nodeA.OldestPendingAge);
     }
 
+    /// <summary>Verifies the three legs merge per key and a key with only one leg still surfaces.</summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [Fact]
+    public async Task PerSiteKpis_MergeLegsPerKey()
+    {
+        await using var ctx = SqliteTestHelper.CreateInMemoryContext(new SelectCountingInterceptor());
+        var now = DateTimeOffset.UtcNow;
+        ctx.Notifications.Add(NewNotification("plant-a", "node-a", NotificationStatus.Pending, now.AddMinutes(-30)));
+        ctx.Notifications.Add(NewNotification("plant-a", "node-a", NotificationStatus.Parked, now.AddMinutes(-30)));
+        ctx.Notifications.Add(NewNotification("plant-b", "node-b", NotificationStatus.Parked, now.AddMinutes(-30)));
+        ctx.Notifications.Add(NewNotification("plant-c", "node-a", NotificationStatus.Delivered, now.AddMinutes(-30), now.AddMinutes(-1)));
+        ctx.Notifications.Add(NewNotification("plant-d", "node-b", NotificationStatus.Discarded, now.AddMinutes(-30)));
+        await ctx.SaveChangesAsync();
+
+        var repo = new NotificationOutboxRepository(ctx);
+        var sites = await repo.ComputePerSiteKpisAsync(now.AddMinutes(-5), now.AddMinutes(-60));
+
+        Assert.Equal(new[] { "plant-a", "plant-b", "plant-c" }, sites.Select(s => s.SourceSiteId));
+        Assert.Equal((1, 1, 1, 0), (sites[0].QueueDepth, sites[0].StuckCount, sites[0].ParkedCount, sites[0].DeliveredLastInterval));
+        Assert.NotNull(sites[0].OldestPendingAge);
+        Assert.Equal((0, 1, 0), (sites[1].QueueDepth, sites[1].ParkedCount, sites[1].DeliveredLastInterval));
+        Assert.Null(sites[1].OldestPendingAge);
+        Assert.Equal((0, 0, 1), (sites[2].QueueDepth, sites[2].ParkedCount, sites[2].DeliveredLastInterval));
+
+        var global = await repo.ComputeKpisAsync(now.AddMinutes(-5), now.AddMinutes(-60));
+        Assert.Equal((1, 1, 2, 1), (global.QueueDepth, global.StuckCount, global.ParkedCount, global.DeliveredLastInterval));
+    }
+
+    // An OR across the parked and live predicates is what made SQL Server scan the
+    // clustered index on a large table (#122); each leg must name one status set.
+    private static void AssertNoLegMixesParkedWithLive(IEnumerable<string> commands) =>
+        Assert.All(commands, sql => Assert.False(
+            sql.Contains("'Parked'", StringComparison.Ordinal) && sql.Contains("'Pending'", StringComparison.Ordinal),
+            $"a KPI leg ORs Parked with the live queue:\n{sql}"));
+
     /// <summary>
     /// Records the text of every reader command touching the Notifications table so a
     /// test can count the round trips a KPI snapshot makes and inspect their shape.
diff --git a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/AuditLogRepositoryTests.cs b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/AuditLogRepositoryTests.cs
index 5a03b193..ea7e59b3 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/AuditLogRepositoryTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.ConfigurationDatabase.Tests/Repositories/AuditLogRepositoryTests.cs
@@ -635,6 +635,65 @@ public class AuditLogRepositoryTests : IClassFixture<MsSqlMigrationFixture>
             rows.Select(r => r.EventId).OrderBy(g => g).ToArray());
     }
 
+    /// <summary>
+    /// Verifies that concurrent batch inserts do not deadlock each other (#124). Before the
+    /// probe was forced to seek, four writers of 256-row packets lost most packets to error
+    /// 1205 within seconds: the anti-semi-join scanned the EventId index and the writers
+    /// deadlocked on each other's S/IX page locks.
+    /// </summary>
+    /// <returns>A task that represents the asynchronous operation.</returns>
+    [SkippableFact]
+    public async Task InsertManyIfNotExistsAsync_ConcurrentWriters_DoNotDeadlock()
+    {
+        Skip.IfNot(_fixture.Available, _fixture.SkipReason);
+
+        const int writers = 4, packetsPerWriter = 15, packetSize = 256, seedPackets = 40;
+        var siteId = NewSiteId();
+        var baseTime = new DateTime(2026, 5, 22, 8, 0, 0, DateTimeKind.Utc);
+
+        // Every packet spans the SAME time window, as live telemetry from several sites
+        // does, so the writers' rows interleave on the same clustered pages. The seed
+        // grows the index first: the deadlock rate rose with table size (bench, 4
+        // writers: 3 victims in 65 packets on an empty table, 17 in 52 at 5 k rows).
+        List<IReadOnlyList<AuditEvent>> Packets(int count, int offset) =>
+            Enumerable.Range(offset, count)
+                .Select(p => (IReadOnlyList<AuditEvent>)Enumerable.Range(0, packetSize)
+                    .Select(i => NewEvent(siteId, occurredAtUtc: baseTime.AddMilliseconds(i).AddTicks(p)))
+                    .ToList())
+                .ToList();
+
+        foreach (var seed in Packets(seedPackets, 0))
+        {
+            await using var context = CreateContext();
+            await new AuditLogRepository(context).InsertManyIfNotExistsAsync(seed);
+        }
+
+        var packets = Packets(writers * packetsPerWriter, seedPackets);
+
+        var deadlocks = 0;
+        await Task.WhenAll(Enumerable.Range(0, writers).Select(w => Task.Run(async () =>
+        {
+            for (var p = w; p < packets.Count; p += writers)
+            {
+                try
+                {
+                    await using var context = CreateContext();
+                    await new AuditLogRepository(context).InsertManyIfNotExistsAsync(packets[p]);
+                }
+                catch (SqlException ex) when (ex.Number == 1205)
+                {
+                    Interlocked.Increment(ref deadlocks);
+                }
+            }
+        })));
+
+        Assert.Equal(0, deadlocks);
+
+        await using var readContext = CreateContext();
+        var count = await readContext.Set<AuditLogRow>().CountAsync(e => e.SourceSiteId == siteId);
+        Assert.Equal((seedPackets + (writers * packetsPerWriter)) * packetSize, count);
+    }
+
     /// <summary>Verifies that duplicate event ids within one batch collapse to a single row.</summary>
     /// <returns>A task that represents the asynchronous operation.</returns>
     [SkippableFact]
diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SchemaIndexTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SchemaIndexTests.cs
index 92111d1c..86199a25 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SchemaIndexTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SchemaIndexTests.cs
@@ -47,7 +47,7 @@ public class SchemaIndexTests : IDisposable
         using var reader = cmd.ExecuteReader();
         while (reader.Read()) indexes.Add(reader.GetString(0));
 
-        Assert.Contains("idx_events_severity", indexes);
+        Assert.Contains("idx_events_severity_timestamp", indexes);
     }
 
     /// <summary>Verifies that severity filtered query, when uses index, not full scan.</summary>
@@ -70,7 +70,75 @@ public class SchemaIndexTests : IDisposable
         }
 
         var planText = plan.ToString();
-        Assert.Contains("idx_events_severity", planText);
+        Assert.Contains("idx_events_severity_timestamp", planText);
         Assert.DoesNotContain("SCAN site_events", planText);
     }
+
+    /// <summary>
+    /// Verifies that the query service's own page shape — an equality filter, ordered by
+    /// (timestamp, id), limited to a page — is answered in index order rather than by
+    /// sorting every matching row (#123).
+    /// </summary>
+    /// <param name="column">The equality-filtered column.</param>
+    /// <param name="expectedIndex">The composite index expected to answer it.</param>
+    [Theory]
+    [InlineData("severity", "idx_events_severity_timestamp")]
+    [InlineData("event_type", "idx_events_type_timestamp")]
+    public void EqualityFilteredPage_ReadsInTimestampOrder_WithoutSortingEveryMatch(string column, string expectedIndex)
+    {
+        foreach (var window in new[] { "", "AND timestamp >= '2026-01-01' " })
+        {
+            using var cmd = _verifyConnection.CreateCommand();
+            cmd.CommandText =
+                $"EXPLAIN QUERY PLAN SELECT id FROM site_events WHERE {column} = 'x' {window}" +
+                "ORDER BY timestamp ASC, id ASC LIMIT 501";
+            var plan = new System.Text.StringBuilder();
+            using var reader = cmd.ExecuteReader();
+            while (reader.Read())
+                plan.Append(reader.GetString(reader.GetOrdinal("detail"))).Append('\n');
+
+            var planText = plan.ToString();
+            Assert.Contains(expectedIndex, planText);
+            // "USE TEMP B-TREE FOR LAST TERM OF ORDER BY" (ties on timestamp, broken by id)
+            // is fine; a whole-ORDER-BY temp b-tree means every match is read and sorted.
+            Assert.DoesNotContain("USE TEMP B-TREE FOR ORDER BY", planText);
+        }
+    }
+
+    /// <summary>Verifies that a pre-#123 file loses its single-column type/severity indexes on the next schema apply.</summary>
+    [Fact]
+    public void Apply_OnAPre123File_ReplacesTheSingleColumnIndexes()
+    {
+        using var connection = new SqliteConnection("Data Source=:memory:");
+        connection.Open();
+        using (var cmd = connection.CreateCommand())
+        {
+            cmd.CommandText = """
+                CREATE TABLE site_events (id TEXT NOT NULL PRIMARY KEY, timestamp TEXT NOT NULL,
+                    event_type TEXT NOT NULL, severity TEXT NOT NULL, instance_id TEXT,
+                    source TEXT NOT NULL, message TEXT NOT NULL, details TEXT);
+                CREATE INDEX idx_events_timestamp ON site_events(timestamp);
+                CREATE INDEX idx_events_type ON site_events(event_type);
+                CREATE INDEX idx_events_instance ON site_events(instance_id);
+                CREATE INDEX idx_events_severity ON site_events(severity);
+                INSERT INTO site_events VALUES ('a', '2026-01-01T00:00:00.0000000+00:00', 'alarm', 'Info', NULL, 's', 'm', NULL);
+                """;
+            cmd.ExecuteNonQuery();
+        }
+
+        SiteEventLogSchema.Apply(connection);
+
+        using var list = connection.CreateCommand();
+        list.CommandText = "SELECT name FROM pragma_index_list('site_events') WHERE origin = 'c' ORDER BY name";
+        var indexes = new List<string>();
+        using (var reader = list.ExecuteReader())
+            while (reader.Read()) indexes.Add(reader.GetString(0));
+
+        Assert.Equal(
+            ["idx_events_instance", "idx_events_severity_timestamp", "idx_events_timestamp", "idx_events_type_timestamp"],
+            indexes);
+        using var count = connection.CreateCommand();
+        count.CommandText = "SELECT COUNT(*) FROM site_events INDEXED BY idx_events_severity_timestamp WHERE severity = 'Info'";
+        Assert.Equal(1L, (long)count.ExecuteScalar()!);
+    }
 }
diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLogSchemaTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLogSchemaTests.cs
index 18e54d12..86d37bee 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLogSchemaTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventLogSchemaTests.cs
@@ -121,7 +121,7 @@ public class SiteEventLogSchemaTests
         while (reader.Read()) indexes.Add(reader.GetString(0));
 
         Assert.Equal(
-            ["idx_events_instance", "idx_events_severity", "idx_events_timestamp", "idx_events_type"],
+            ["idx_events_instance", "idx_events_severity_timestamp", "idx_events_timestamp", "idx_events_type_timestamp"],
             indexes);
     }
 
diff --git a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventsSchemaCalibrationTests.cs b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventsSchemaCalibrationTests.cs
index f926a0b7..e11b3b45 100644
--- a/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventsSchemaCalibrationTests.cs
+++ b/tests/ZB.MOM.WW.ScadaBridge.SiteEventLogging.Tests/SiteEventsSchemaCalibrationTests.cs
@@ -76,35 +76,36 @@ public class SiteEventsSchemaCalibrationTests
             indexes.SequenceEqual(
                 [
                     "idx_events_instance",
-                    "idx_events_severity",
+                    "idx_events_severity_timestamp",
                     "idx_events_timestamp",
-                    "idx_events_type",
+                    "idx_events_type_timestamp",
                     "sqlite_autoindex_site_events_1",
                 ]),
             $"{RecalibrationNotice} Actual indexes: {string.Join(", ", indexes)}");
     }
 
-    /// <summary>Verifies that each index covers exactly the column the sampler sums.</summary>
+    /// <summary>Verifies that each index covers exactly the columns the sampler sums for it.</summary>
     /// <param name="indexName">The index name.</param>
-    /// <param name="expectedColumn">The column the index is expected to cover.</param>
+    /// <param name="expectedColumns">The columns the index is expected to cover, comma-separated, in key order.</param>
     [Theory]
     [InlineData("idx_events_timestamp", "timestamp")]
-    [InlineData("idx_events_type", "event_type")]
     [InlineData("idx_events_instance", "instance_id")]
-    [InlineData("idx_events_severity", "severity")]
-    public void EachIndex_CoversExactlyTheColumnTheSamplerSums(string indexName, string expectedColumn)
+    [InlineData("idx_events_type_timestamp", "event_type,timestamp")]
+    [InlineData("idx_events_severity_timestamp", "severity,timestamp")]
+    public void EachIndex_CoversExactlyTheColumnsTheSamplerSums(string indexName, string expectedColumns)
     {
-        // The sampler's indexed_bytes expression sums octet_length(timestamp) +
-        // octet_length(event_type) + octet_length(severity) + octet_length(instance_id) —
-        // one column per index. Widening an index to a composite (or re-pointing it) would
-        // change the real entry width without changing what is sampled.
+        // The sampler's indexed_bytes expression sums 3 * octet_length(timestamp) +
+        // octet_length(event_type) + octet_length(severity) + octet_length(instance_id):
+        // timestamp once for its own index and once for each of the two composites (#123).
+        // Widening or re-pointing an index changes the real entry width without changing
+        // what is sampled.
         using var connection = OpenWithSchema();
 
         var columns = ReadStrings(
-            connection, $"SELECT name FROM pragma_index_info('{indexName}')");
+            connection, $"SELECT name FROM pragma_index_info('{indexName}') ORDER BY seqno");
 
         Assert.True(
-            columns.SequenceEqual([expectedColumn]),
+            columns.SequenceEqual(expectedColumns.Split(',')),
             $"{RecalibrationNotice} Index {indexName} covers: {string.Join(", ", columns)}");
     }
 }