ADR 0154 increment B: synchronous captured-downstream reply (reply_from)

docs/PHI.md

@@ -908,7 +908,7 @@ with materially different PHI profiles, so they get their own rows; stream 4 is
 | **3. `messagefoundry.audit` off-box tee** (sub-stream of 1) | one JSON object per **committed** `audit_log` row: `event`/`ts`/`action`/`actor`/`channel_id`/`client`/`detail` | JSON | emitted after the row is durably committed and **outside** the store write lock; rides stream 1's handlers | shipping audit evidence to a SIEM so it survives a host compromise | inherits stream 1's | inherits stream 1's | `detail` is passed through the `safe_text` PHI chokepoint **before** it leaves the process; `client` is forwarded verbatim as a discrete field so a SIEM can index it. Best-effort: a logging failure is caught, never raised into the audit write. **Pinned to `INFO`** — it is emitted even at `[logging].level = WARNING` |
 | **4. Off-box syslog/SIEM forwarder** — the shared **transport** for 1–3 | a copy of every record from 1–3 | `forward_format`, default **JSON** (independent of the stdout format) | the operator's collector (`forward_host`/`_port`) | off-box evidence retention / SIEM correlation | **default-on when a collector is named.** Transport: `udp` (default) / `tcp` / **`tls`** (RFC 5425, CA-anchored, verified by default). `serve` gates the hop on the shared posture gradient before the handler is installed: verified TLS ungated; otherwise loopback / attested / synthetic ALLOW, non-enforcing PHI WARN, **enforcing PHI REFUSE (exit 2)** | the collector's, not the engine's | the identical three filters are installed on this handler, so the forwarded copy is PHI-redacted — but it still carries usernames, connection names, message ids, client addresses and the audit chain. That is the engine's own stated reason for gating the hop |
 | **5. `audit_log` table** (SQLite, Postgres, SQL Server) | who / what / **where-from** / when of auth + PHI *access* and admin actions — plus, when the opt-in `[security].audit_all_authorization_decisions` is on (**default `false`**; the internal field it desugars to is `audit_all_authz`, whose old `[diagnostics]` TOML spelling is **refused at load** — ADR 0118), an `authz` row for **every** authorization decision including successes, which multiplies this stream's volume — `actor`, `action`, `channel_id`, `client`, `detail`, `row_hash` | JSON `detail`; **tamper-evident hash chain** over `prev_hash` + the row (the `client` address is **inside** the chained payload — ADR 0150) | the store database | HIPAA §164.312(b) audit controls; incident response; `verify_audit_chain` integrity checks | `GET /audit` requires **`audit:read`**; `GET /audit/export` requires the separate **`audit:export`** and streams CSV with formula-injection neutralisation, recording its own `audit.export` row *before* streaming; `GET /me/security-events` is a per-user view of the same table | **`[retention].audit_days` is reserved and NOT enforced — keep-forever by design** (deleting rows would break the chain; HIPAA expects ~6 years) | `detail` is stored **in the clear** (it is not a cipher-covered column): its protection is that writers only ever store filter shapes, counts and ids — never bodies or credentials — plus the store ACL and the volume layer |
