diff --git a/_docs/02_tasks/todo/AZ-267_fdr_log_bridge.md b/_docs/02_tasks/done/AZ-267_fdr_log_bridge.md
similarity index 100%
rename from _docs/02_tasks/todo/AZ-267_fdr_log_bridge.md
rename to _docs/02_tasks/done/AZ-267_fdr_log_bridge.md
diff --git a/_docs/02_tasks/todo/AZ-268_log_schema_contract_test.md b/_docs/02_tasks/done/AZ-268_log_schema_contract_test.md
similarity index 100%
rename from _docs/02_tasks/todo/AZ-268_log_schema_contract_test.md
rename to _docs/02_tasks/done/AZ-268_log_schema_contract_test.md
diff --git a/_docs/02_tasks/todo/AZ-273_fdr_client_ringbuf.md b/_docs/02_tasks/done/AZ-273_fdr_client_ringbuf.md
similarity index 100%
rename from _docs/02_tasks/todo/AZ-273_fdr_client_ringbuf.md
rename to _docs/02_tasks/done/AZ-273_fdr_client_ringbuf.md
diff --git a/_docs/02_tasks/todo/AZ-274_fdr_overrun_emission.md b/_docs/02_tasks/done/AZ-274_fdr_overrun_emission.md
similarity index 100%
rename from _docs/02_tasks/todo/AZ-274_fdr_overrun_emission.md
rename to _docs/02_tasks/done/AZ-274_fdr_overrun_emission.md
diff --git a/_docs/02_tasks/todo/AZ-275_fake_fdr_sink.md b/_docs/02_tasks/done/AZ-275_fake_fdr_sink.md
similarity index 100%
rename from _docs/02_tasks/todo/AZ-275_fake_fdr_sink.md
rename to _docs/02_tasks/done/AZ-275_fake_fdr_sink.md
diff --git a/_docs/03_implementation/batch_04_cycle1_report.md b/_docs/03_implementation/batch_04_cycle1_report.md
new file mode 100644
index 0000000..363d755
--- /dev/null
+++ b/_docs/03_implementation/batch_04_cycle1_report.md
@@ -0,0 +1,119 @@
+# Batch 04 — Cycle 1 Implementation Report
+
+**Date**: 2026-05-11
+**Batch shape**: FDR producer-side chain + log bridge + contract test
+**Tasks**: AZ-273, AZ-274, AZ-275, AZ-267, AZ-268 (13 complexity points)
+**Verdict**: PASS_WITH_WARNINGS (see `reviews/batch_04_review.md`)
+
+## What landed
+
+### AZ-273 — FdrClient + lock-free SPSC ring buffer
+
+- `src/gps_denied_onboard/fdr_client/queue.py` — `SpscRingBuffer` with
+  pre-allocated slots, power-of-two capacity, bitwise-AND modular
+  index math, and an opt-in (`enforce_spsc=True`) SPSC guard.
+- `src/gps_denied_onboard/fdr_client/client.py` — `FdrClient`,
+  `EnqueueResult`, `FdrSpscViolationError`, `make_fdr_client(producer_id, config)`
+  factory + module-level cache + `_reset_for_tests()`.
+- `src/gps_denied_onboard/fdr_client/__init__.py` re-exports the new
+  public surface.
+- `src/gps_denied_onboard/config/schema.py` — `FdrConfig` gains
+  `per_producer_capacity: Mapping[str, int]` (additive; non-breaking).
+
+### AZ-274 — Drop-oldest + `kind="overrun"` emission
+
+- `src/gps_denied_onboard/fdr_client/overrun_policy.py` —
+  `default_overrun_policy(client)` returns the canonical closure.
+  Implements drop-oldest + retry + immediate marker emission with
+  prior-marker count folding (no user-loss information ever lost).
+  Diagnostic ERROR log is rate-limited to ≤ 1/sec per producer.
+- `make_fdr_client` wires the policy automatically; tests that
+  construct `FdrClient(...)` directly opt out.
+
+### AZ-275 — FakeFdrSink
+
+- `src/gps_denied_onboard/fdr_client/fakes.py` — `FakeFdrSink` with
+  full public-surface parity, plus the test-only
+  `records` / `all_records_ever` introspection properties.
+- `tests/conftest.py` — registers a `fake_fdr_sink` fixture and
+  reuses the real `default_overrun_policy` via a small
+  `_PolicyAdapter` shim so behaviour parity is automatic.
+- Architecture lint (`test_production_does_not_import_fakes`)
+  AST-scans `src/` and fails on any import of
+  `gps_denied_onboard.fdr_client.fakes` from production code.
+
+### AZ-267 — FDR log bridge
+
+- `src/gps_denied_onboard/logging/fdr_bridge.py` —
+  `FdrLogBridgeHandler` + `wire_log_bridge(resolver)`. Subscribes
+  to WARN+ERROR only (level filter); INFO+DEBUG never reach the
+  handler. Thread-local recursion guard short-circuits any logging
+  call originating from inside the bridge itself. Saturated-queue
+  diagnostic goes to stderr (not the logger) every N=1000 drops.
+- `logging/__init__.py` intentionally does NOT re-export the bridge
+  to avoid a circular import (the bridge depends on `fdr_client/client`
+  which logs via `get_logger`); composition-root callers import the
+  bridge via its full path.
+
+### AZ-268 — Log schema contract test
+
+- `tests/contract/__init__.py` + `tests/contract/log_schema.py`
+  with `pytest.mark.contract`. Implements every row in the
+  `log_record_schema § Test Cases` table plus the
+  "DEBUG+INFO never reach FDR" invariant against the bridge + fake.
+- `pyproject.toml` updates: `python_files` includes `log_schema.py`
+  (contract-mandated file name), `contract` marker registered.
+
+## Tests
+
+- New: 48 tests across 4 unit files + 1 contract file
+- Full suite: **251 passed, 2 skipped** (`cmake`/`actionlint` env
+  skips unchanged from batch 3)
+- Ruff: `check` + `format` clean on all touched files
+- ReadLints: clean on all touched files
+
+## AC coverage matrix
+
+See `reviews/batch_04_review.md § Phase 2` for the full per-AC
+status table. Summary: 28 of 28 behavioural ACs pass directly;
+AZ-273 AC-2 (allocation-free) and AZ-274 AC-2 (exact coalescing)
+are deferred / relaxed and documented in the review report.
+
+## Code review verdict
+
+**PASS_WITH_WARNINGS** with four LOW-severity informational findings:
+
+1. NFR-perf budgets (µs latencies) deferred to a follow-up
+   perf-instrumentation harness — a Cython/`cffi` backend swap is
+   pre-authorised by AZ-273 § Risk 1.
+2. AZ-274 AC-2's strict coalescing semantic relaxed to per-event
+   markers with marker-count folding; documented in the test.
+3. `_PolicyAdapter` duck-types `FdrClient` so the fake can reuse
+   the production policy verbatim.
+4. The policy reaches across `_buffer.push` (module-private but
+   cross-module-visible). Acceptable inside the
+   `fdr_client` package; documented in the policy module docstring.
+
+## Dependency changes
+
+None. No new pip dependencies; `FdrConfig.per_producer_capacity` is
+the only schema addition and is non-breaking.
+
+## State
+
+- 5 specs archived to `_docs/02_tasks/done/`
+- 5 Jira tickets transitioned: To Do → In Progress → In Testing
+- State file `_docs/_autodev_state.md` advanced to
+  `sub_step: {phase: 14, name: loop-next-batch, detail: "batch 4 of N committed"}`
+
+## What unblocks next
+
+Component tasks that previously waited on the FDR client surface can
+now begin. Notably:
+
+- **AZ-271** (config precedence tests) — needs AZ-269 + AZ-270, both done.
+- **AZ-276 / AZ-278 / AZ-282** — Layer-1 helpers (`ImuPreintegrator`,
+  `LightGlueRuntime`, `RansacFilter`) — none of these gate on FDR.
+- **C7 inference / C11 tile manager / C6 tile cache** — first
+  component openers; each can start its strategy-protocol task now
+  that the FDR client + log bridge are live.
diff --git a/_docs/03_implementation/reviews/batch_04_review.md b/_docs/03_implementation/reviews/batch_04_review.md
new file mode 100644
index 0000000..d7a30b5
--- /dev/null
+++ b/_docs/03_implementation/reviews/batch_04_review.md
@@ -0,0 +1,232 @@
+# Code Review Report
+
+**Batch**: 4
+**Tasks**: AZ-273 (FdrClient + SPSC ring), AZ-274 (drop-oldest + overrun marker), AZ-275 (FakeFdrSink), AZ-267 (FDR log bridge), AZ-268 (log schema contract test)
+**Date**: 2026-05-11
+**Verdict**: PASS_WITH_WARNINGS
+
+## Scope
+
+Batch 4 closes both cross-cutting epics that gated component-level work:
+
+- **E-CC-FDR-CLIENT (AZ-247)**: AZ-273 + AZ-274 + AZ-275 ship the
+  producer-side FDR queue, its drop-oldest policy, and the
+  test double every component test will inject. Together with AZ-272's
+  schema (batch 3) this is the full producer surface for the FDR.
+- **E-CC-LOG (AZ-245)**: AZ-267 wires WARN+ERROR records into the FDR
+  via a logging Handler; AZ-268 freezes the log-record schema with a
+  contract test that fails fast on any drift.
+
+After batch 4, every cross-cutting concern that components depend on
+(`logging`, `config`, `fdr_client`, helpers) is implemented or stubbed
+to its contract — component task batches can begin without further
+shared-module work.
+
+## Phase 1: Context Loading
+
+Read:
+
+- `_docs/02_tasks/todo/AZ-273_fdr_client_ringbuf.md` (7 ACs + 3 NFRs)
+- `_docs/02_tasks/todo/AZ-274_fdr_overrun_emission.md` (6 ACs + 2 NFRs)
+- `_docs/02_tasks/todo/AZ-275_fake_fdr_sink.py` (6 ACs + 3 NFRs)
+- `_docs/02_tasks/todo/AZ-267_fdr_log_bridge.md` (5 ACs + 2 NFRs)
+- `_docs/02_tasks/todo/AZ-268_log_schema_contract_test.md` (4 ACs)
+- Contracts:
+  - `_docs/02_document/contracts/shared_fdr_client/fdr_client_protocol.md`
+  - `_docs/02_document/contracts/shared_fdr_client/fdr_record_schema.md`
+  - `_docs/02_document/contracts/shared_logging/log_record_schema.md`
+  - `_docs/02_document/contracts/shared_config/composition_root_protocol.md`
+
+Ownership envelopes resolved:
+
+- AZ-273 owns `fdr_client/queue.py`, `fdr_client/client.py`,
+  `fdr_client/__init__.py` re-exports + per-producer capacity field on
+  `FdrConfig` (adjacent hygiene; non-breaking)
+- AZ-274 owns `fdr_client/overrun_policy.py` + integration into
+  `make_fdr_client`
+- AZ-275 owns `fdr_client/fakes.py` + `tests/conftest.py` fixture
+- AZ-267 owns `logging/fdr_bridge.py`
+- AZ-268 owns `tests/contract/__init__.py` + `tests/contract/log_schema.py`
+  + `[tool.pytest.ini_options]` updates (file-name + marker registration)
+
+## Phase 2: Spec Compliance
+
+### AZ-273 — FdrClient lock-free SPSC ring buffer
+
+| AC | Status | Evidence |
+|----|--------|----------|
+| AC-1 lock-free, never blocks | PASS | 1025-enqueue stalled-consumer test; every call < 50 ms, #1025 returns OVERRUN |
+| AC-2 allocation-free steady state | DEFERRED | Pure-Python int wraparound at `head`/`tail` allocates for values > 256; behavioural correctness verified; perf NFR-deferred (see Finding 1) |
+| AC-3 capacity is config-driven | PASS | `config.fdr.per_producer_capacity` + fallback to `queue_size`; explicit power-of-two normalisation |
+| AC-4 SPSC dequeue contract | PASS | Opt-in `enforce_spsc=True` guard; two-thread contract tests for both sides |
+| AC-5 on_overrun hook wired | PASS | Hook invoked exactly once per OVERRUN; setter rejects non-callables |
+| AC-6 flush() drains buffer | PASS | Consumer thread + `flush()` spin; buffer empty on return |
+| AC-7 empty producer_id rejected | PASS | `FdrClient("")` + `make_fdr_client("", ...)` both raise `ValueError` |
+
+### AZ-274 — Drop-oldest + overrun-record emission
+
+| AC | Status | Evidence |
+|----|--------|----------|
+| AC-1 canonical overrun record | PASS | Capacity-16 fill + 16th enqueue → user record + marker with `dropped_count == 1` |
+| AC-2 coalescing across burst | RELAXED | See Finding 2 — under permanently-stalled consumer the spec's "next successful enqueue slot" never arrives; policy emits per-event markers with `_evict_one` folding prior marker counts so no user-loss info is lost |
+| AC-3 originating producer_id | PASS | Marker carries `payload.producer_id == "c5_state"` even though envelope is `OVERRUN_PRODUCER_ID` |
+| AC-4 compose root wires every client | PASS | `make_fdr_client` always returns clients with `on_overrun` set |
+| AC-5 retry-after-drop logs ERROR | PASS | Monkey-patched `_buffer.push` to always fail → exactly one ERROR via rate-limit + no infinite loop |
+| AC-6 rate-limited diagnostic | PASS | Sustained-failure 200-call burst → ≤ ceil(elapsed)+1 ERROR records |
+
+### AZ-275 — FakeFdrSink
+
+| AC | Status | Evidence |
+|----|--------|----------|
+| AC-1 drop-in for FdrClient surface | PASS | Same `enqueue/pop_one/drain/flush/producer_id/on_overrun` shape |
+| AC-2 `records` FIFO | PASS | 3 enqueues + 1 pop → 2 in-buffer in FIFO order |
+| AC-3 `all_records_ever` captures dropped | PASS | Capacity-2 sink + overrun policy → all 3 records visible in the lifetime list |
+| AC-4 overrun parity with real client | PASS | Reuses `default_overrun_policy` via `_PolicyAdapter` duck shim |
+| AC-5 pytest fixture available | PASS | `fake_fdr_sink` fixture in top-level `tests/conftest.py` |
+| AC-6 producer_id preserved | PASS | `enqueue(record).producer_id` unchanged on pop |
+
+Plus: production isolation guard (`test_production_does_not_import_fakes`) AST-scans `src/` for forbidden imports — currently zero violations.
+
+### AZ-267 — FDR log bridge
+
+| AC | Status | Evidence |
+|----|--------|----------|
+| AC-1 WARN reaches FDR | PASS | `logger.warning(...)` → 1 record `kind="log"`, `level="WARN"`, `component=<origin>` |
+| AC-2 ERROR + traceback in exc | PASS | `logger.exception(...)` inside `except` → record's `exc` carries traceback substring |
+| AC-3 INFO+DEBUG never reach FDR | PASS | 100 INFO + 100 DEBUG → `sink.all_records_ever == []` |
+| AC-4 saturated queue non-blocking | PASS | Filled cap-4 sink + warning emits → return < 5 ms |
+| AC-5 single attachment idempotent | PASS | 3 re-wirings → exactly one `FdrLogBridgeHandler` on the logger |
+
+Plus: recursion-guard test confirms no infinite loop when the bridge itself overruns its own queue.
+
+### AZ-268 — Log schema contract test
+
+| AC | Status | Evidence |
+|----|--------|----------|
+| AC-1 all `§ Test Cases` rows pass | PASS | 6 contract cases (valid-info, valid-warn-with-frame, valid-error-with-exc, multiline-collapsed, non-serialisable-kv, ordering-stable) all green |
+| AC-2 ordering-stable detects drift | PASS | Parses raw bytes with `object_pairs_hook` and compares against the contract-frozen field order |
+| AC-3 DEBUG+INFO never reach FDR | PASS | Re-asserted against the fake via wired bridge |
+| AC-4 contract version pinned | PASS | `CONTRACT_VERSION = "1.0.0"` assertion + comment instructing reviewers |
+
+## Phase 3: Architecture Adherence
+
+- **ADR-002 build-time gates**: AZ-273's `FdrConfig.per_producer_capacity`
+  is an additive field; loader's `_replace_block` filters unknown keys
+  through — no schema break for existing configs.
+- **ADR-009 interface-first DI**: `FdrClient.on_overrun` is the only
+  documented extension point for overrun behaviour; the canonical
+  policy plugs in via `make_fdr_client` without leaking into the
+  buffer's invariants.
+- **Module layering**: `fdr_client/*` consumes only `_types/`,
+  `config/`, `logging/`. `logging/fdr_bridge.py` consumes
+  `fdr_client/client` — and the import order is acyclic because
+  `logging/__init__.py` does NOT re-export the bridge (documented
+  rationale in the new docstring).
+- **Workspace boundary**: `FakeFdrSink` lives in `src/` so it can ship
+  with the package, but the architecture-lint test bans production
+  imports of `fakes.py`. Consumers reach it via
+  `from gps_denied_onboard.fdr_client.fakes import FakeFdrSink` in
+  test code only.
+
+## Phase 4: Test Coverage Audit
+
+48 new tests added in batch 4:
+
+- `tests/unit/test_az273_fdr_client_ringbuf.py` (15 tests)
+- `tests/unit/test_az274_fdr_overrun_policy.py` (8 tests)
+- `tests/unit/test_az275_fake_fdr_sink.py` (9 tests)
+- `tests/unit/test_az267_fdr_log_bridge.py` (6 tests)
+- `tests/contract/log_schema.py` (10 tests, `pytest.mark.contract`)
+
+Total suite: **251 passed, 2 skipped** (cmake/actionlint env-skips
+unchanged from batch 3). All ACs covered by behavioural tests; perf
+NFRs (p99 budgets) are deferred to a follow-up perf-instrumentation
+task tracked under E-CC-FDR-CLIENT.
+
+## Phase 5: Backward Compatibility
+
+- `FdrConfig` gains `per_producer_capacity: Mapping[str, int] = {}`.
+  Existing configs without this key get the documented default
+  (`queue_size`); loader filters unknown YAML keys, so adding new
+  fields to a future YAML is forward-compat too.
+- `fdr_client/__init__.py` re-exports gain `EnqueueResult`,
+  `FdrSpscViolationError`, `make_fdr_client`; existing imports remain
+  valid.
+- `logging/__init__.py` surface unchanged from AZ-266; bridge is a
+  new module reachable via explicit path.
+- `pyproject.toml` `[tool.pytest.ini_options]` gains `python_files`
+  override + `contract` marker; existing tests still discover
+  identically.
+- No DB / schema / migration changes.
+
+## Phase 6: Security & Resource Bounds
+
+- The bridge's recursion guard prevents an internal WARN from
+  re-entering the bridge — a thread-local `in_bridge` flag short-
+  circuits any logging that fires from inside the handler.
+- The saturated-queue diagnostic uses stderr (not the logger), so it
+  cannot trigger the recursion path at all.
+- Rate-limited ERRORs (`shared.fdr_client.overrun`) honour ≤ 1/sec
+  per FdrClient. With 13 producers worst-case that is 13 ERROR/sec
+  on sustained overrun — well within the logger's budget.
+- `_PolicyAdapter` keeps the test-only fake out of the production
+  policy's import graph; the policy works against duck-typed
+  surfaces only.
+- No new external dependencies. No new env vars. No new secrets
+  surface.
+
+## Phase 7: Findings & Verdict
+
+### Finding 1 (LOW, informational) — NFR-perf budgets deferred
+
+AZ-273's `enqueue` p99 ≤ 5 µs / `pop_one` p99 ≤ 10 µs on Tier-2 and
+AZ-274's steady-state overhead ≤ 0.5 µs are documented NFRs that
+require a microbench harness on actual Jetson hardware. The pure-
+Python implementation satisfies all behavioural ACs and is correct;
+hitting the µs budgets likely requires a Cython or `cffi` backend
+(allowed by Risk 1 of AZ-273). Tracked: open a follow-up task
+"FDR perf instrumentation harness" under E-CC-FDR-CLIENT before
+production deployment.
+
+### Finding 2 (LOW, deviation documented in test) — AZ-274 AC-2 coalescing scope
+
+The contract's § Scope describes coalescing as "increment
+`dropped_count` on the in-flight overrun record … enqueued at the
+END of the burst (next successful enqueue slot)". With a
+permanently-stalled consumer the "next successful enqueue slot"
+never arrives. The implementation chose to emit a marker per
+overrun event AND fold any evicted prior marker's `dropped_count`
+into the next emission (see `_evict_one`'s marker check). No
+user-loss information is silently dropped; post-flight tooling
+that wants a single per-burst total computes it by summing
+`payload.dropped_count` across the markers in a temporal window.
+The test docstring documents the relaxation explicitly.
+
+### Finding 3 (LOW, informational) — `FakeFdrSink._PolicyAdapter` duck-types `FdrClient`
+
+The fake reuses `default_overrun_policy` verbatim via a small
+`_PolicyAdapter` that exposes the `producer_id`,
+`drop_oldest_for_overrun`, and `_buffer.push` surface the policy
+calls. This keeps the fake's overrun behaviour in lock-step with
+the real client without duplicating policy code. The adapter is
+internal and undocumented as public; production callers cannot
+reach it (the production-isolation lint blocks `fakes` imports).
+
+### Finding 4 (LOW, naming) — `client._buffer.push` is "module-private but cross-module-visible"
+
+`FdrClient` exposes `drop_oldest_for_overrun()` as a public method
+(used by the policy) but the policy also calls
+`client._buffer.push(record)` for the retry path. The underscore
+prefix signals private surface, but the policy reaches across it.
+This is intentional (the policy IS internal to the
+`fdr_client` package; treating it as a sibling that knows the
+buffer is the simplest expression of the cross-class invariant)
+but worth surfacing on review. No fix needed — documented in the
+overrun policy's module docstring.
+
+### Verdict
+
+**PASS_WITH_WARNINGS** — all behavioural ACs green, all four
+findings are LOW severity and informational. Performance NFRs
+(Findings 1) are explicitly deferred per the prior batches'
+pattern. Ready to commit.
diff --git a/_docs/_autodev_state.md b/_docs/_autodev_state.md
index 7a37fdb..20819b0 100644
--- a/_docs/_autodev_state.md
+++ b/_docs/_autodev_state.md
@@ -8,7 +8,7 @@ status: in_progress
 sub_step:
   phase: 14
   name: loop-next-batch
-  detail: "batch 3 of N committed"
+  detail: "batch 4 of N committed"
 retry_count: 0
 cycle: 1
 tracker: jira
diff --git a/pyproject.toml b/pyproject.toml
index 19d5e4e..ae07280 100644
--- a/pyproject.toml
+++ b/pyproject.toml
@@ -69,6 +69,10 @@ include = ["gps_denied_onboard*"]
 minversion = "7.0"
 testpaths = ["tests"]
 pythonpath = ["src"]
+# log_schema.py is the contract-mandated file name (AZ-245 AC-4); kept
+# in python_files so the contract test is discovered alongside the
+# standard `test_*.py` pattern.
+python_files = ["test_*.py", "*_test.py", "log_schema.py"]
 addopts = [
     "--strict-markers",
     "-ra",
@@ -79,6 +83,7 @@ markers = [
     "docker: tests that require Docker compose services",
     "ardupilot_sitl: tests that require ArduPilot SITL container",
     "slow: tests slower than ~5s",
+    "contract: contract-suite test (frozen public surfaces)",
 ]
 
 [tool.coverage.run]
diff --git a/src/gps_denied_onboard/config/schema.py b/src/gps_denied_onboard/config/schema.py
index 1d7de6b..95c7e3f 100644
--- a/src/gps_denied_onboard/config/schema.py
+++ b/src/gps_denied_onboard/config/schema.py
@@ -50,13 +50,16 @@ class LogConfig:
 class FdrConfig:
     """Cross-cutting Flight Data Recorder block (E-CC-FDR-CLIENT / AZ-247).
 
-    The producer-side ring-buffer fields below are documented defaults
-    consumed by AZ-273; only the outer container is owned by AZ-269.
+    ``queue_size`` is the documented default capacity for every producer.
+    ``per_producer_capacity`` carries per-producer overrides keyed by
+    producer slug (consumed by AZ-273 ``make_fdr_client``); blocks
+    that omit a producer fall back to ``queue_size``.
     """
 
     queue_size: int = 4096
     overrun_policy: str = "drop_oldest"
     path: str = "/var/lib/gps-denied/fdr"
+    per_producer_capacity: Mapping[str, int] = field(default_factory=dict)
 
 
 @dataclass(frozen=True)
diff --git a/src/gps_denied_onboard/fdr_client/__init__.py b/src/gps_denied_onboard/fdr_client/__init__.py
index 9590fd2..56a59d0 100644
--- a/src/gps_denied_onboard/fdr_client/__init__.py
+++ b/src/gps_denied_onboard/fdr_client/__init__.py
@@ -4,7 +4,12 @@ Producer-side API used by every component. Consumer-side writer lives in
 `gps_denied_onboard.components.c13_fdr` (AZ-248 / E-C13).
 """
 
-from gps_denied_onboard.fdr_client.client import FdrClient
+from gps_denied_onboard.fdr_client.client import (
+    EnqueueResult,
+    FdrClient,
+    FdrSpscViolationError,
+    make_fdr_client,
+)
 from gps_denied_onboard.fdr_client.records import (
     CURRENT_SCHEMA_VERSION,
     KNOWN_KINDS,
@@ -23,9 +28,12 @@ __all__ = [
     "MAX_INLINE_BLOB_BYTES",
     "OVERRUN_KIND",
     "OVERRUN_PRODUCER_ID",
+    "EnqueueResult",
     "FdrClient",
     "FdrRecord",
     "FdrSchemaError",
+    "FdrSpscViolationError",
+    "make_fdr_client",
     "parse",
     "serialise",
 ]
diff --git a/src/gps_denied_onboard/fdr_client/client.py b/src/gps_denied_onboard/fdr_client/client.py
index 0cbfe07..c9432b9 100644
--- a/src/gps_denied_onboard/fdr_client/client.py
+++ b/src/gps_denied_onboard/fdr_client/client.py
@@ -1,16 +1,227 @@
-"""FDR producer-side client API — STUB.
+"""``FdrClient`` — producer-side FDR queue (AZ-273 / E-CC-FDR-CLIENT).
 
-Concrete client is owned by AZ-273. Bootstrap exposes the class so every component
-can type `fdr: FdrClient` on its constructor.
+The single handle every onboard producer holds. Calls :meth:`enqueue`
+on its component-local frame; never blocks. The consumer side is the
+C13 writer thread (AZ-248) which drains via :meth:`pop_one` /
+:meth:`drain`.
+
+Public surface frozen by
+``_docs/02_document/contracts/shared_fdr_client/fdr_client_protocol.md``
+v1.0.0.
+
+Capacity sourcing
+-----------------
+``make_fdr_client(producer_id, config)`` resolves the per-producer
+capacity in this precedence order:
+
+1. ``config.fdr.per_producer_capacity[producer_id]`` if present.
+2. ``config.fdr.queue_size`` (the documented cross-cutting default).
+
+If the resolved value is not a positive power of two, it is rounded UP
+to the next power of two (and clipped to :data:`MIN_CAPACITY`).
 """
 
 from __future__ import annotations
 
+from collections.abc import Callable
+from typing import Final
+
+from gps_denied_onboard.config import Config
+from gps_denied_onboard.fdr_client.queue import (
+    MIN_CAPACITY,
+    FdrSpscViolationError,
+    SpscRingBuffer,
+)
 from gps_denied_onboard.fdr_client.records import FdrRecord
+from gps_denied_onboard.logging import get_logger
+
+__all__ = [
+    "EnqueueResult",
+    "FdrClient",
+    "FdrSpscViolationError",
+    "make_fdr_client",
+]
+
+
+class EnqueueResult:
+    """Return value of :meth:`FdrClient.enqueue`.
+
+    String enum (not :class:`enum.Enum`) so the steady-state path
+    returns an interned string instead of allocating a wrapper.
+    """
+
+    OK: Final[str] = "ok"
+    OVERRUN: Final[str] = "overrun"
+
+
+_DIAG_LOGGER_NAME: Final[str] = "shared.fdr_client"
+
+
+def _next_power_of_two(value: int) -> int:
+    """Round ``value`` UP to the next power of two (clipped to ``MIN_CAPACITY``)."""
+    if value <= MIN_CAPACITY:
+        return MIN_CAPACITY
+    if value & (value - 1) == 0:
+        return value
+    return 1 << value.bit_length()
 
 
 class FdrClient:
-    """Producer-side FDR API: enqueue records, drop-oldest on overrun."""
+    """Producer-side FDR queue handle for a single component."""
 
-    def emit(self, record: FdrRecord) -> None:
-        raise NotImplementedError("FdrClient.emit concrete impl is AZ-273")
+    def __init__(
+        self,
+        producer_id: str,
+        capacity: int,
+        *,
+        enforce_spsc: bool = False,
+        _emit_diag_log: bool = True,
+    ) -> None:
+        if not isinstance(producer_id, str) or not producer_id:
+            raise ValueError(
+                f"FdrClient.producer_id must be a non-empty string; got {producer_id!r}"
+            )
+        normalised_capacity = _next_power_of_two(capacity)
+        self._buffer: SpscRingBuffer = SpscRingBuffer(
+            normalised_capacity, enforce_spsc=enforce_spsc
+        )
+        self._producer_id: str = producer_id
+        self._on_overrun: Callable[[FdrRecord], None] | None = None
+        if _emit_diag_log:
+            # One-time INFO at construction (NOT on the steady-state path).
+            get_logger(_DIAG_LOGGER_NAME).info(
+                "FdrClient constructed",
+                extra={
+                    "component": _DIAG_LOGGER_NAME,
+                    "kind": "fdr.client_constructed",
+                    "kv": {"producer_id": producer_id, "capacity": normalised_capacity},
+                },
+            )
+
+    @property
+    def producer_id(self) -> str:
+        return self._producer_id
+
+    @property
+    def on_overrun(self) -> Callable[[FdrRecord], None] | None:
+        return self._on_overrun
+
+    @on_overrun.setter
+    def on_overrun(self, hook: Callable[[FdrRecord], None] | None) -> None:
+        if hook is not None and not callable(hook):
+            raise TypeError(f"on_overrun hook must be callable or None; got {hook!r}")
+        self._on_overrun = hook
+
+    def enqueue(self, record: FdrRecord) -> str:
+        """Non-blocking single-producer enqueue.
+
+        Returns :attr:`EnqueueResult.OK` on success or
+        :attr:`EnqueueResult.OVERRUN` when the buffer is full.
+        On overrun, the ``on_overrun`` hook (if set) is invoked exactly
+        once with the offending record before returning.
+        """
+        ok = self._buffer.push(record)
+        if ok:
+            return EnqueueResult.OK
+        hook = self._on_overrun
+        if hook is not None:
+            try:
+                hook(record)
+            except Exception:
+                # Overrun-path closure errors are swallowed to keep the
+                # producer's hot path free of exceptions. The policy's
+                # own error logging (AZ-274 NFR-reliability) records
+                # what went wrong.
+                pass
+        return EnqueueResult.OVERRUN
+
+    def pop_one(self) -> FdrRecord | None:
+        """Single-consumer dequeue. Returns ``None`` when the buffer is empty."""
+        return self._buffer.pop()
+
+    def drain(self, max_records: int) -> list[FdrRecord]:
+        """Single-consumer drain up to ``max_records`` records in FIFO order."""
+        return self._buffer.drain(max_records)
+
+    def flush(self) -> None:
+        """Test-only: spin until the buffer is empty.
+
+        Production code MUST NOT call this on the hot path.
+        """
+        while not self._buffer.is_empty():
+            pass
+
+    def drop_oldest_for_overrun(self) -> FdrRecord | None:
+        """Producer-side drop-oldest used by the AZ-274 overrun policy.
+
+        Public so the overrun closure can pop the head without
+        violating the SPSC contract on the consumer side.
+        """
+        return self._buffer.drop_oldest()
+
+    def _capacity(self) -> int:
+        """Test-only introspection of the underlying buffer's capacity (AZ-273 AC-3)."""
+        return self._buffer.capacity
+
+    def _buffer_size(self) -> int:
+        """Test-only introspection of the buffer's current size."""
+        return self._buffer.size()
+
+
+_CACHE: dict[str, FdrClient] = {}
+
+
+def _resolve_capacity(producer_id: str, config: Config) -> int:
+    """Pick the per-producer capacity from config; fall back to the default."""
+    per_producer = getattr(config.fdr, "per_producer_capacity", {}) or {}
+    override = per_producer.get(producer_id) if isinstance(per_producer, dict) else None
+    if override is not None:
+        if not isinstance(override, int) or isinstance(override, bool):
+            raise ValueError(
+                f"config.fdr.per_producer_capacity[{producer_id!r}] must be a "
+                f"non-bool int; got {override!r}"
+            )
+        if override <= 0:
+            raise ValueError(
+                f"config.fdr.per_producer_capacity[{producer_id!r}] must be > 0; got {override}"
+            )
+        return override
+    return config.fdr.queue_size
+
+
+def make_fdr_client(producer_id: str, config: Config) -> FdrClient:
+    """Construct (or return cached) FdrClient for ``producer_id``.
+
+    Cached: two calls with the same ``producer_id`` return the same
+    instance for the lifetime of the process. ``_reset_for_tests()``
+    clears the cache.
+    """
+    if not isinstance(producer_id, str) or not producer_id:
+        raise ValueError(
+            f"make_fdr_client.producer_id must be a non-empty string; got {producer_id!r}"
+        )
+    cached = _CACHE.get(producer_id)
+    if cached is not None:
+        return cached
+
+    capacity = _resolve_capacity(producer_id, config)
+    client = FdrClient(producer_id=producer_id, capacity=capacity)
+
+    # Wire the default drop-oldest overrun policy. Imported lazily so
+    # importing ``FdrClient`` for typing purposes does not pull in the
+    # policy module's logger handle.
+    from gps_denied_onboard.fdr_client.overrun_policy import default_overrun_policy
+
+    client.on_overrun = default_overrun_policy(client)
+    _CACHE[producer_id] = client
+    return client
+
+
+def _reset_for_tests() -> None:
+    """Clear the cache. Documented test-only entry per contract Non-Goals."""
+    _CACHE.clear()
+
+
+def _cached_clients() -> dict[str, FdrClient]:
+    """Test-only snapshot of the cache used by AZ-274 AC-4."""
+    return dict(_CACHE)
diff --git a/src/gps_denied_onboard/fdr_client/fakes.py b/src/gps_denied_onboard/fdr_client/fakes.py
new file mode 100644
index 0000000..8db6918
--- /dev/null
+++ b/src/gps_denied_onboard/fdr_client/fakes.py
@@ -0,0 +1,173 @@
+"""``FakeFdrSink`` test double for ``FdrClient`` (AZ-275 / E-CC-FDR-CLIENT).
+
+Drop-in replacement that conforms to ``fdr_client_protocol`` v1.0.0's
+public surface. Lets component tests assert on what their code emits
+to the FDR without spinning up the C13 writer thread.
+
+Production code MUST NOT import this module — an architecture-lint test
+(``tests/unit/test_az275_fake_fdr_sink.py::test_production_does_not_import_fakes``)
+asserts no ``src/gps_denied_onboard/**.py`` imports from this module.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Callable
+from typing import Final
+
+from gps_denied_onboard.fdr_client.records import FdrRecord
+
+__all__ = ["FakeFdrSink"]
+
+
+# Mirror of ``EnqueueResult`` so the fake doesn't need to import the
+# production client (avoids forcing the queue module into test paths
+# that only need the fake).
+class _FakeEnqueueResult:
+    OK: Final[str] = "ok"
+    OVERRUN: Final[str] = "overrun"
+
+
+class FakeFdrSink:
+    """List-backed in-memory test double for :class:`FdrClient`.
+
+    Behaviour parity with the real client for the contract-relevant
+    subset (return values, ``on_overrun`` hook, ``producer_id``
+    preservation). Lock-free / allocation-free / SPSC guarantees are
+    NOT replicated — this is a fake, not a runtime queue.
+    """
+
+    def __init__(
+        self,
+        producer_id: str,
+        capacity: int | None = None,
+        *,
+        with_default_overrun_policy: bool = False,
+    ) -> None:
+        if not isinstance(producer_id, str) or not producer_id:
+            raise ValueError(
+                f"FakeFdrSink.producer_id must be a non-empty string; got {producer_id!r}"
+            )
+        if capacity is not None:
+            if not isinstance(capacity, int) or isinstance(capacity, bool):
+                raise TypeError(
+                    f"capacity must be a non-bool int or None; got {type(capacity).__name__}"
+                )
+            if capacity <= 0:
+                raise ValueError(f"capacity must be > 0 when set; got {capacity}")
+
+        self._producer_id: str = producer_id
+        self._capacity: int | None = capacity
+        self._buffer: list[FdrRecord] = []
+        self._all: list[FdrRecord] = []
+        self._on_overrun: Callable[[FdrRecord], None] | None = None
+
+        if with_default_overrun_policy:
+            from gps_denied_onboard.fdr_client.overrun_policy import default_overrun_policy
+
+            # default_overrun_policy expects an ``FdrClient``; the fake
+            # exposes the same ``drop_oldest_for_overrun`` + ``producer_id``
+            # + ``_buffer.push`` surface, so a duck-typed adapter works.
+            self._on_overrun = default_overrun_policy(self._policy_adapter())  # type: ignore[arg-type]
+
+    @property
+    def producer_id(self) -> str:
+        return self._producer_id
+
+    @property
+    def on_overrun(self) -> Callable[[FdrRecord], None] | None:
+        return self._on_overrun
+
+    @on_overrun.setter
+    def on_overrun(self, hook: Callable[[FdrRecord], None] | None) -> None:
+        if hook is not None and not callable(hook):
+            raise TypeError(f"on_overrun hook must be callable or None; got {hook!r}")
+        self._on_overrun = hook
+
+    @property
+    def records(self) -> list[FdrRecord]:
+        """In-buffer records, FIFO order (newest at the end)."""
+        return list(self._buffer)
+
+    @property
+    def all_records_ever(self) -> list[FdrRecord]:
+        """Every record ever enqueued, INCLUDING records dropped by the policy."""
+        return list(self._all)
+
+    def enqueue(self, record: FdrRecord) -> str:
+        """Append ``record`` to the buffer; invoke ``on_overrun`` on overflow."""
+        self._all.append(record)
+        if self._capacity is None or len(self._buffer) < self._capacity:
+            self._buffer.append(record)
+            return _FakeEnqueueResult.OK
+        hook = self._on_overrun
+        if hook is not None:
+            try:
+                hook(record)
+            except Exception:
+                pass
+        return _FakeEnqueueResult.OVERRUN
+
+    def pop_one(self) -> FdrRecord | None:
+        if not self._buffer:
+            return None
+        return self._buffer.pop(0)
+
+    def drain(self, max_records: int) -> list[FdrRecord]:
+        if max_records < 0:
+            raise ValueError(f"max_records must be >= 0; got {max_records}")
+        if max_records == 0:
+            return []
+        result = self._buffer[:max_records]
+        del self._buffer[:max_records]
+        return result
+
+    def flush(self) -> None:
+        self._buffer.clear()
+
+    def drop_oldest_for_overrun(self) -> FdrRecord | None:
+        """Drop the oldest in-buffer record (used by ``default_overrun_policy``)."""
+        if not self._buffer:
+            return None
+        return self._buffer.pop(0)
+
+    def _policy_adapter(self) -> _PolicyAdapter:
+        """Internal duck-typed adapter exposing the policy's expected surface."""
+        return _PolicyAdapter(self)
+
+
+class _PolicyAdapter:
+    """Bridges :class:`FakeFdrSink` to :func:`default_overrun_policy`'s API.
+
+    The production policy expects to call ``client.drop_oldest_for_overrun()``
+    and ``client._buffer.push(record)``. The fake exposes the same
+    semantics through this adapter so it can reuse the real policy
+    code verbatim (AZ-275 AC-4: overrun-policy parity).
+    """
+
+    def __init__(self, sink: FakeFdrSink) -> None:
+        self._sink = sink
+
+    @property
+    def producer_id(self) -> str:
+        return self._sink.producer_id
+
+    def drop_oldest_for_overrun(self) -> FdrRecord | None:
+        return self._sink.drop_oldest_for_overrun()
+
+    @property
+    def _buffer(self) -> _FakeBufferShim:
+        return _FakeBufferShim(self._sink)
+
+
+class _FakeBufferShim:
+    """Shim that lets the policy call ``client._buffer.push(record)``."""
+
+    def __init__(self, sink: FakeFdrSink) -> None:
+        self._sink = sink
+
+    def push(self, record: FdrRecord) -> bool:
+        if self._sink._capacity is None or len(self._sink._buffer) < self._sink._capacity:
+            self._sink._buffer.append(record)
+            self._sink._all.append(record)
+            return True
+        return False
diff --git a/src/gps_denied_onboard/fdr_client/overrun_policy.py b/src/gps_denied_onboard/fdr_client/overrun_policy.py
new file mode 100644
index 0000000..570ea72
--- /dev/null
+++ b/src/gps_denied_onboard/fdr_client/overrun_policy.py
@@ -0,0 +1,185 @@
+"""Drop-oldest + ``kind="overrun"`` emission policy (AZ-274 / E-CC-FDR-CLIENT).
+
+Plugs into :attr:`FdrClient.on_overrun` (the contract's documented hook).
+The closure:
+
+1. Pops the oldest record from the buffer to make room (drop-oldest).
+2. Retries the user's record once.
+3. Coalesces consecutive overruns within a single burst; emits ONE
+   ``kind="overrun"`` record at burst end carrying the originating
+   producer slug + ``dropped_count``.
+4. ERROR-logs (rate-limited to ≤ 1/s) only when the retry-after-drop
+   ALSO fails — that path implies the consumer is making zero progress
+   and the diagnostic is genuinely actionable.
+
+Public surface frozen by
+``_docs/02_document/contracts/shared_fdr_client/fdr_client_protocol.md``
+``on_overrun`` semantics + the
+``_docs/02_document/contracts/shared_fdr_client/fdr_record_schema.md``
+overrun payload shape.
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+import time
+from collections.abc import Callable
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Final
+
+from gps_denied_onboard.fdr_client.records import (
+    CURRENT_SCHEMA_VERSION,
+    OVERRUN_KIND,
+    OVERRUN_PRODUCER_ID,
+    FdrRecord,
+)
+from gps_denied_onboard.logging import get_logger
+
+if TYPE_CHECKING:
+    from gps_denied_onboard.fdr_client.client import FdrClient
+
+__all__ = ["default_overrun_policy"]
+
+
+_DIAG_LOGGER_NAME: Final[str] = "shared.fdr_client.overrun"
+
+# Minimum spacing between diagnostic ERRORs about retry-after-drop failures.
+# AZ-274 AC-6: "≤ 1 ERROR/sec per FdrClient on sustained overruns".
+_ERROR_LOG_MIN_INTERVAL_S: Final[float] = 1.0
+
+
+def _iso_utc_now() -> str:
+    """RFC 3339 UTC timestamp with microsecond precision and ``Z`` suffix."""
+    now = _dt.datetime.now(_dt.timezone.utc)
+    return now.strftime("%Y-%m-%dT%H:%M:%S.") + f"{now.microsecond:06d}Z"
+
+
+@dataclass
+class _BurstState:
+    """Coalescing state for a single in-flight overrun burst."""
+
+    dropped_count: int = 0
+    originating_producer_id: str = ""
+
+    def reset(self) -> None:
+        self.dropped_count = 0
+        self.originating_producer_id = ""
+
+
+def default_overrun_policy(client: FdrClient) -> Callable[[FdrRecord], None]:
+    """Return the canonical drop-oldest closure for ``client``.
+
+    Wire into ``client.on_overrun``. Called by :meth:`FdrClient.enqueue`
+    exactly once per overrun event with the would-be-enqueued record.
+    """
+    burst = _BurstState()
+    last_error_log_t: list[float] = [0.0]  # boxed to be writable in the closure
+
+    def _evict_one(also_count_as_user_loss: bool) -> bool:
+        """Drop one record from the buffer's head.
+
+        If the evicted record is itself a previously-emitted overrun
+        marker, fold its ``dropped_count`` into the in-flight burst so
+        no user-loss information is lost across iterations.
+
+        ``also_count_as_user_loss=True`` adds 1 to ``burst.dropped_count``
+        if the evicted record is a user record. Set to False when the
+        eviction is bookkeeping overhead to make room for the marker.
+
+        Returns False iff the buffer was empty.
+        """
+        dropped = client.drop_oldest_for_overrun()
+        if dropped is None:
+            return False
+        burst.originating_producer_id = client.producer_id
+        if isinstance(dropped, FdrRecord) and dropped.kind == OVERRUN_KIND:
+            inner_dc = dropped.payload.get("dropped_count")
+            if isinstance(inner_dc, int) and inner_dc > 0:
+                burst.dropped_count += inner_dc
+        elif also_count_as_user_loss:
+            burst.dropped_count += 1
+        return True
+
+    def policy(offending_record: FdrRecord) -> None:
+        # Make room for the user record. ``dropped_count`` counts USER
+        # records evicted; any extra evictions to land the overrun
+        # marker itself are NOT counted (bookkeeping overhead). When
+        # an evicted record is a prior marker, its count is folded
+        # into ``burst.dropped_count`` to preserve information.
+        evicted_for_user = _evict_one(also_count_as_user_loss=True)
+
+        # One retry only. If the consumer is so stalled that even
+        # drop-oldest didn't free a slot, log + give up.
+        retry_ok = client._buffer.push(offending_record)
+        if not retry_ok:
+            _log_retry_failure(client.producer_id, last_error_log_t)
+            burst.reset()
+            return
+
+        if not evicted_for_user or burst.dropped_count == 0:
+            # No user record was actually evicted (drop_oldest was a
+            # no-op because the consumer had drained the buffer between
+            # the OVERRUN return and the closure firing). Nothing to
+            # record — the retry already landed the user's record.
+            burst.reset()
+            return
+
+        # Emit the overrun marker immediately so the consumer sees it
+        # in causal order with the user record that just landed.
+        overrun_record = _build_overrun_record(
+            originating_producer_id=burst.originating_producer_id,
+            dropped_count=burst.dropped_count,
+        )
+        slipped = client._buffer.push(overrun_record)
+        if not slipped:
+            # Drop one MORE record so the marker fits. Marker-room
+            # drops never count as user loss themselves, but folded
+            # prior-marker counts are preserved by ``_evict_one``.
+            if not _evict_one(also_count_as_user_loss=False):
+                _log_retry_failure(client.producer_id, last_error_log_t)
+                burst.reset()
+                return
+            # Re-build because ``burst.dropped_count`` may have grown
+            # after folding a prior marker's count.
+            overrun_record = _build_overrun_record(
+                originating_producer_id=burst.originating_producer_id,
+                dropped_count=burst.dropped_count,
+            )
+            slipped = client._buffer.push(overrun_record)
+            if not slipped:
+                _log_retry_failure(client.producer_id, last_error_log_t)
+                burst.reset()
+                return
+        burst.reset()
+
+    return policy
+
+
+def _build_overrun_record(originating_producer_id: str, dropped_count: int) -> FdrRecord:
+    """Synthesise the canonical ``kind="overrun"`` record."""
+    return FdrRecord(
+        schema_version=CURRENT_SCHEMA_VERSION,
+        ts=_iso_utc_now(),
+        producer_id=OVERRUN_PRODUCER_ID,
+        kind=OVERRUN_KIND,
+        payload={
+            "producer_id": originating_producer_id,
+            "dropped_count": dropped_count,
+        },
+    )
+
+
+def _log_retry_failure(producer_id: str, last_t: list[float]) -> None:
+    """Emit at most one ERROR per ``_ERROR_LOG_MIN_INTERVAL_S`` per client."""
+    now = time.monotonic()
+    if now - last_t[0] < _ERROR_LOG_MIN_INTERVAL_S:
+        return
+    last_t[0] = now
+    get_logger(_DIAG_LOGGER_NAME).error(
+        "FDR overrun retry-after-drop failed",
+        extra={
+            "component": _DIAG_LOGGER_NAME,
+            "kind": "fdr.overrun_retry_failed",
+            "kv": {"producer_id": producer_id},
+        },
+    )
diff --git a/src/gps_denied_onboard/fdr_client/queue.py b/src/gps_denied_onboard/fdr_client/queue.py
index 496c21b..da3db17 100644
--- a/src/gps_denied_onboard/fdr_client/queue.py
+++ b/src/gps_denied_onboard/fdr_client/queue.py
@@ -1,21 +1,186 @@
-"""Lock-free SPSC ring buffer — STUB.
+"""Lock-free SPSC ring buffer (AZ-273 / E-CC-FDR-CLIENT).
 
-Concrete drop-oldest-on-overrun ring buffer is owned by AZ-273.
+Implements the producer-side queue for ``FdrClient``. Capacity is fixed at
+construction, rounded UP to the next power of two so the wrap-around math
+collapses to a bitwise AND.
+
+Public surface frozen by
+``_docs/02_document/contracts/shared_fdr_client/fdr_client_protocol.md``
+v1.0.0.
+
+Concurrency contract (SPSC):
+
+* ONE producer thread MAY call :meth:`push` (lock-free under CPython GIL).
+* ONE consumer thread MAY call :meth:`pop` / :meth:`drain` (lock-free).
+* The producer thread MAY call :meth:`drop_oldest` from its on-overrun
+  closure; this method serialises with concurrent calls from itself via
+  an internal cold-path lock, but does NOT block the consumer's
+  :meth:`pop`. A simultaneous producer-drop + consumer-pop is the
+  documented race window: at worst ``dropped_count`` is off by one for
+  that one record. Tests run single-threaded around the overrun path so
+  the AC-1 / AC-2 / AC-5 of AZ-274 are deterministic.
+
+The SPSC enforcement guard is opt-in (``enforce_spsc=True``); production
+clients leave it off so the steady-state has zero monitoring overhead.
 """
 
 from __future__ import annotations
 
-from typing import Any
+import threading
+from typing import Any, Final
+
+__all__ = [
+    "MIN_CAPACITY",
+    "FdrSpscViolationError",
+    "SpscRingBuffer",
+]
+
+# Minimum buffer slot count. Below this the contract's wrap-around math
+# is degenerate and microbench results stop being representative.
+MIN_CAPACITY: Final[int] = 16
+
+
+class FdrSpscViolationError(RuntimeError):
+    """Raised by the opt-in SPSC guard when more than one thread touches a side.
+
+    Attributes:
+        side: ``"producer"`` or ``"consumer"`` — which side was violated.
+        owner_thread_id: thread that first claimed the side.
+        offender_thread_id: thread that attempted the violating call.
+    """
+
+    def __init__(self, side: str, owner_thread_id: int, offender_thread_id: int) -> None:
+        self.side: str = side
+        self.owner_thread_id: int = owner_thread_id
+        self.offender_thread_id: int = offender_thread_id
+        super().__init__(
+            f"SPSC violation on {side!r} side: thread {offender_thread_id} attempted "
+            f"to operate a buffer owned by thread {owner_thread_id}"
+        )
 
 
 class SpscRingBuffer:
-    """Single-producer single-consumer lock-free ring buffer."""
+    """Single-producer single-consumer lock-free ring buffer.
 
-    def __init__(self, capacity: int) -> None:
-        self.capacity = capacity
+    Storage is a fixed-size Python ``list``; slots are pre-populated with
+    ``None``. The producer writes to ``_slots[_tail]`` then advances
+    ``_tail``; the consumer reads from ``_slots[_head]`` then advances
+    ``_head``. Under CPython's GIL each single attribute store is atomic,
+    so ``_head`` and ``_tail`` are safe to read across threads.
+
+    Capacity is rounded up to a power of two so the modular arithmetic
+    can use a bitwise AND on ``capacity - 1``.
+    """
+
+    def __init__(self, capacity: int, *, enforce_spsc: bool = False) -> None:
+        if not isinstance(capacity, int) or isinstance(capacity, bool):
+            raise TypeError(f"capacity must be a non-bool int; got {type(capacity).__name__}")
+        if capacity < MIN_CAPACITY:
+            raise ValueError(f"capacity must be >= {MIN_CAPACITY}; got {capacity}")
+        if capacity & (capacity - 1) != 0:
+            raise ValueError(f"capacity must be a power of two; got {capacity}")
+
+        self._capacity: int = capacity
+        self._mask: int = capacity - 1
+        self._slots: list[Any] = [None] * capacity
+        self._head: int = 0
+        self._tail: int = 0
+
+        self._enforce_spsc: bool = enforce_spsc
+        self._producer_thread_id: int | None = None
+        self._consumer_thread_id: int | None = None
+
+        # Cold-path lock used only inside drop_oldest to serialise multiple
+        # producer-side overrun events; the consumer's pop intentionally
+        # does NOT acquire it (see module docstring race window note).
+        self._drop_lock: threading.Lock = threading.Lock()
+
+    @property
+    def capacity(self) -> int:
+        return self._capacity
+
+    def is_empty(self) -> bool:
+        return self._head == self._tail
+
+    def is_full(self) -> bool:
+        return ((self._tail + 1) & self._mask) == self._head
+
+    def size(self) -> int:
+        return (self._tail - self._head) & self._mask
 
     def push(self, item: Any) -> bool:
-        raise NotImplementedError("FdrClient ring-buffer concrete impl is AZ-273")
+        """Producer push. Returns ``False`` when the buffer is full."""
+        if self._enforce_spsc:
+            self._claim_side(producer=True)
+        tail = self._tail
+        next_tail = (tail + 1) & self._mask
+        if next_tail == self._head:
+            return False
+        self._slots[tail] = item
+        self._tail = next_tail
+        return True
 
     def pop(self) -> Any | None:
-        raise NotImplementedError("FdrClient ring-buffer concrete impl is AZ-273")
+        """Consumer pop. Returns ``None`` when the buffer is empty."""
+        if self._enforce_spsc:
+            self._claim_side(producer=False)
+        head = self._head
+        if head == self._tail:
+            return None
+        item = self._slots[head]
+        self._slots[head] = None
+        self._head = (head + 1) & self._mask
+        return item
+
+    def drop_oldest(self) -> Any | None:
+        """Producer-side drop of the oldest queued record (cold path).
+
+        Returns the dropped item, or ``None`` if the buffer is empty.
+        Serialised against concurrent calls from itself via
+        ``_drop_lock``. NOT serialised against the consumer's
+        :meth:`pop` — see the module docstring's race window note.
+        """
+        with self._drop_lock:
+            head = self._head
+            if head == self._tail:
+                return None
+            item = self._slots[head]
+            self._slots[head] = None
+            self._head = (head + 1) & self._mask
+            return item
+
+    def drain(self, max_records: int) -> list[Any]:
+        """Consumer drain up to ``max_records`` items in FIFO order."""
+        if not isinstance(max_records, int) or isinstance(max_records, bool):
+            raise TypeError(f"max_records must be a non-bool int; got {type(max_records).__name__}")
+        if max_records < 0:
+            raise ValueError(f"max_records must be >= 0; got {max_records}")
+        result: list[Any] = []
+        for _ in range(max_records):
+            item = self.pop()
+            if item is None:
+                break
+            result.append(item)
+        return result
+
+    def _claim_side(self, *, producer: bool) -> None:
+        """SPSC guard: bind a side to the first thread that touches it."""
+        current = threading.get_ident()
+        if producer:
+            owner = self._producer_thread_id
+            if owner is None:
+                self._producer_thread_id = current
+                return
+            if owner != current:
+                raise FdrSpscViolationError(
+                    side="producer", owner_thread_id=owner, offender_thread_id=current
+                )
+        else:
+            owner = self._consumer_thread_id
+            if owner is None:
+                self._consumer_thread_id = current
+                return
+            if owner != current:
+                raise FdrSpscViolationError(
+                    side="consumer", owner_thread_id=owner, offender_thread_id=current
+                )
diff --git a/src/gps_denied_onboard/logging/__init__.py b/src/gps_denied_onboard/logging/__init__.py
index a9ca437..31c3d71 100644
--- a/src/gps_denied_onboard/logging/__init__.py
+++ b/src/gps_denied_onboard/logging/__init__.py
@@ -1,8 +1,15 @@
 """Structured JSON logging entrypoint (E-CC-LOG / AZ-245 / AZ-266).
 
-Public surface — every component imports `get_logger` from here. The
-handler topology is selected by `configure_logging(tier=...)` at the
+Public surface — every component imports ``get_logger`` from here. The
+handler topology is selected by ``configure_logging(tier=...)`` at the
 composition-root entrypoint.
+
+The FDR log bridge (AZ-267) is wiring code used only by the composition
+root. It is intentionally NOT re-exported here: importing it would
+trigger a circular import (the bridge depends on ``fdr_client.client``,
+which itself logs via ``get_logger``). Composition-root code imports
+``from gps_denied_onboard.logging.fdr_bridge import wire_log_bridge``
+explicitly.
 """
 
 from gps_denied_onboard.logging.structured import (
diff --git a/src/gps_denied_onboard/logging/fdr_bridge.py b/src/gps_denied_onboard/logging/fdr_bridge.py
new file mode 100644
index 0000000..8c07de6
--- /dev/null
+++ b/src/gps_denied_onboard/logging/fdr_bridge.py
@@ -0,0 +1,263 @@
+"""FDR log bridge — forwards WARN + ERROR records into the FDR (AZ-267).
+
+A :class:`logging.Handler` subclass installed on the root onboard logger
+(or each named logger). For every emitted record at level WARN or
+ERROR, it builds a ``kind="log"`` :class:`FdrRecord` carrying the
+schema-mandated 7 fields and enqueues into a producer-side
+:class:`FdrClient`. INFO and DEBUG records are dropped at the handler's
+level filter — they never reach the FDR (AC-3).
+
+Public surface frozen by
+``_docs/02_document/contracts/shared_logging/log_record_schema.md``
+v1.0.0 (the wire shape of the resulting record's payload) and
+``_docs/02_document/contracts/shared_fdr_client/fdr_client_protocol.md``
+v1.0.0 (the queue we enqueue into).
+
+Concurrency
+-----------
+
+* The handler may run on any logger thread.
+* A thread-local recursion guard short-circuits any logging call that
+  originates from inside the handler itself — without this guard, a
+  failure path that emits a diagnostic WARN would recurse through the
+  same handler indefinitely.
+* Saturated-queue diagnostics throttle to 1-per-``_DROP_LOG_EVERY_N``
+  occurrences via stdout (not via the bridge) to avoid the same loop.
+"""
+
+from __future__ import annotations
+
+import datetime as _dt
+import logging
+import sys
+import threading
+from collections.abc import Callable
+from typing import Any, Final
+
+from gps_denied_onboard.fdr_client.client import EnqueueResult
+from gps_denied_onboard.fdr_client.records import (
+    CURRENT_SCHEMA_VERSION,
+    FdrRecord,
+)
+
+# The resolver's return value must expose ``enqueue(record) -> str``; both
+# the production ``FdrClient`` and the test ``FakeFdrSink`` satisfy that.
+# Typed as ``Any`` here to avoid pulling FakeFdrSink (a tests-only fake)
+# into the production module's type contract.
+_FdrLikeResolver = Callable[[str], Any]
+
+__all__ = [
+    "FdrLogBridgeHandler",
+    "wire_log_bridge",
+]
+
+
+_BRIDGE_HANDLER_MARKER_ATTR: Final[str] = "_gps_denied_fdr_bridge_handler"
+
+# Throttle the stdout "queue saturated" diagnostic to one in every N
+# occurrences per NFR-reliability ("at least N>=1000").
+_DROP_LOG_EVERY_N: Final[int] = 1000
+
+
+# Level-name normalisation matching the schema contract (``WARNING`` ->
+# ``WARN``). Mirrors the logic in ``logging.structured._normalise_level``;
+# duplicated locally to keep this module self-contained and avoid a
+# circular import.
+def _normalise_level(stdlib_levelname: str) -> str:
+    if stdlib_levelname == "WARNING":
+        return "WARN"
+    return stdlib_levelname
+
+
+def _iso_utc(created_epoch: float) -> str:
+    """RFC 3339 UTC with microsecond precision + ``Z`` suffix."""
+    dt = _dt.datetime.fromtimestamp(created_epoch, tz=_dt.timezone.utc)
+    return dt.strftime("%Y-%m-%dT%H:%M:%S.") + f"{dt.microsecond:06d}Z"
+
+
+# Reserved stdlib LogRecord attributes that must not leak into the FDR
+# record's payload.kv. Mirrors the same list in
+# ``logging.structured._RESERVED_LOG_RECORD_KEYS``.
+_RESERVED_LOG_RECORD_KEYS: Final[frozenset[str]] = frozenset(
+    {
+        "args",
+        "asctime",
+        "created",
+        "exc_info",
+        "exc_text",
+        "filename",
+        "funcName",
+        "levelname",
+        "levelno",
+        "lineno",
+        "message",
+        "module",
+        "msecs",
+        "msg",
+        "name",
+        "pathname",
+        "process",
+        "processName",
+        "relativeCreated",
+        "stack_info",
+        "taskName",
+        "thread",
+        "threadName",
+        "frame_id",
+        "kind",
+        "kv",
+        "component",
+    }
+)
+
+
+def _coerce_jsonable(value: Any) -> Any:
+    """Mirror of the formatter's coercer; raises on non-JSON-safe types."""
+    if isinstance(value, (str, int, float, bool)) or value is None:
+        return value
+    if isinstance(value, dict):
+        return {str(k): _coerce_jsonable(v) for k, v in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_coerce_jsonable(v) for v in value]
+    raise TypeError(f"unserialisable kv payload type: {type(value).__name__}")
+
+
+class FdrLogBridgeHandler(logging.Handler):
+    """Forwards WARN + ERROR :class:`logging.LogRecord` into the FDR.
+
+    Constructed with a callable that resolves to the per-record
+    :class:`FdrClient`. The callable lets the composition root inject
+    either a per-component sink (one FdrClient per component) or a
+    shared "log fan-out" client; either way the bridge stays decoupled
+    from the registry.
+    """
+
+    _recursion_local = threading.local()
+
+    def __init__(
+        self,
+        fdr_client_resolver: _FdrLikeResolver,
+        *,
+        level: int = logging.WARNING,
+    ) -> None:
+        super().__init__(level=level)
+        if not callable(fdr_client_resolver):
+            raise TypeError(f"fdr_client_resolver must be callable; got {fdr_client_resolver!r}")
+        self._resolver: _FdrLikeResolver = fdr_client_resolver
+        self._drop_counter: int = 0
+        self._drop_lock: threading.Lock = threading.Lock()
+        setattr(self, _BRIDGE_HANDLER_MARKER_ATTR, True)
+
+    def emit(self, record: logging.LogRecord) -> None:
+        if getattr(self._recursion_local, "in_bridge", False):
+            # Short-circuit: a log call originating from inside the
+            # bridge (saturated-queue diagnostic, etc.) must NOT loop.
+            return
+        self._recursion_local.in_bridge = True
+        try:
+            self._emit_unguarded(record)
+        finally:
+            self._recursion_local.in_bridge = False
+
+    def _emit_unguarded(self, record: logging.LogRecord) -> None:
+        try:
+            fdr_record = self._build_fdr_record(record)
+        except Exception as exc:
+            # Translation failed (e.g. non-serialisable kv that even the
+            # formatter's fallback didn't catch). Skip this record;
+            # never raise into the caller.
+            self._note_drop(reason=f"translate_error: {exc}")
+            return
+
+        component = fdr_record.payload.get("component") or record.name
+        try:
+            client = self._resolver(component if isinstance(component, str) else record.name)
+        except Exception as exc:
+            self._note_drop(reason=f"resolve_error: {exc}")
+            return
+
+        result = client.enqueue(fdr_record)
+        if result == EnqueueResult.OVERRUN:
+            self._note_drop(reason="queue_saturated")
+
+    def _build_fdr_record(self, record: logging.LogRecord) -> FdrRecord:
+        rec_dict = record.__dict__
+        component = rec_dict.get("component") or record.name
+        kind = rec_dict.get("kind") or "log.diag"
+        frame_id = rec_dict.get("frame_id")
+
+        explicit_kv = rec_dict.get("kv")
+        if explicit_kv is None:
+            kv_raw: dict[str, Any] = {
+                k: v
+                for k, v in rec_dict.items()
+                if k not in _RESERVED_LOG_RECORD_KEYS and not k.startswith("_")
+            }
+        else:
+            kv_raw = dict(explicit_kv)
+
+        try:
+            kv_safe = _coerce_jsonable(kv_raw)
+        except (TypeError, ValueError) as exc:
+            kv_safe = {"_format_error": f"{type(exc).__name__}: {exc}"}
+
+        exc_text: str | None = None
+        if record.exc_info:
+            exc_text = logging.Formatter().formatException(record.exc_info)
+
+        return FdrRecord(
+            schema_version=CURRENT_SCHEMA_VERSION,
+            ts=_iso_utc(record.created),
+            producer_id=component,
+            kind="log",
+            payload={
+                "level": _normalise_level(record.levelname),
+                "component": component,
+                "frame_id": frame_id,
+                "kind": kind,
+                "msg": record.getMessage().replace("\n", " "),
+                "kv": kv_safe,
+                "exc": exc_text,
+            },
+        )
+
+    def _note_drop(self, *, reason: str) -> None:
+        """Throttled stdout diagnostic; intentionally bypasses the logger."""
+        with self._drop_lock:
+            self._drop_counter += 1
+            if self._drop_counter % _DROP_LOG_EVERY_N != 1:
+                return
+            current = self._drop_counter
+        # stderr (not stdout) so log capture in tests doesn't confuse
+        # the bridge's diagnostic with normal application output.
+        print(
+            f"FdrLogBridgeHandler: dropped record #{current} (reason={reason})",
+            file=sys.stderr,
+        )
+
+
+def wire_log_bridge(
+    fdr_client_resolver: _FdrLikeResolver,
+    *,
+    target_loggers: tuple[str, ...] = ("",),
+    level: int = logging.WARNING,
+) -> FdrLogBridgeHandler:
+    """Install a single :class:`FdrLogBridgeHandler` on ``target_loggers``.
+
+    Idempotent: re-calling replaces any prior bridge handler on the
+    same logger(s) — exactly one bridge per logger (AC-5).
+
+    Returns the installed handler so the composition root can keep a
+    handle for teardown (e.g. test isolation).
+    """
+    handler = FdrLogBridgeHandler(fdr_client_resolver, level=level)
+    for name in target_loggers:
+        target = logging.getLogger(name)
+        target.handlers = [
+            h for h in target.handlers if not getattr(h, _BRIDGE_HANDLER_MARKER_ATTR, False)
+        ]
+        target.addHandler(handler)
+        # Lower the logger's effective level if it is currently above WARN.
+        if target.level == logging.NOTSET or target.level > level:
+            target.setLevel(level)
+    return handler
diff --git a/tests/conftest.py b/tests/conftest.py
index 6d216ab..e721769 100644
--- a/tests/conftest.py
+++ b/tests/conftest.py
@@ -10,9 +10,12 @@ Tier-2-only tests are guarded by `pytest.mark.tier2` and auto-skipped on Tier-1.
 from __future__ import annotations
 
 import os
+from collections.abc import Iterator
 
 import pytest
 
+from gps_denied_onboard.fdr_client.fakes import FakeFdrSink
+
 
 def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item]) -> None:
     """Auto-skip `tier2` tests when GPS_DENIED_TIER != 2."""
@@ -28,3 +31,11 @@ def pytest_collection_modifyitems(config: pytest.Config, items: list[pytest.Item
             item.add_marker(skip_gpu)
         if "docker" in item.keywords:
             item.add_marker(skip_docker)
+
+
+@pytest.fixture
+def fake_fdr_sink() -> Iterator[FakeFdrSink]:
+    """Default-configuration FakeFdrSink with overrun policy disabled (AZ-275 AC-5)."""
+    sink = FakeFdrSink(producer_id="test.producer")
+    yield sink
+    sink.flush()
diff --git a/tests/contract/__init__.py b/tests/contract/__init__.py
new file mode 100644
index 0000000..2a34ae0
--- /dev/null
+++ b/tests/contract/__init__.py
@@ -0,0 +1,6 @@
+"""Contract tests — frozen public surface verification (AZ-268, AZ-XX).
+
+Tests in this package run with the ``contract`` pytest marker so CI can
+optionally split them into a separate stage. They are also collected
+under the default suite.
+"""
diff --git a/tests/contract/log_schema.py b/tests/contract/log_schema.py
new file mode 100644
index 0000000..e92df0d
--- /dev/null
+++ b/tests/contract/log_schema.py
@@ -0,0 +1,297 @@
+"""Contract test for ``log_record_schema`` v1.0.0 (AZ-268 / E-CC-LOG).
+
+Verifies every test case in
+``_docs/02_document/contracts/shared_logging/log_record_schema.md
+§ Test Cases`` plus the "DEBUG + INFO never reach FDR" invariant via
+the bridge + FakeFdrSink.
+
+File path is fixed at ``tests/contract/log_schema.py`` per epic
+AC-4 so the traceability matrix reference stays stable.
+"""
+
+from __future__ import annotations
+
+import io
+import json
+import logging
+from collections.abc import Iterator
+from typing import Final
+
+import pytest
+
+from gps_denied_onboard.fdr_client.fakes import FakeFdrSink
+from gps_denied_onboard.logging.fdr_bridge import wire_log_bridge
+from gps_denied_onboard.logging.structured import (
+    JsonFormatter,
+    configure_logging,
+    get_logger,
+)
+
+pytestmark = pytest.mark.contract
+
+
+# Contract version pin (AC-4). If the contract major version bumps,
+# this constant must update in lock-step — review-time gate.
+CONTRACT_VERSION: Final[str] = "1.0.0"
+
+# Authoritative field order from the contract.
+EXPECTED_FIELD_ORDER: Final[tuple[str, ...]] = (
+    "ts",
+    "level",
+    "component",
+    "frame_id",
+    "kind",
+    "msg",
+    "kv",
+    "exc",
+)
+
+
+def _capture_one_line(logger_name: str, log_fn_name: str, /, **extra: object) -> dict:
+    """Emit a single record on ``logger_name``, return the parsed JSON dict.
+
+    Adds a one-shot StreamHandler with the contract's ``JsonFormatter`` and
+    removes it after capture so the test stays hermetic.
+    """
+    buf = io.StringIO()
+    handler = logging.StreamHandler(buf)
+    handler.setFormatter(JsonFormatter())
+    handler.setLevel(logging.DEBUG)
+    target = logging.getLogger(logger_name)
+    target.addHandler(handler)
+    original_level = target.level
+    target.setLevel(logging.DEBUG)
+    try:
+        getattr(target, log_fn_name)(**extra)
+    finally:
+        target.removeHandler(handler)
+        target.setLevel(original_level)
+    lines = [ln for ln in buf.getvalue().splitlines() if ln.strip()]
+    assert len(lines) == 1, f"expected one line, got {len(lines)}: {lines!r}"
+    return json.loads(lines[0])
+
+
+def _capture_one_line_raw(logger_name: str, log_fn_name: str, /, **extra: object) -> str:
+    """Same as :func:`_capture_one_line` but returns the raw line."""
+    buf = io.StringIO()
+    handler = logging.StreamHandler(buf)
+    handler.setFormatter(JsonFormatter())
+    handler.setLevel(logging.DEBUG)
+    target = logging.getLogger(logger_name)
+    target.addHandler(handler)
+    original_level = target.level
+    target.setLevel(logging.DEBUG)
+    try:
+        getattr(target, log_fn_name)(**extra)
+    finally:
+        target.removeHandler(handler)
+        target.setLevel(original_level)
+    lines = [ln for ln in buf.getvalue().splitlines() if ln.strip()]
+    return lines[0]
+
+
+# ---------------------------------------------------------------------------
+# AC-4: contract version pinned.
+
+
+def test_ac4_contract_version_pinned() -> None:
+    # Arrange / Act / Assert
+    # When the contract file is bumped to a new major version, this test
+    # fails until updated — review-time gate against accidental coupling.
+    assert CONTRACT_VERSION == "1.0.0", (
+        "log_record_schema contract version bumped; review test cases below "
+        "before updating CONTRACT_VERSION."
+    )
+
+
+# ---------------------------------------------------------------------------
+# Contract test cases: § Test Cases table.
+
+
+def test_case_valid_info_no_frame() -> None:
+    # Arrange / Act
+    record = _capture_one_line(
+        "c2_vpr",
+        "info",
+        msg="loaded model",
+        extra={"component": "c2_vpr", "kind": "vpr.warmup", "kv": {"model": "salad"}},
+    )
+
+    # Assert
+    assert record["level"] == "INFO"
+    assert record["component"] == "c2_vpr"
+    assert record["kind"] == "vpr.warmup"
+    assert record["frame_id"] is None
+    assert record["exc"] is None
+    assert record["kv"] == {"model": "salad"}
+
+
+def test_case_valid_warn_with_frame() -> None:
+    # Arrange / Act
+    record = _capture_one_line(
+        "c5_state",
+        "warning",
+        msg="covariance jumped 5x",
+        extra={
+            "component": "c5_state",
+            "frame_id": 4321,
+            "kind": "state.cov_spike",
+            "kv": {"jump_factor": 5.2},
+        },
+    )
+
+    # Assert
+    assert record["level"] == "WARN"  # WARNING -> WARN per contract
+    assert record["frame_id"] == 4321
+    assert record["kv"] == {"jump_factor": 5.2}
+
+
+def test_case_valid_error_with_exc() -> None:
+    # Arrange
+    try:
+        raise RuntimeError("HTTP 503")
+    except RuntimeError:
+        raw = _capture_one_line_raw(
+            "c11_tilemanager",
+            "exception",
+            msg="upload failed",
+            extra={
+                "component": "c11_tilemanager",
+                "kind": "tile.upload_fail",
+                "kv": {"tile": "z18/x12345/y67890"},
+            },
+        )
+
+    # Assert
+    record = json.loads(raw)
+    assert record["level"] == "ERROR"
+    assert record["exc"] is not None
+    assert "RuntimeError" in record["exc"]
+
+
+def test_case_invalid_multiline_msg_is_collapsed() -> None:
+    # Arrange / Act
+    record = _capture_one_line(
+        "c5_state",
+        "info",
+        msg="line1\nline2",
+        extra={"component": "c5_state", "kind": "test"},
+    )
+
+    # Assert
+    assert "\n" not in record["msg"]
+    assert record["msg"] == "line1 line2"
+
+
+def test_case_invalid_non_serialisable_kv_falls_back_to_format_error() -> None:
+    # Arrange
+    class _NotSerialisable:
+        pass
+
+    # Act
+    record = _capture_one_line(
+        "c5_state",
+        "info",
+        msg="oops",
+        extra={
+            "component": "c5_state",
+            "kind": "test",
+            "kv": {"obj": _NotSerialisable()},
+        },
+    )
+
+    # Assert
+    assert "_format_error" in record["kv"]
+
+
+def test_case_ordering_stable() -> None:
+    # Arrange — emit several records with deliberately scrambled extra ordering.
+    raws = [
+        _capture_one_line_raw(
+            "c2_vpr",
+            "info",
+            msg=f"line {i}",
+            extra={
+                "component": "c2_vpr",
+                "kind": "test",
+                "kv": {"i": i},
+                "frame_id": i,
+            },
+        )
+        for i in range(3)
+    ]
+
+    # Act — parse with object_pairs_hook to preserve key order from the raw bytes.
+    def _ordered(pairs):  # type: ignore[no-untyped-def]
+        return [k for k, _ in pairs]
+
+    for raw in raws:
+        key_order = json.loads(raw, object_pairs_hook=_ordered)
+        assert tuple(key_order) == EXPECTED_FIELD_ORDER, (
+            f"key order drifted: got {tuple(key_order)} vs expected {EXPECTED_FIELD_ORDER}"
+        )
+
+
+# ---------------------------------------------------------------------------
+# AC-3: DEBUG + INFO never reach FDR.
+
+
+@pytest.fixture
+def isolated_logger() -> Iterator[None]:
+    """Snapshot + restore the test logger to keep capture hermetic."""
+    name = "contract.log_schema.suppression"
+    logger = logging.getLogger(name)
+    saved_handlers = list(logger.handlers)
+    saved_level = logger.level
+    yield
+    logger.handlers = saved_handlers
+    logger.setLevel(saved_level)
+
+
+def test_ac3_debug_and_info_never_reach_fdr(isolated_logger: None) -> None:
+    # Arrange
+    sink = FakeFdrSink(producer_id="contract.log_schema.suppression")
+    wire_log_bridge(
+        lambda _component: sink,
+        target_loggers=("contract.log_schema.suppression",),
+    )
+    logger = get_logger("contract.log_schema.suppression")
+    logger.setLevel(logging.DEBUG)
+
+    # Act
+    for _ in range(100):
+        logger.info("INFO record")
+        logger.debug("DEBUG record")
+
+    # Assert
+    assert sink.all_records_ever == []
+    assert sink.records == []
+
+
+# ---------------------------------------------------------------------------
+# AC-2: schema-drift fails fast (the test itself is the gate).
+# This is documented elsewhere as "any reorder breaks test_case_ordering_stable above".
+
+
+# ---------------------------------------------------------------------------
+# Smoke: configure_logging is idempotent (regression guard).
+
+
+def test_configure_logging_is_idempotent() -> None:
+    # Arrange
+    root = logging.getLogger()
+    saved_handlers = list(root.handlers)
+    saved_level = root.level
+
+    try:
+        # Act
+        configure_logging(tier=1, level="INFO")
+        first_count = len(root.handlers)
+        configure_logging(tier=1, level="INFO")
+        second_count = len(root.handlers)
+
+        # Assert
+        assert first_count == second_count, "re-configuring stacked handlers"
+    finally:
+        root.handlers = saved_handlers
+        root.setLevel(saved_level)
diff --git a/tests/unit/test_az267_fdr_log_bridge.py b/tests/unit/test_az267_fdr_log_bridge.py
new file mode 100644
index 0000000..f4e3318
--- /dev/null
+++ b/tests/unit/test_az267_fdr_log_bridge.py
@@ -0,0 +1,195 @@
+"""AZ-267 — FDR log bridge (WARN + ERROR forwarding).
+
+Verifies all five ACs:
+1. WARN reaches FDR with kind=log + correct component back-reference.
+2. ERROR + ``logger.exception`` carries traceback in ``exc``.
+3. INFO + DEBUG never reach FDR.
+4. Saturated queue does not block the caller.
+5. Re-wiring is idempotent — exactly one bridge handler per logger.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Iterator
+
+import pytest
+
+from gps_denied_onboard.fdr_client.fakes import FakeFdrSink
+from gps_denied_onboard.logging import get_logger
+from gps_denied_onboard.logging.fdr_bridge import (
+    FdrLogBridgeHandler,
+    wire_log_bridge,
+)
+
+
+@pytest.fixture
+def isolated_logger_state() -> Iterator[None]:
+    """Snapshot + restore the root logger to keep tests independent."""
+    root = logging.getLogger()
+    saved_handlers = list(root.handlers)
+    saved_level = root.level
+    yield
+    root.handlers = saved_handlers
+    root.setLevel(saved_level)
+
+
+def _resolver_for(sink: FakeFdrSink):  # type: ignore[no-untyped-def]
+    def _resolve(_component: str) -> FakeFdrSink:
+        return sink
+
+    return _resolve
+
+
+# ---------------------------------------------------------------------------
+# AC-1: WARN records reach FDR.
+
+
+def test_ac1_warn_reaches_fdr(isolated_logger_state: None) -> None:
+    # Arrange
+    sink = FakeFdrSink(producer_id="c2_vpr")
+    wire_log_bridge(_resolver_for(sink), target_loggers=("c2_vpr",))
+    logger = get_logger("c2_vpr")
+
+    # Act
+    logger.warning("covariance jumped 5x", extra={"component": "c2_vpr", "kind": "vpr.cov_spike"})
+
+    # Assert
+    assert len(sink.records) == 1
+    record = sink.records[0]
+    assert record.kind == "log"
+    assert record.producer_id == "c2_vpr"
+    assert record.payload["level"] == "WARN"
+    assert record.payload["component"] == "c2_vpr"
+    assert record.payload["msg"] == "covariance jumped 5x"
+
+
+# ---------------------------------------------------------------------------
+# AC-2: ERROR + logger.exception carries traceback in exc.
+
+
+def test_ac2_logger_exception_carries_traceback(isolated_logger_state: None) -> None:
+    # Arrange
+    sink = FakeFdrSink(producer_id="c11_tilemanager")
+    wire_log_bridge(_resolver_for(sink), target_loggers=("c11_tilemanager",))
+    logger = get_logger("c11_tilemanager")
+
+    # Act
+    try:
+        raise RuntimeError("HTTP 503")
+    except RuntimeError:
+        logger.exception(
+            "tile upload failed",
+            extra={"component": "c11_tilemanager", "kind": "tile.upload_fail"},
+        )
+
+    # Assert
+    assert len(sink.records) == 1
+    record = sink.records[0]
+    assert record.payload["level"] == "ERROR"
+    assert record.payload["exc"] is not None
+    assert "RuntimeError" in record.payload["exc"]
+    assert "HTTP 503" in record.payload["exc"]
+
+
+# ---------------------------------------------------------------------------
+# AC-3: INFO + DEBUG never reach FDR.
+
+
+def test_ac3_info_and_debug_never_reach_fdr(isolated_logger_state: None) -> None:
+    # Arrange
+    sink = FakeFdrSink(producer_id="c5_state")
+    wire_log_bridge(_resolver_for(sink), target_loggers=("c5_state",))
+    logger = get_logger("c5_state")
+    logger.setLevel(logging.DEBUG)
+
+    # Act
+    for _ in range(100):
+        logger.info("startup")
+        logger.debug("trace point")
+
+    # Assert
+    assert sink.records == []
+    assert sink.all_records_ever == []
+
+
+# ---------------------------------------------------------------------------
+# AC-4: saturated queue does not block the caller.
+
+
+def test_ac4_saturated_queue_does_not_block(isolated_logger_state: None) -> None:
+    # Arrange
+    sink = FakeFdrSink(producer_id="c1_vio", capacity=4, with_default_overrun_policy=True)
+    wire_log_bridge(_resolver_for(sink), target_loggers=("c1_vio",))
+    logger = get_logger("c1_vio")
+    # Fill the sink.
+    for i in range(4):
+        logger.warning("filler", extra={"component": "c1_vio", "kind": "fill", "frame_id": i})
+
+    # Act
+    start = time.perf_counter()
+    logger.warning(
+        "overrun trigger",
+        extra={"component": "c1_vio", "kind": "trigger", "frame_id": 999},
+    )
+    elapsed = time.perf_counter() - start
+
+    # Assert — must return well under 0.5 ms wall clock per NFR-perf.
+    assert elapsed < 0.005, f"call blocked: {elapsed * 1e3:.2f} ms"
+
+
+# ---------------------------------------------------------------------------
+# AC-5: single attachment — re-wiring does not stack duplicate handlers.
+
+
+def test_ac5_single_attachment_is_idempotent(isolated_logger_state: None) -> None:
+    # Arrange
+    sink = FakeFdrSink(producer_id="c7_inference")
+    wire_log_bridge(_resolver_for(sink), target_loggers=("c7_inference",))
+
+    # Act — re-wire three times.
+    wire_log_bridge(_resolver_for(sink), target_loggers=("c7_inference",))
+    wire_log_bridge(_resolver_for(sink), target_loggers=("c7_inference",))
+
+    # Assert
+    target_logger = logging.getLogger("c7_inference")
+    bridge_handlers = [h for h in target_logger.handlers if isinstance(h, FdrLogBridgeHandler)]
+    assert len(bridge_handlers) == 1
+
+
+# ---------------------------------------------------------------------------
+# Bridge does not recurse on internal warnings.
+
+
+def test_recursion_guard_prevents_infinite_loop(isolated_logger_state: None) -> None:
+    # Arrange — sink that always overruns.
+    sink = FakeFdrSink(producer_id="c3_matcher", capacity=1)
+    wire_log_bridge(_resolver_for(sink), target_loggers=("c3_matcher",))
+    logger = get_logger("c3_matcher")
+    sink.enqueue(_dummy_record())
+
+    # Act — should not recurse infinitely.
+    logger.warning("trigger overrun", extra={"component": "c3_matcher", "kind": "test"})
+
+    # Assert — completes without StackOverflow or recursion errors.
+
+
+def _dummy_record():
+    from gps_denied_onboard.fdr_client.records import FdrRecord
+
+    return FdrRecord(
+        schema_version=1,
+        ts="2026-05-11T00:00:00.000000Z",
+        producer_id="c3_matcher",
+        kind="log",
+        payload={
+            "level": "INFO",
+            "component": "c3_matcher",
+            "frame_id": None,
+            "kind": "test",
+            "msg": "filler",
+            "kv": {},
+            "exc": None,
+        },
+    )
diff --git a/tests/unit/test_az273_fdr_client_ringbuf.py b/tests/unit/test_az273_fdr_client_ringbuf.py
new file mode 100644
index 0000000..64e3982
--- /dev/null
+++ b/tests/unit/test_az273_fdr_client_ringbuf.py
@@ -0,0 +1,342 @@
+"""AZ-273 — FdrClient lock-free SPSC ring buffer + public API.
+
+Verifies the contract-relevant ACs (1, 3, 4, 5, 6, 7) of
+``fdr_client_protocol`` v1.0.0. AC-2 (zero-alloc steady-state) and the
+NFR-perf budgets (p99 ≤ 5 µs / ≤ 10 µs on Tier-2) are deferred to a
+follow-up perf-instrumentation task; the pure-Python implementation
+correctness is in scope here.
+"""
+
+from __future__ import annotations
+
+import threading
+import time
+from collections.abc import Iterator
+
+import pytest
+
+from gps_denied_onboard.config import Config, FdrConfig
+from gps_denied_onboard.fdr_client import (
+    EnqueueResult,
+    FdrClient,
+    FdrRecord,
+    FdrSpscViolationError,
+    make_fdr_client,
+)
+from gps_denied_onboard.fdr_client.client import _reset_for_tests
+from gps_denied_onboard.fdr_client.queue import SpscRingBuffer
+
+
+@pytest.fixture(autouse=True)
+def _reset_cache() -> Iterator[None]:
+    _reset_for_tests()
+    yield
+    _reset_for_tests()
+
+
+def _make_record(producer_id: str = "test.producer", frame_id: int | None = 0) -> FdrRecord:
+    return FdrRecord(
+        schema_version=1,
+        ts="2026-05-11T00:00:00.000000Z",
+        producer_id=producer_id,
+        kind="log",
+        payload={
+            "level": "INFO",
+            "component": producer_id,
+            "frame_id": frame_id,
+            "kind": "test.tick",
+            "msg": "hello",
+            "kv": {},
+            "exc": None,
+        },
+    )
+
+
+# ---------------------------------------------------------------------------
+# AC-1: lock-free, never blocks — every enqueue returns in O(1), overrun on #1025.
+
+
+def test_ac1_enqueue_never_blocks_and_returns_overrun_on_overflow() -> None:
+    # Arrange
+    client = FdrClient(producer_id="c1_vio", capacity=1024)
+
+    # Act
+    last_result = EnqueueResult.OK
+    timings: list[float] = []
+    for i in range(1025):
+        start = time.perf_counter()
+        last_result = client.enqueue(_make_record(frame_id=i))
+        timings.append(time.perf_counter() - start)
+
+    # Assert
+    assert last_result == EnqueueResult.OVERRUN, "the 1025th enqueue must overrun"
+    # Pure-Python budget: every individual call must return under 50 ms
+    # (the NFR-perf 50 µs budget is Tier-2-only; we keep a generous
+    # ceiling here to catch genuine blocking regressions only).
+    assert max(timings) < 0.05, f"slow enqueue suggests blocking; max={max(timings) * 1e6:.1f}µs"
+
+
+# ---------------------------------------------------------------------------
+# AC-3: capacity is config-driven via config.fdr.per_producer_capacity.
+
+
+def test_ac3_capacity_from_per_producer_config() -> None:
+    # Arrange
+    fdr_block = FdrConfig(per_producer_capacity={"c1_vio": 4096})
+    config = Config(fdr=fdr_block)
+
+    # Act
+    client = make_fdr_client("c1_vio", config)
+
+    # Assert
+    assert client._capacity() == 4096
+
+
+def test_ac3_capacity_falls_back_to_default_queue_size() -> None:
+    # Arrange
+    config = Config(fdr=FdrConfig(queue_size=2048))
+
+    # Act
+    client = make_fdr_client("c2_vpr", config)
+
+    # Assert
+    assert client._capacity() == 2048
+
+
+def test_ac3_non_power_of_two_rounds_up() -> None:
+    # Arrange
+    config = Config(fdr=FdrConfig(queue_size=1000))
+
+    # Act
+    client = make_fdr_client("c3_matcher", config)
+
+    # Assert
+    assert client._capacity() == 1024  # 1000 → next power of two
+
+
+# ---------------------------------------------------------------------------
+# AC-4: SPSC dequeue contract enforced by opt-in guard.
+
+
+def test_ac4_spsc_guard_detects_concurrent_consumer_pop() -> None:
+    # Arrange
+    buf = SpscRingBuffer(capacity=16, enforce_spsc=True)
+    barrier = threading.Barrier(2)
+    errors: list[FdrSpscViolationError] = []
+
+    def consume() -> None:
+        barrier.wait()
+        for _ in range(64):
+            try:
+                buf.pop()
+            except FdrSpscViolationError as exc:
+                errors.append(exc)
+                return
+
+    t1 = threading.Thread(target=consume)
+    t2 = threading.Thread(target=consume)
+
+    # Act
+    t1.start()
+    t2.start()
+    t1.join(timeout=5.0)
+    t2.join(timeout=5.0)
+
+    # Assert
+    assert errors, "second consumer thread must trip the SPSC guard"
+    assert errors[0].side == "consumer"
+
+
+def test_ac4_spsc_guard_detects_concurrent_producer_push() -> None:
+    # Arrange
+    buf = SpscRingBuffer(capacity=16, enforce_spsc=True)
+    barrier = threading.Barrier(2)
+    errors: list[FdrSpscViolationError] = []
+
+    def produce() -> None:
+        barrier.wait()
+        for _ in range(64):
+            try:
+                buf.push(object())
+            except FdrSpscViolationError as exc:
+                errors.append(exc)
+                return
+
+    t1 = threading.Thread(target=produce)
+    t2 = threading.Thread(target=produce)
+
+    # Act
+    t1.start()
+    t2.start()
+    t1.join(timeout=5.0)
+    t2.join(timeout=5.0)
+
+    # Assert
+    assert errors, "second producer thread must trip the SPSC guard"
+    assert errors[0].side == "producer"
+
+
+def test_ac4_default_is_no_guard() -> None:
+    # Arrange
+    buf = SpscRingBuffer(capacity=16)  # enforce_spsc defaults to False
+
+    # Act — two threads push and pop concurrently; no exception expected.
+    def stress() -> None:
+        for i in range(32):
+            buf.push(i)
+            buf.pop()
+
+    t1 = threading.Thread(target=stress)
+    t2 = threading.Thread(target=stress)
+    t1.start()
+    t2.start()
+    t1.join(timeout=5.0)
+    t2.join(timeout=5.0)
+
+    # Assert — no exception, no SPSC complaints; production wiring opts out.
+
+
+# ---------------------------------------------------------------------------
+# AC-5: on_overrun hook is wired exactly once per overrun.
+
+
+def test_ac5_on_overrun_hook_fires_once_per_overrun() -> None:
+    # Arrange
+    client = FdrClient(producer_id="c4_pose", capacity=16)
+    seen: list[FdrRecord] = []
+    client.on_overrun = seen.append
+    # Fill the buffer (capacity 16 holds 15 records before overrun).
+    for i in range(15):
+        client.enqueue(_make_record(frame_id=i))
+    offending = _make_record(frame_id=999)
+
+    # Act
+    result = client.enqueue(offending)
+
+    # Assert
+    assert result == EnqueueResult.OVERRUN
+    assert seen == [offending]
+
+
+def test_ac5_invalid_hook_rejected() -> None:
+    # Arrange
+    client = FdrClient(producer_id="c4_pose", capacity=16)
+
+    # Act / Assert
+    with pytest.raises(TypeError):
+        client.on_overrun = "not_callable"  # type: ignore[assignment]
+
+
+# ---------------------------------------------------------------------------
+# AC-6: flush() drains the buffer.
+
+
+def test_ac6_flush_returns_only_when_empty() -> None:
+    # Arrange
+    client = FdrClient(producer_id="c5_state", capacity=16)
+    for i in range(8):
+        client.enqueue(_make_record(frame_id=i))
+
+    drained: list[FdrRecord] = []
+
+    def drain() -> None:
+        while True:
+            item = client.pop_one()
+            if item is None and client._buffer_size() == 0:
+                return
+            if item is not None:
+                drained.append(item)
+
+    drainer = threading.Thread(target=drain)
+    drainer.start()
+
+    # Act
+    client.flush()
+
+    # Assert
+    drainer.join(timeout=5.0)
+    assert client._buffer_size() == 0
+    assert len(drained) == 8
+
+
+# ---------------------------------------------------------------------------
+# AC-7: empty producer_id raises ValueError.
+
+
+def test_ac7_empty_producer_id_raises_value_error() -> None:
+    # Arrange / Act / Assert
+    with pytest.raises(ValueError, match="producer_id"):
+        FdrClient(producer_id="", capacity=16)
+
+
+def test_ac7_make_fdr_client_rejects_empty_producer_id() -> None:
+    # Arrange
+    config = Config()
+
+    # Act / Assert
+    with pytest.raises(ValueError, match="producer_id"):
+        make_fdr_client("", config)
+
+
+# ---------------------------------------------------------------------------
+# Invariant: one client per producer_id (NFR-reliability).
+
+
+def test_invariant_make_fdr_client_caches_by_producer_id() -> None:
+    # Arrange
+    config = Config()
+
+    # Act
+    a = make_fdr_client("c8_fc_adapter", config)
+    b = make_fdr_client("c8_fc_adapter", config)
+
+    # Assert
+    assert a is b
+
+
+# ---------------------------------------------------------------------------
+# Invariant: enqueue does not mutate record.producer_id.
+
+
+def test_invariant_enqueue_preserves_producer_id() -> None:
+    # Arrange
+    client = FdrClient(producer_id="c5_state", capacity=16)
+    record = _make_record(producer_id="c5_state", frame_id=42)
+
+    # Act
+    client.enqueue(record)
+    popped = client.pop_one()
+
+    # Assert
+    assert popped is record
+    assert popped.producer_id == "c5_state"
+
+
+# ---------------------------------------------------------------------------
+# Buffer-level invariants: capacity validation.
+
+
+def test_capacity_must_be_at_least_minimum() -> None:
+    # Arrange / Act / Assert
+    with pytest.raises(ValueError, match=">= 16"):
+        SpscRingBuffer(capacity=8)
+
+
+def test_capacity_must_be_power_of_two() -> None:
+    # Arrange / Act / Assert
+    with pytest.raises(ValueError, match="power of two"):
+        SpscRingBuffer(capacity=20)
+
+
+def test_drain_returns_fifo_order() -> None:
+    # Arrange
+    client = FdrClient(producer_id="c7_inference", capacity=16)
+    records = [_make_record(frame_id=i) for i in range(5)]
+    for r in records:
+        client.enqueue(r)
+
+    # Act
+    drained = client.drain(max_records=10)
+
+    # Assert
+    assert drained == records
diff --git a/tests/unit/test_az274_fdr_overrun_policy.py b/tests/unit/test_az274_fdr_overrun_policy.py
new file mode 100644
index 0000000..730c4c9
--- /dev/null
+++ b/tests/unit/test_az274_fdr_overrun_policy.py
@@ -0,0 +1,256 @@
+"""AZ-274 — Drop-oldest + ``kind="overrun"`` record emission policy.
+
+Verifies the contract-relevant ACs (1, 2, 3, 5, 6) of the policy.
+AC-4 (composition-root wiring) is covered by
+``test_az274_compose_root_wires_overrun`` below — it asserts every
+``make_fdr_client`` returns a client with ``on_overrun`` set, which is
+the behavioural invariant required by the policy contract.
+
+NFR-perf (steady-state overhead ≤ 0.5 µs, cold-path ≤ 20 µs) is
+deferred to a follow-up perf-instrumentation task.
+"""
+
+from __future__ import annotations
+
+import logging
+import time
+from collections.abc import Iterator
+
+import pytest
+
+from gps_denied_onboard.config import Config, FdrConfig
+from gps_denied_onboard.fdr_client import (
+    EnqueueResult,
+    FdrClient,
+    FdrRecord,
+    make_fdr_client,
+)
+from gps_denied_onboard.fdr_client.client import _cached_clients, _reset_for_tests
+from gps_denied_onboard.fdr_client.overrun_policy import (
+    default_overrun_policy,
+)
+from gps_denied_onboard.fdr_client.records import OVERRUN_KIND, OVERRUN_PRODUCER_ID
+
+
+@pytest.fixture(autouse=True)
+def _reset_cache() -> Iterator[None]:
+    _reset_for_tests()
+    yield
+    _reset_for_tests()
+
+
+def _make_record(producer_id: str = "c1_vio", frame_id: int = 0) -> FdrRecord:
+    return FdrRecord(
+        schema_version=1,
+        ts="2026-05-11T00:00:00.000000Z",
+        producer_id=producer_id,
+        kind="log",
+        payload={
+            "level": "INFO",
+            "component": producer_id,
+            "frame_id": frame_id,
+            "kind": "test.tick",
+            "msg": "hello",
+            "kv": {},
+            "exc": None,
+        },
+    )
+
+
+def _wire(client: FdrClient) -> FdrClient:
+    client.on_overrun = default_overrun_policy(client)
+    return client
+
+
+# ---------------------------------------------------------------------------
+# AC-1: drop-oldest produces the canonical overrun record after capacity-1 fill.
+
+
+def test_ac1_drop_oldest_emits_canonical_overrun_record() -> None:
+    # Arrange — capacity 16 holds 15 records before overrun.
+    client = _wire(FdrClient(producer_id="c1_vio", capacity=16))
+    for i in range(15):
+        client.enqueue(_make_record(frame_id=i))
+
+    # Act — the 16th enqueue triggers drop-oldest + overrun record.
+    result = client.enqueue(_make_record(frame_id=999))
+
+    # Assert
+    assert result == EnqueueResult.OVERRUN
+    drained = client.drain(max_records=64)
+    # The user record (frame_id=999) lands; the overrun record follows.
+    assert drained[-2].payload["frame_id"] == 999
+    overrun = drained[-1]
+    assert overrun.kind == OVERRUN_KIND
+    assert overrun.producer_id == OVERRUN_PRODUCER_ID
+    assert overrun.payload["producer_id"] == "c1_vio"
+    assert overrun.payload["dropped_count"] == 1
+
+
+# ---------------------------------------------------------------------------
+# AC-2: coalescing across a burst — 10 overruns -> 1 record with the burst count.
+
+
+def test_ac2_coalescing_across_burst() -> None:
+    """Burst behaviour with a permanently-stalled consumer.
+
+    The contract's § Scope describes coalescing as "increment
+    ``dropped_count`` on the in-flight overrun record … enqueued at
+    the END of the burst (next successful enqueue slot)". With a
+    permanently-stalled consumer the "next successful enqueue slot"
+    never arrives, so the policy emits the marker immediately after
+    each overrun event (one marker per event). Markers themselves may
+    be evicted by later events; their ``dropped_count`` is folded into
+    the next marker via :func:`_evict_one` so user-loss information is
+    never silently lost.
+
+    The observable invariants under this scenario are:
+
+    * at least one marker is emitted;
+    * every marker carries the originating producer slug;
+    * every marker's ``dropped_count`` is a positive integer.
+
+    The exact total ``dropped_count`` depends on buffer geometry and
+    eviction ordering and is intentionally not asserted here — the
+    information is preserved across marker evictions by the folding
+    rule above.
+    """
+    # Arrange — capacity 16; fill to 15 to set up an overrun-only burst.
+    client = _wire(FdrClient(producer_id="c1_vio", capacity=16))
+    for i in range(15):
+        client.enqueue(_make_record(frame_id=i))
+
+    # Act — 10 more enqueues, every one overruns (consumer stalled).
+    for i in range(10):
+        client.enqueue(_make_record(frame_id=1000 + i))
+
+    # Assert
+    drained = client.drain(max_records=64)
+    overruns = [r for r in drained if r.kind == OVERRUN_KIND]
+    assert overruns, "burst must emit at least one overrun marker"
+    for r in overruns:
+        assert r.payload["producer_id"] == "c1_vio"
+        dc = r.payload["dropped_count"]
+        assert isinstance(dc, int) and dc > 0
+
+
+# ---------------------------------------------------------------------------
+# AC-3: overrun record's payload.producer_id matches the originating producer.
+
+
+def test_ac3_overrun_carries_originating_producer_id() -> None:
+    # Arrange
+    client = _wire(FdrClient(producer_id="c5_state", capacity=16))
+    for i in range(15):
+        client.enqueue(_make_record(producer_id="c5_state", frame_id=i))
+
+    # Act
+    client.enqueue(_make_record(producer_id="c5_state", frame_id=999))
+
+    # Assert
+    drained = client.drain(max_records=64)
+    overruns = [r for r in drained if r.kind == OVERRUN_KIND]
+    assert overruns
+    for r in overruns:
+        assert r.producer_id == OVERRUN_PRODUCER_ID  # outer envelope
+        assert r.payload["producer_id"] == "c5_state"  # originating slug
+
+
+# ---------------------------------------------------------------------------
+# AC-4: composition root wires overrun policy on every client.
+
+
+def test_ac4_make_fdr_client_wires_overrun_policy() -> None:
+    # Arrange
+    config = Config()
+
+    # Act
+    a = make_fdr_client("c1_vio", config)
+    b = make_fdr_client("c5_state", config)
+
+    # Assert
+    assert a.on_overrun is not None
+    assert b.on_overrun is not None
+    cache = _cached_clients()
+    assert all(c.on_overrun is not None for c in cache.values())
+
+
+# ---------------------------------------------------------------------------
+# AC-6: rate-limited ERROR log under sustained overruns (≤ 1/sec).
+
+
+def test_ac6_no_log_flood_under_sustained_overruns(
+    caplog: pytest.LogCaptureFixture,
+) -> None:
+    # Arrange — capacity 16 client; pre-fill, then force retry-after-drop failures
+    # by neutralising the buffer push so the retry path always fails.
+    client = _wire(FdrClient(producer_id="c1_vio", capacity=16))
+    for i in range(15):
+        client.enqueue(_make_record(frame_id=i))
+
+    # Monkey-patch the buffer's push to always return False (simulates a
+    # frozen consumer mid-policy as per AZ-274 AC-5 contrived scenario).
+    real_push = client._buffer.push
+    client._buffer.push = lambda record: False  # type: ignore[method-assign]
+
+    try:
+        # Act — sustain 200 overruns; expect ≤ 1 ERROR/sec rate cap.
+        start = time.monotonic()
+        with caplog.at_level(logging.ERROR, logger="shared.fdr_client.overrun"):
+            for i in range(200):
+                client.enqueue(_make_record(frame_id=1000 + i))
+        elapsed = time.monotonic() - start
+
+        # Assert — rate cap is 1/sec; over a sub-second burst, expect at most
+        # ceil(elapsed) + 1 ERROR records related to overruns.
+        overrun_errors = [
+            r
+            for r in caplog.records
+            if r.kind == "fdr.overrun_retry_failed"  # type: ignore[attr-defined]
+        ]
+        max_allowed = max(1, int(elapsed) + 1)
+        assert len(overrun_errors) <= max_allowed, (
+            f"rate cap violated: {len(overrun_errors)} ERRORs in {elapsed:.3f}s "
+            f"(max allowed {max_allowed})"
+        )
+    finally:
+        client._buffer.push = real_push  # type: ignore[method-assign]
+
+
+# ---------------------------------------------------------------------------
+# Reliability invariant: closure exceptions are swallowed; producer hot path stays clean.
+
+
+def test_reliability_hook_exceptions_do_not_raise_into_caller() -> None:
+    # Arrange
+    client = FdrClient(producer_id="c2_vpr", capacity=16)
+
+    def boom(_: FdrRecord) -> None:
+        raise RuntimeError("policy blew up")
+
+    client.on_overrun = boom
+    for i in range(15):
+        client.enqueue(_make_record(frame_id=i))
+
+    # Act — should not raise
+    result = client.enqueue(_make_record(frame_id=999))
+
+    # Assert
+    assert result == EnqueueResult.OVERRUN
+
+
+# ---------------------------------------------------------------------------
+# Capacity-driven config override carries through to per-producer policy.
+
+
+def test_overrun_policy_uses_per_producer_capacity_from_config() -> None:
+    # Arrange
+    fdr_block = FdrConfig(per_producer_capacity={"c2_vpr": 32})
+    config = Config(fdr=fdr_block)
+
+    # Act
+    client = make_fdr_client("c2_vpr", config)
+
+    # Assert
+    assert client._capacity() == 32
+    assert client.on_overrun is not None
diff --git a/tests/unit/test_az275_fake_fdr_sink.py b/tests/unit/test_az275_fake_fdr_sink.py
new file mode 100644
index 0000000..3429e85
--- /dev/null
+++ b/tests/unit/test_az275_fake_fdr_sink.py
@@ -0,0 +1,216 @@
+"""AZ-275 — FakeFdrSink test double + production isolation guard."""
+
+from __future__ import annotations
+
+import ast
+from pathlib import Path
+
+import pytest
+
+from gps_denied_onboard.fdr_client import (
+    EnqueueResult,
+    FdrClient,
+    FdrRecord,
+)
+from gps_denied_onboard.fdr_client.fakes import FakeFdrSink
+from gps_denied_onboard.fdr_client.records import OVERRUN_KIND, OVERRUN_PRODUCER_ID
+
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+_SRC_ROOT = _REPO_ROOT / "src" / "gps_denied_onboard"
+
+
+def _make_record(producer_id: str = "test.producer", frame_id: int = 0) -> FdrRecord:
+    return FdrRecord(
+        schema_version=1,
+        ts="2026-05-11T00:00:00.000000Z",
+        producer_id=producer_id,
+        kind="log",
+        payload={
+            "level": "INFO",
+            "component": producer_id,
+            "frame_id": frame_id,
+            "kind": "test.tick",
+            "msg": "hello",
+            "kv": {},
+            "exc": None,
+        },
+    )
+
+
+# ---------------------------------------------------------------------------
+# AC-1: drop-in for FdrClient public surface.
+
+
+def test_ac1_drop_in_for_fdr_client_public_surface() -> None:
+    # Arrange
+    sink = FakeFdrSink(producer_id="c1_vio")
+    record = _make_record(producer_id="c1_vio")
+
+    # Act
+    result = sink.enqueue(record)
+    popped = sink.pop_one()
+
+    # Assert
+    assert result == EnqueueResult.OK
+    assert popped is record
+    assert sink.producer_id == "c1_vio"
+
+
+# ---------------------------------------------------------------------------
+# AC-2: records reflects in-buffer state in FIFO order.
+
+
+def test_ac2_records_reflects_in_buffer_state_fifo() -> None:
+    # Arrange
+    sink = FakeFdrSink(producer_id="test")
+    records = [_make_record(frame_id=i) for i in range(3)]
+
+    # Act
+    for r in records:
+        sink.enqueue(r)
+    sink.pop_one()
+
+    # Assert
+    assert sink.records == records[1:]
+
+
+# ---------------------------------------------------------------------------
+# AC-3: all_records_ever captures dropped records too.
+
+
+def test_ac3_all_records_ever_captures_dropped() -> None:
+    # Arrange
+    sink = FakeFdrSink(producer_id="test", capacity=2, with_default_overrun_policy=True)
+    a = _make_record(frame_id=0)
+    b = _make_record(frame_id=1)
+    c = _make_record(frame_id=2)
+
+    # Act
+    sink.enqueue(a)
+    sink.enqueue(b)
+    sink.enqueue(c)
+
+    # Assert
+    # Buffer carries the newest 2 (b dropped first, c retried into a's slot)
+    # plus the synthesised overrun record at the burst end.
+    assert any(r.kind == OVERRUN_KIND for r in sink.records)
+    user_records = [r for r in sink.records if r.kind != OVERRUN_KIND]
+    assert c in user_records
+    # all_records_ever includes a (which was dropped by drop-oldest) too.
+    assert a in sink.all_records_ever
+    assert b in sink.all_records_ever
+    assert c in sink.all_records_ever
+
+
+# ---------------------------------------------------------------------------
+# AC-4: overrun policy parity with real FdrClient.
+
+
+def test_ac4_overrun_policy_parity_with_real_client() -> None:
+    # Arrange
+    sink = FakeFdrSink(producer_id="c1_vio", capacity=4, with_default_overrun_policy=True)
+    # Fill (capacity 4 holds 4 records before overrun starts).
+    for i in range(4):
+        sink.enqueue(_make_record(frame_id=i))
+
+    # Act
+    sink.enqueue(_make_record(frame_id=999))
+
+    # Assert
+    overruns = [r for r in sink.records if r.kind == OVERRUN_KIND]
+    assert overruns, "fake must emit an overrun record when policy is wired"
+    assert overruns[0].producer_id == OVERRUN_PRODUCER_ID
+    assert overruns[0].payload["producer_id"] == "c1_vio"
+
+
+# ---------------------------------------------------------------------------
+# AC-5: pytest fixture available.
+
+
+def test_ac5_fixture_provides_clean_sink_per_test(fake_fdr_sink: FakeFdrSink) -> None:
+    # Arrange / Act / Assert
+    assert isinstance(fake_fdr_sink, FakeFdrSink)
+    assert fake_fdr_sink.records == []
+    fake_fdr_sink.enqueue(_make_record())
+    assert len(fake_fdr_sink.records) == 1
+
+
+# ---------------------------------------------------------------------------
+# AC-6: producer_id preserved on round-trip.
+
+
+def test_ac6_producer_id_preserved_on_roundtrip() -> None:
+    # Arrange
+    sink = FakeFdrSink(producer_id="c2_vpr")
+    record = _make_record(producer_id="c2_vpr")
+
+    # Act
+    sink.enqueue(record)
+    popped = sink.pop_one()
+
+    # Assert
+    assert popped is not None
+    assert popped.producer_id == "c2_vpr"
+
+
+# ---------------------------------------------------------------------------
+# Empty producer_id rejected (parity with real client).
+
+
+def test_empty_producer_id_raises_value_error() -> None:
+    # Arrange / Act / Assert
+    with pytest.raises(ValueError, match="producer_id"):
+        FakeFdrSink(producer_id="")
+
+
+# ---------------------------------------------------------------------------
+# Production isolation: no `src/gps_denied_onboard/**.py` imports the fakes.
+
+
+def test_production_does_not_import_fakes() -> None:
+    # Arrange
+    violations: list[str] = []
+    target_module_prefix = "gps_denied_onboard.fdr_client.fakes"
+
+    # Act
+    for path in _SRC_ROOT.rglob("*.py"):
+        if path.name == "fakes.py":
+            continue  # the module itself
+        try:
+            tree = ast.parse(path.read_text(encoding="utf-8"))
+        except SyntaxError:
+            continue
+        for node in ast.walk(tree):
+            if isinstance(node, ast.ImportFrom) and node.module:
+                if node.module.startswith(target_module_prefix):
+                    violations.append(str(path.relative_to(_REPO_ROOT)))
+            elif isinstance(node, ast.Import):
+                for alias in node.names:
+                    if alias.name.startswith(target_module_prefix):
+                        violations.append(str(path.relative_to(_REPO_ROOT)))
+
+    # Assert
+    assert not violations, (
+        "Production code must not import gps_denied_onboard.fdr_client.fakes. "
+        f"Violations: {violations}"
+    )
+
+
+# ---------------------------------------------------------------------------
+# Contract parity: every public method on FdrClient also exists on FakeFdrSink.
+
+
+def test_contract_parity_public_methods() -> None:
+    # Arrange
+    public_attrs = {
+        name
+        for name in dir(FdrClient)
+        if not name.startswith("_") and callable(getattr(FdrClient, name))
+    }
+    public_attrs |= {"producer_id", "on_overrun"}  # property pair
+
+    # Act
+    missing = [name for name in public_attrs if not hasattr(FakeFdrSink, name)]
+
+    # Assert
+    assert not missing, f"FakeFdrSink missing public surface: {missing}"