-| **6. `message_events` table** | the per-message disposition timeline — the **complete** vocabulary is `received`, `routed`, `unrouted`, `filtered`, `transformed`, `delivered`, `failed`, `dead`, `error`, `replayed`, `resent`, `reingressed`, `passthrough`, `passthrough_dropped`, `cancelled`, `edit_resend`, `edit_resubmit`, `viewed`, `not_deployed` (CI asserts this list against the engine's own `MESSAGE_EVENT_KINDS`). `[diagnostics].message_events` can thin the set, but never below the compliance floor `viewed` / `dead` / `error` / `failed` / `not_deployed` | rows: `message_id`, `ts`, `event`, `destination`, `detail` | the store database | operator timeline on the message-detail view; the `viewed` row is the HIPAA PHI-access record | `GET /messages/{id}` under **`messages:view_raw`** + `require_phi_read`; the read itself writes a `viewed` event **and** a `message_view` audit row | no dedicated window — `purge_message_bodies` sets `message_events.detail` to `NULL` in the same transaction that blanks the body, so it inherits `[retention].messages_days` | `detail` is `safe_text()`-scrubbed **then** cipher-encrypted (AAD `("message_events","detail",message_id,ts,event)`). Verbosity gate `[diagnostics].message_events` = `all` (default) / `errors` / `off`, with a **compliance floor that can never be thinned**: `viewed`, `dead`, `error`, `failed`, `not_deployed` are retained at every level |
+| **6. `message_events` table** | the per-message disposition timeline — the **complete** vocabulary is `received`, `routed`, `unrouted`, `filtered`, `transformed`, `delivered`, `failed`, `dead`, `error`, `replayed`, `resent`, `reingressed`, `passthrough`, `passthrough_dropped`, `cancelled`, `edit_resend`, `edit_resubmit`, `viewed`, `not_deployed`, and the ADR 0154 synchronous-reply pair `reply_returned` / `reply_timeout` (names, counts and `waited_ms` only — **never** a fragment of the partner's reply body) (CI asserts this list against the engine's own `MESSAGE_EVENT_KINDS`). `[diagnostics].message_events` can thin the set, but never below the compliance floor `viewed` / `dead` / `error` / `failed` / `not_deployed` / `reply_timeout` | rows: `message_id`, `ts`, `event`, `destination`, `detail` | the store database | operator timeline on the message-detail view; the `viewed` row is the HIPAA PHI-access record | `GET /messages/{id}` under **`messages:view_raw`** + `require_phi_read`; the read itself writes a `viewed` event **and** a `message_view` audit row | no dedicated window — `purge_message_bodies` sets `message_events.detail` to `NULL` in the same transaction that blanks the body, so it inherits `[retention].messages_days` | `detail` is `safe_text()`-scrubbed **then** cipher-encrypted (AAD `("message_events","detail",message_id,ts,event)`). Verbosity gate `[diagnostics].message_events` = `all` (default) / `errors` / `off`, with a **compliance floor that can never be thinned**: `viewed`, `dead`, `error`, `failed`, `not_deployed`, `reply_timeout` are retained at every level (`reply_timeout` is the one row that explains a "we called you and got a 504" complaint, so an instance that thinned its logs would lose exactly the record it is later asked for) |
 | **7. `connection_event` table — DEFAULT ON** (`[diagnostics].connection_events = true`) | transport/lifecycle events per connection: `established`, `closed` (reason `eof` or `idle_timeout` — no path produces any other), `idle_timeout`, `at_capacity`, `peer_not_allowlisted`, `frame_oversize`, `framing_error`, `peer_reset`, the inbound-HTTP intake-auth refusals `intake_auth_failed` / `auth_subject_denied` / `auth_rate_limited` (ADR 0154 D6 — peer address and mode only; **never** the credential, a prefix of it, or its length. Each of these also writes a tamper-evident audit-log row — the copy that survives an operator turning this diagnostics stream off), plus the runner's `connection_lost` / `connection_restored`. That is the whole vocabulary, asserted in CI against the literal emit call sites in `transports/` and the pipeline runner **and** cross-checked against the console's own filter tuple. The MLLP, raw-TCP and HTTP listeners emit these; the **DICOM inbound C-STORE SCP** and the **`ISA`/`IEA`-framed X12 inbound** emit none — the runner injects the sink onto **every** source (`wiring_runner.py`, over the base-class `on_connection_event` field), so both connectors *have* the wiring and simply never call it — so this stream covers those three listeners plus the runner's outbound-lane transitions — not literally every connection. An X12 feed's connects, allow-list refusals and at-capacity refusals are therefore **absent** from this stream | rows: `ts`, `connection`, `transport`, `direction`, `kind`, `peer_host`, `message_id` (correlation hint), `reason` | the store database, **all three backends** | Corepoint-style transport diagnostics — "did the sender connect, and why did it drop" | `GET /events` and `GET /connections/{name}/events` under **`monitoring:read`** (**not** a PHI permission) with per-channel RBAC — an out-of-scope `connection=` is 403'd *and* audited — server-clamped to ≤1000 rows | `[retention].connection_event_retention_hours` (its own **hours** window); 0 inherits `[retention].messages_days`; both 0 = keep forever. Plain age `DELETE` (metadata-only) | **`reason` is free text that can carry sensitive fragments.** Defended twice — `safe_exc()` at the source, `safe_text(reason)[:200]` at the store — then cipher-encrypted (AAD `("connection_event","reason",connection,ts,kind)`). Every other column is config metadata; the table is documented **metadata-only** — never a frame, body or HL7 field value. Writes are a pure side observer: a bounded in-memory queue drained by a background task outside any handoff transaction, so a flood can never block a listener or pin a message disposition |
 | **8. `alert_instance` table — default on wherever an `[alerts]` notifier exists** | resolvable operator alerts: `connection_stopped`, `queue_buildup`, `lane_stuck`, `message_stall`, `saturation`, `connection_error`, `content_match`, `storage_threshold`, `cert_expiry`, `secret_rotation`, `bootstrap_admin_expiring` (the UNCLAIMED first-run bootstrap admin nearing its auto-disable deadline — ASVS 6.4.5; its payload carries only the ISO deadline plus whole hours remaining, never the password or any secret), `integrity_drift`, `update_available`, `backup_failed`, `rcsi_off_degraded`, `leadership_acquired`, `dr_activated`, `gcm_invocations` (the per-key AES-GCM invocation bound crossing its 2^31 soft warn — ASVS 11.3.4; its payload carries a one-way `key_id` fingerprint plus counters, never key bytes) The three reachable **inverse** signals — `connection_restored`, `leadership_lost`, `dr_released` — are never rows here: `_record_state` routes an inverse through `_AUTO_RESOLVE` to `resolve_alert_instances_for`, never to `upsert_alert_instance`. (A fourth mapped key, `connection_started`, is emitted by no code path today.) | rows: `event_type`, `connection`, `severity`, `status`, `first_seen`, `last_seen`, `count`, `reason`, `acked_by`, `acked_at`, `resolved_at`, `suspended_until`, `escalation_tier` | the store database, **all three backends** | the operator alert list — acknowledge / resolve / suspend. Durable state is recorded **before** any suppression or throttle return, so a muted alert still leaves a record | `GET /alerts/active` under **`monitoring:diagnose`** (**not** a PHI permission) with the same per-channel scope; ack/resolve/suspend/**resume** are POSTs on the same tier, and the separate read-only `GET /alerts/rules` view sits on its own gate | shares the connection-event window; **only RESOLVED instances are DELETEd**, by `resolved_at` — an open or acknowledged condition is never aged out from under an operator | **`reason` is free text** taken from the event's `detail`/`reason`/`label`: `safe_text(reason)[:200]` then cipher-encrypted (AAD `("alert_instance","reason",event_type,connection)` — the de-dup grain, so one AAD covers both the INSERT and the re-fire UPDATE). `content_match` is **PHI-free by contract**: the sink method takes no value parameter, only the connection, an operator label and an optional rule id |
 | **9. `response` rows with `kind='ack_sent'` — DEFAULT ON** (`[diagnostics].response_sent = true`) | the ACK/NAK the engine returned to an inbound sender, under a sentinel destination `\x1fack:<inbound>` | rows: `ack_code` (`AA`/`AE`/`AR`/`CA`/`CE`/`CR`), `ack_phase` (`decode`/`parse`/`strict`/`ingest`), `outcome`, `body`, `detail` | the store database | "what did we actually reply, and why" — the operator's answer to a sender disputing an ACK | `GET /messages/{id}/responses` under `messages:read` + `require_phi_read`; the `body` only for a caller who also holds `messages:view_raw`; every read writes a `response.read` audit row | `body`, `detail` and `resp_headers` are set to `NULL` in place by `purge_message_bodies` on the message-body window, on all three backends | **PHI fail-safe:** the ACK **body** is stored **only when the store cipher is active** — on a keyless store it is `NULL` rather than plaintext — and every NAK passes no body at all, so the offending field value is never persisted. The disposition metadata (`ack_code`/`ack_phase`/`outcome`) is non-PHI and always captured; `detail` is `safe_text`-scrubbed, 200-char bounded and encrypted |

messagefoundry/api/metrics.py

@@ -37,6 +37,7 @@
 )
 
 from messagefoundry import __version__
+from messagefoundry.pipeline.sync_reply import SyncReplyMetrics
 from messagefoundry.store.pool_metrics import PoolStatus
 from messagefoundry.store.store import (
     ClaimProcStatus,
@@ -201,6 +202,10 @@
     # is off — the gauges are then ABSENT rather than 0, so a scrape can tell "not requested" from
     # "requested and degraded" (a constant 0 on every SQLite fleet would be pure alert noise).
     claim_proc: ClaimProcStatus | None = None
+    # ADR 0154 D8: per-inbound synchronous-reply counters, read from the runner rather than the
+    # store — they are process-lifetime in-memory counts, not persisted aggregates. Empty on every
+    # instance with no reply_from inbound, so those scrapes are byte-identical.
+    sync_replies: dict[str, SyncReplyMetrics] = field(default_factory=dict)
 
 
 async def gather_snapshot(engine: Engine) -> _Snapshot:
@@ -209,6 +214,10 @@
     All ``await``s — and therefore all store I/O — live here; nothing downstream blocks.
     """
     now = time.time()
+    # Engine.registry_runner is a public property; the counters live with the runner that owns the
+    # resolvers because api/metrics.py otherwise builds every family from engine.store alone.
+    runner = engine.registry_runner
+    sync_replies = runner.sync_reply_metrics() if runner is not None else {}
     cm = await engine.store.connection_metrics(
         since=engine.started_at or now, now=now, rate_window=_RATE_WINDOW
     )
@@ -225,6 +234,7 @@
     committed_txns = int(getattr(engine.store, "committed_txns", 0))
     body_copies = int(getattr(engine.store, "body_copies", 0))
     return _Snapshot(
+        sync_replies=sync_replies,
         version=__version__,
         inbound=cm.inbound,
         destinations=cm.destinations,
@@ -253,7 +263,7 @@
     def __init__(self, snap: _Snapshot) -> None:
         self._s = snap
 
     def collect(self) -> Iterable[Any]:
         s = self._s
 
         build = GaugeMetricFamily(
@@ -310,6 +320,37 @@
             received.add_metric([channel_id], float(im.read))
             errored.add_metric([channel_id], float(im.errored))
         yield received
+
+        # ADR 0154 D8 — the synchronous-reply SLO series. Labelled `status`, NOT `outcome`: the label
+        # allowlist above is a PHI contract, and the outcome enum is a fixed non-PHI constant set, so
+        # it rides an existing label rather than widening a deliberately closed one.
+        # rate(timeout)/rate(total) IS the proxy API's error budget, which is why `degraded` is a
+        # distinct label value rather than folded into timeout.
+        if s.sync_replies:
+            replies = CounterMetricFamily(
+                "messagefoundry_http_sync_replies_total",
+                "Synchronous captured-downstream replies resolved, by outcome (process lifetime).",
+                labels=["connection", "status"],
+            )
+            wait = GaugeMetricFamily(
+                "messagefoundry_http_sync_reply_wait_seconds",
+                "Mean time an HTTP turn blocked on a captured downstream reply (process lifetime). "
+                "Answers 'is this approaching reply_timeout?' before the pager does.",
+                labels=["connection"],
+            )
+            waiters = GaugeMetricFamily(
+                "messagefoundry_http_sync_reply_waiters",
+                "HTTP turns currently blocked on a captured downstream reply.",
+                labels=["connection"],
+            )
+            for connection, m in sorted(s.sync_replies.items()):
+                for status, count in sorted(m.totals.items()):
+                    replies.add_metric([connection, status], float(count))
+                wait.add_metric([connection], m.mean_wait_seconds)
+                waiters.add_metric([connection], float(m.live))
+            yield replies
+            yield wait
+            yield waiters
         yield errored
 
         # --- outbound counters + gauges (per connection/destination) ---------

messagefoundry/checks.py

@@ -1303,6 +1303,10 @@ def _check_build(
             # hop raises here rather than shipping and only refusing at serve.
             posture=hop_posture_from_ai(settings.ai, enforcement=settings.security.enforcement),
             trust_anchor_policy=settings.tls.policy(),
+            # ADR 0154 D4: the EFFECTIVE ordering / max_attempts refusals need the resolved
+            # [delivery] defaults. Without them that arm is skipped rather than guessed, and the
+            # misconfiguration would surface only at serve rather than at commit/CI.
+            delivery=settings.delivery,
         )
     except WiringError as exc:
         return CheckResult("build-check", ok=False, required=True, detail=str(exc))