diff --git a/docs/plans/mcp-observability/execution-ledger.md b/docs/plans/mcp-observability/execution-ledger.md
new file mode 100644
index 00000000..dcaace74
--- /dev/null
+++ b/docs/plans/mcp-observability/execution-ledger.md
@@ -0,0 +1,70 @@
+# Execution ledger
+
+2026-10-14: Read issue #3 and all comments (none) via authorized GitHub access.
+Cloned Coding-Standards separately; baseline 188beda1fa477d21d576c233dd8c7f4c4c267d23.
+Core/Router, selected owners, MCP/catalog/runtime identity, contract discovery,
+input feedback and observer-v3 qualification/reconciliation inspected.
+Python 3.12.14 is inside the declared >=3.11,<3.13 range.
+User subsequently explicitly authorized push + PR after testing and review;
+merge and deployment remain excluded. No other repository modified.
+
+2026-10-14: Implemented one recorder boundary and operator report;22 focused
+synthetic tests passed on Python 3.12.14 in 27.628 s. Independent GPT-6.1 Sol High
+review accepted exact source hashes after resolving seven findings, including
+rejection of prior private SQLite remnants. Observer-v3 suites passed 39 tests
+in 19.044 s. Contract projection check passed. Verifier initially exposed stale
+README/source input digests; refreshed through public verify_repository debug
+transport, then 73 suites/121 checks passed. Generated suite-inputs manifest added
+to allowed adjacent write set. Full Engine run remains active; baseline fixture
+failure I1 reproduced on a clean independent baseline clone, unchanged by this
+feature. Four-mode cold baseline/candidate comparison preserved counts, outputs
+and exact serialized byte totals. No raw traces or databases are deliverables.
+
+2026-10-14: Final focused recorder+MCP+runtime-identity run passed 40 tests in 42.362 s
+(23 recorder,13 MCP,4 runtime identity). Final constants use one owner with the
+same defaults; reviewer accepted and rebound exact source hashes. Repeated
+four-configuration baseline/candidate response comparison passed on final source.
+The broad Engine run completed 93 tests (one known baseline failure), then was
+interrupted during the next owner module to parallelize remaining independent
+modules in four isolated Python processes. The interrupted case is re-run in full
+with its module; only completed prior modules count toward aggregate coverage.
+No interrupted case is counted as passed. All test stores are private temporary
+resources. Generated inputs refreshed after staging new module membership;
+verify_repository returned 73 suites/121 checks passed. Source, docs, synthetic
+tests, staged scope and sensitive-file risk reviewed before the implementation
+commit; no database, raw interaction log, secret or personal configuration staged.
+
+2026-10-14: Accepted local implementation boundary at e2334869. Broad aggregate
+coverage completed 636 cases (93 completed serial prefix +543 remaining),635
+qualified passes and the unchanged baseline I1 failure. Initial efficiency child
+launch without CI's PYTHONPATH=. failed equally on baseline/candidate; both passed
+with that declared environment. No additional feature failure remained. Final
+contract projection and73-suite/121-check verifier passed. Exact independent
+review bindings still match committed sources. CLI GitHub authentication reports
+invalid GH_TOKEN (I2); no push/PR or fallback credential/connection attempted.
+Local scope is complete, publication externally blocked. Final evidence commit
+retains clean source/branch and sanitized reports for parent; bundle outside repo
+preserves exact committed ref. Parent owns subsequent publication/merge.
+
+2026-10-14: Parent approved bounded I1 repair for PR #4 after prior service failures.
+Verified real checkout read/write/fsync, clean branch/head and remote PR head
+4887b7af; no workspace recovery or duplicate push needed. Revisited I1 disposition
+under explicit follow-up authority. Fixture now resolves POLICY from its immutable
+compiled snapshot, asserts existence and selects semantic_revision and its single
+successor before the existing counting mocks. Production validator, normative
+content and both count assertions are unchanged. Failing case, analysis module,
+full Engine suite and independent review are required before the next commit/push.
+
+2026-10-15: Fixture repair qualified. Owner targeted case passed 1 test in 3.748 s;
+independent targeted case passed in 3.676 s. Refreshed the registered test input
+through public Engine verify_repository; staged the new review report membership
+and refreshed again before freezing inputs. The initial analysis/full attempts
+with stale generated input were interrupted and are not acceptance evidence.
+Final analysis module passed 23 tests in 113.275 s. One uninterrupted full serial
+Engine invocation passed all 636 tests in 2161.675 s under PYTHONPATH=.; the formerly
+failing case and all observer-v3 tests passed in that run. Independent review bound
+test_analysis.py SHA256 7f51e403f03f5c57be8b1ca5736a79bcf139c87dd9eb4703ce3f7e59f1ba3d01.
+No production, validator or normative edits; both call-count assertions unchanged.
+Final evidence retains the historical I1 failure and resolves prior I2 transport
+confusion. Reviewed bounded staged scope before committing for approved PR #4
+branch push; publication identity is the Git ref, merge remains parent-owned.
diff --git a/docs/plans/mcp-observability/issues.md b/docs/plans/mcp-observability/issues.md
new file mode 100644
index 00000000..0d4b619d
--- /dev/null
+++ b/docs/plans/mcp-observability/issues.md
@@ -0,0 +1,39 @@
+# Findings
+
+## I1 — baseline proposal-analysis fixture (medium, resolved)
+
+`test_analysis.AnalysisWorkflowTest.test_proposal_analysis_uses_one_projection_and_one_verified_base`
+originally hardcoded accepted semantic revision 1/proposed 2. Both the candidate and a clean
+Git clone of baseline `188beda1fa477d21d576c233dd8c7f4c4c267d23` return
+`AUTHORING.INVALID_SEMANTIC_REVISION` before the test reaches its projection-count
+assertions. Independent one-test baseline run:1 test, 1 failure, 1.397 s; candidate:
+1 test, 1 failure, 1.606 s. At initial handoff head 4887b7af this source test was unchanged by issue3.
+
+Owner: analysis fixture maintainers. Parent explicitly authorized this bounded
+follow-up for PR #4. Disposition: resolved by deriving the accepted semantic
+revision and its successor from the same immutable snapshot used for the proposal,
+before the counting mocks. Policy existence, real proposal/analysis results and
+both count assertions remain required. Validator and normative content unchanged.
+The targeted case passed (3.748 s), all 23 analysis tests passed (113.275 s), and
+the full serial Engine suite passed all 636 tests (2161.675 s). Independent fixture
+review passed; see reports/fixture-review.md. Historical baseline failure evidence
+above remains retained rather than relabeled as a baseline pass.
+
+## Resolved review findings
+
+Independent reviewer found and verified resolution of R1–R7: diagnostic field
+binding, malformed-envelope coverage, measurement failure, normal retention loss,
+report scope/bounds, pre-migration capability checks and residual private SQLite
+content. See reports/independent-review.md for source-bound evidence.
+
+Actual model/client traces are outside this task: synthetic process evidence does
+not certify model intent or acquisition. No new model qualification is claimed.
+
+## I2 — publication authentication (resolved)
+
+The GitHub CLI reported an invalid configured `GH_TOKEN`, initially treated as a
+publication blocker. A later explicitly authorized attempt through the existing
+Git remote succeeded without inspecting/copying tokens or changing authentication.
+Remote head 4887b7af was verified and the parent opened draft PR #4. The CLI failure
+did not establish failure of the separate Git transport. Existing bundle and
+checkout were preserved; subsequent publication uses the approved Git branch.
diff --git a/docs/plans/mcp-observability/plan.md b/docs/plans/mcp-observability/plan.md
new file mode 100644
index 00000000..c59ac413
--- /dev/null
+++ b/docs/plans/mcp-observability/plan.md
@@ -0,0 +1,144 @@
+# Persistent MCP interaction observability (issue 3)
+
+Status: Accepted (local fixture qualification). Acceptance: satisfied. Operation: verify. Owner: implementation agent;
+publication and merge authority remain with the parent/operator. Baseline:
+`188beda1fa477d21d576c233dd8c7f4c4c267d23` (interface 46).
+Canonical plan: `docs/plans/mcp-observability/plan.md`.
+
+## Objective and authority
+
+Implement opt-in, local, agent-invisible SQLite observations at the existing MCP
+boundary. Keep domain transactions, tools, outcomes, protocol, retries, generated
+contracts, normative standards and observer-v3 qualification unchanged. No model
+calls, Tuldok changes, installation configuration or live telemetry collection.
+The subsequent user approval permits pushing the verified feature branch and
+opening a PR containing only code, synthetic tests and sanitized documentation.
+Parent owns merge. No deployment or enabling on existing installations.
+
+Private feature branch `feat/issue3-mcp-observability` in the separate clone
+`/workspace/Coding-Standards`; target `main`, integration owner parent. Retain
+branch/checkout for publication decision and review; no cleanup or history rewrite.
+No writes to Nearwork or other mounted repositories.
+
+## Routed owners and decisions
+
+Core and Router select implementation, verification/oracles, planning/discovery,
+proportionality, commit, documentation; library application, IPC and persistence; diagnostics,
+architecture, code-design, contracts/evolution/protocols, concurrency, security,
+resilience, performance and cross-platform. Python has no language profile.
+No normative or generated contract changes: no generated-contract profile,
+standards-authoring workflow, release or framework profile is needed.
+No AGENTS.md exists in this clone or its workspace ancestors. Runtime identity
+uses process UUID and exact catalog/schema/implementation digests, independent
+of Git/content authority. Observer v3 remains qualification evidence and retains
+its independent exact-source replay. No production dependency on test code.
+
+## Binding design
+
+- MCP owns request execution and response delivery. Recorder owns only sanitized
+  observations, SQLite transactions, retention and health. Operator CLI reads
+  that separate database; Engine stores and authorization remain untouched.
+- Explicit operator CLI configuration supplies a database in an existing trusted
+  directory and a deployment label. Omission creates no recorder/file. POSIX local
+  regular files only; reject symlinks, nonprivate files, unrelated databases and
+  unsupported schemas. The trusted directory must exclude concurrent hostile
+  path replacement. Same-user agents can read files: filesystem modes are damage
+  reduction, never an agent authorization boundary. Store no secrets regardless.
+- Schema v1: immutable interaction starts plus independent terminal events, joined
+  by generated UUID; process/session UUID and ordinal establish local chronology.
+  Sessions bind runtime metadata, deployment label and per-operation schema
+  digests. No client-supplied request IDs, names or paths are stored verbatim.
+- Pair end/error with begin only after response write/flush in stdio. Direct
+  dispatch records a constructed response (distinct delivery evidence). Parsing,
+  validation, typed rejection, tool exceptions and interruption are classified
+  without raw messages. Unmatched starts remain incomplete/unknown, including
+  process kill and response delivery failure; never infer completion on restart.
+- Bounded input shape: fixed schema-derived property names, scalar type markers,
+  depth/node/array caps; redact sensitive subtrees and unknown member names.
+  Never retain scalar values, arbitrary map keys, prompts, bodies, tokens,
+  credentials, outputs, exception messages or client metadata. Static schema
+  selector/operation names are admitted only by exact installed-catalog membership.
+  Feedback retains sanitized field pointers and fixed constraint keywords only.
+- No raw input digests (dictionary attacks). Shape digests identify structural
+  variants only. Catalog and implementation digests identify exact source material.
+  MCP has no legitimate cross-process session/task continuation in this server;
+  do not invent one from prompts or domain handles. Reports state that limitation.
+- Serial synchronous recorder, no queue/thread/retry. SQLite `BEGIN IMMEDIATE`
+  owns each mutation and rollback. Finite busy timeout and page quota; rollback
+  journal rather than unbounded WAL. Maximum retained interactions and age are
+  operator policies with bounded defaults; prune starts and events atomically.
+  Retention may evict unmatched observations and is explicitly reported.
+- Version 0 is accepted only for a zero-length new database and migrates transactionally
+  to v1. Existing v1 reopens intact; future/corrupt/unrelated databases fail open
+  without replacement. No inferred migration or automatic destructive recovery.
+- Recorder failure degrades once, disables further recording for that process,
+  and reports a fixed nonsensitive reason plus counters on operator stderr.
+  It never retries domain work or changes its return. Operator fixes storage and
+  restarts to recover recording. Health is independent of the failing database.
+- Read-only bounded JSON report: operation/error/shape counts, observed latency
+  and bytes, incomplete counts, discovery/failed-attempt sequences before a
+  later success in the same process/operation, catalog comparisons. Denominators
+  and delivery scope explicit. Correlation cannot prove intent, actual repair,
+  client acquisition, token usage or billing. Retained traces cannot reconstruct
+  redacted requests; observer v3 remains the source-backed replay authority.
+
+## Composed-design admission: applicable
+
+1. Independent concerns: MCP owns execution/delivery; recorder owns privacy,
+   observations and local retention; operator report owns read-only interpretation;
+   RuntimeIdentity/catalog own source/contract attribution. No observer-v3 coupling.
+2. Necessary interleaving: begin before dispatch, terminal after delivery, metadata
+   captured at recorder construction. Accidental coupling avoided: no Engine SQL,
+   tool registration, hidden model work or telemetry-driven domain decisions.
+3. Composition root knows configuration, immutable runtime/catalog and close;
+   dispatch/stdio know begin/finish lifecycle. Callers never own SQL or redaction.
+4. Privacy changes touch recorder/tests/docs; schema revisions flow through existing
+   catalog identity; delivery changes touch MCP integration; storage revisions
+   touch recorder migration and CLI together. Domain changes need no recorder edits.
+5. Stable runtime metadata crosses the seam; no store handles or validator internals.
+   Schema property vocabulary is diagnostic disclosure authority, not validation.
+6. Recorder can fail and be disabled independently; CLI reads across restart.
+   MCP remains independently usable. Tests exercise both enabled and omitted paths.
+7. Deleting recorder moves necessary privacy/storage/retention/lifecycle complexity
+   back into MCP, so retain it. No generic adapter registry, background worker,
+   second protocol, exported observer tool or migration framework is admitted.
+8. Inherent complexity is bounded evidence, privacy and SQLite lifecycle; keep it
+   in one production module plus its read-only reporting entrypoint, not harnesses.
+
+## Milestones and evidence
+
+A Accepted: design/source/privacy review; write set this plan directory only.
+B Accepted: recorder and MCP integration; write set
+`tools/standards_engine/standards_engine/{interaction_observer.py,mcp.py}`,
+`tools/standards_engine/tests/test_interaction_observer.py`. Gate: focused
+privacy, failure and outcome-preservation tests.
+C Accepted: operator report, configuration and operational documentation; same
+module plus `tools/standards_engine/MCP-OBSERVABILITY.md` and README link.
+Gate: bounded read-only report, catalog changes, accurate denominators.
+D Accepted: cold stdio/process replacement, exceptions/interruption, real unavailable,
+locked/full SQLite, concurrent processes, migration/retention/privacy. Existing MCP,
+runtime identity, discovery and observer-v3 tests plus full Engine suite and
+contract projection check. Synthetic baseline/candidate call and byte comparison.
+E Accepted: independent GPT-6.1 Sol High design/source/privacy review accepted with all seven findings resolved; staged and full branch review, local commits, authorized push and PR.
+No real logs or confidential artifacts in Git. Parent owns merge. The generated
+`evaluation/standards-effectiveness/generated/suite-inputs.json` is an allowed
+adjacent write: refresh through the Engine after staging declared inputs; it
+tracks source/README digests and is not a normative standards change.
+
+F Accepted: repair I1 using the actual policy semantic revision from the immutable
+snapshot, before entering the existing counting mocks. Write set:
+`tools/standards_engine/tests/test_analysis.py`, this plan/ledger/issues, a bounded
+fixture-review report and current verification evidence; generated verifier input
+manifest only if its compiler requires a refresh. Gate: failing case, analysis
+module and full Engine suite pass; independent review; reviewed commit/push to
+existing approved PR #4 branch. Validator, normative content and both count
+assertions remain unchanged. No telemetry activation or raw log publication.
+
+Current phase: complete local qualification. Next slice: none. Blockers: none.
+F evidence: targeted case passed; analysis module 23/23; full serial Engine suite
+636/636; independent fixture review satisfied. I1 is resolved without changing
+validator, normative content or either count assertion; historical baseline failure
+remains in issues/evidence. I2 resolved through the existing Git transport; parent
+opened PR #4. The verified follow-up is packaged for the explicitly approved branch
+push; Git/PR refs own publication identity and parent owns hosted CI/merge.
+See [verification](reports/verification.md), [fixture review](reports/fixture-review.md), [independent source/privacy review](reports/independent-review.md), [ledger](execution-ledger.md) and [issues](issues.md).
diff --git a/docs/plans/mcp-observability/reports/baseline-comparison.md b/docs/plans/mcp-observability/reports/baseline-comparison.md
new file mode 100644
index 00000000..04cb7eb9
--- /dev/null
+++ b/docs/plans/mcp-observability/reports/baseline-comparison.md
@@ -0,0 +1,18 @@
+# Synthetic baseline/candidate comparison
+
+Baseline: `188beda1fa477d21d576c233dd8c7f4c4c267d23`, extracted with `git archive` into an isolated temporary directory. Candidate: reviewed source hashes in [independent review](independent-review.md). Python 3.12.14; SQLite 3.53.1; locked dependency versions matched. No model/client session or real interaction log was collected.
+
+Each cold process received initialize, initialized notification, tools/list, describe_input(runtime_info), an invalid runtime_info request, a valid runtime_info request and ping. Six requests, one notification, three tool calls, six replies ; 688 sent stdio bytes. The operator recorder created only temporary synthetic databases, removed after the comparison.
+
+| Purpose | Output schemas | Baseline bytes | Candidate off | Candidate on | Replies equal |
+| --- | --- | ---: | ---: | ---: | --- |
+| authoring | eager | 753498 | 753498 | 753498 | yes |
+| authoring | on-demand | 115030 | 115030 | 115030 | yes |
+| application | eager | 89041 | 89041 | 89041 | yes |
+| application | on-demand | 22115 | 22115 | 22115 | yes |
+
+Comparison parsed every reply, including schema catalog, typed rejection, text content and runtime observations. Only process UUID and implementation digest were normalized: these must differ for distinct loaded sources/processes. Full JSON response byte totals matched without normalization. No tools, protocol messages, operation retries or domain results were added.
+
+One sample per configuration is outcome/byte preservation evidence, not a latency benchmark. Observed cold process times ranged from 852.56 ms to 1193.81 ms with other tests active. No throughput/overhead, token or billing claim is made.
+
+Reproduction: extract the named baseline with git archive; start the baseline MCP and candidate MCP with --purpose authoring/application and --output-schemas eager/on-demand. Send the seven listed JSON-RPC messages, compare parsed replies after replacing instance_id/implementation_digest (including JSON text blocks), and count UTF-8 bytes of the actual stdio streams. Repeat the candidate with --observe-database pointing at a fresh private temporary path and --observe-deployment synthetic-comparison.
diff --git a/docs/plans/mcp-observability/reports/fixture-review.md b/docs/plans/mcp-observability/reports/fixture-review.md
new file mode 100644
index 00000000..6493db13
--- /dev/null
+++ b/docs/plans/mcp-observability/reports/fixture-review.md
@@ -0,0 +1,68 @@
+# Independent fixture repair review
+
+Date: 2026-10-14. Parent baseline:
+`4887b7aff0ba84acd257da081135b7849cc9c79f`.
+Disposition: **satisfied for this bounded fixture change**. No source, privacy or
+oracle-weakening findings. Overall acceptance remains with the owner's final
+frozen-source analysis/full Engine and generated-input verification gates.
+
+## Exact scope
+
+The only test behavior change is in
+`tools/standards_engine/tests/test_analysis.py:2191`, method
+`test_proposal_analysis_uses_one_projection_and_one_verified_base`:
+
+- Resolve `POLICY` through the compiled corpus of the same `self.snapshot` used
+  as the proposal's `base_snapshot` and source of its body.
+- Assert the policy exists, then use its semantic revision as the accepted
+  revision and that revision plus one as the proposed revision.
+- Perform this setup before installing the two counting mocks.
+
+The diff is seven added lines and one removed line. The actual analysis call,
+`PendingResult` assertion and both exact `call_count == 1` assertions are
+unchanged. No production validator, normative content, telemetry code or public
+contract changes. The coordinated generated `suite-inputs.json` refresh is an
+allowed fixture-repair artifact: it records the changed registered test input and
+review-file membership; it does not alter the validator or weaken its decision.
+
+## Authority and oracle assessment
+
+`StandardsEngine._compiled_snapshot` at `engine.py:2122` loads captured content for
+the explicit snapshot ID and compiles a `FrozenContentSource`; it does not select
+the current live policy implicitly. Snapshot `load_content` at
+`standards_snapshots/module.py:123` verifies captured content identity after its
+durable read. The fixture's `open_repository` call supplies no compiled cache, so
+the extra setup lookup does not introduce a warm-cache substitute for the observed
+analysis path.
+
+The repair follows the existing nearby convention in
+`test_analysis.py:335–351`: obtain the policy's accepted semantic revision from the
+captured corpus and advance it explicitly. This is correct fixture authority,
+rather than a new expected-result oracle derived from the subject under test.
+The test's independent expected behavior remains one verified-base load and one
+proposal projection during the analysis operation. Real proposal construction
+and revision validation still run and `CreateProposalResult` remains required.
+
+## Independent evidence and binding
+
+Command, from `/workspace/Coding-Standards` on CPython 3.12.14:
+
+```sh
+PYTHONPATH=. python -m unittest tools.standards_engine.tests.test_analysis.AnalysisWorkflowTest.test_proposal_analysis_uses_one_projection_and_one_verified_base
+```
+
+Result: **one test passed in 3.676 seconds**. `git diff --check` passed.
+
+Reviewed test-file SHA256:
+`7f51e403f03f5c57be8b1ca5736a79bcf139c87dd9eb4703ce3f7e59f1ba3d01`.
+
+The prior accepted observer, MCP, observer-test and operating-guide hashes were
+rechecked unchanged. The reviewer wrote only this report and performed no
+publication or installation action. No model calls or private logs were used.
+
+The owner's first broader runs were interrupted because the registered test input
+had changed before its generated verifier digest was refreshed; their partial
+results are not acceptance evidence. The owner is refreshing through the public
+Engine verification path and will rerun the analysis module and full serial
+Engine suite from frozen source. Those final results and generated-manifest
+freshness are outside this independently executed targeted check.
diff --git a/docs/plans/mcp-observability/reports/independent-review.md b/docs/plans/mcp-observability/reports/independent-review.md
new file mode 100644
index 00000000..222bb17d
--- /dev/null
+++ b/docs/plans/mcp-observability/reports/independent-review.md
@@ -0,0 +1,296 @@
+# Independent observability review
+
+Reviewer role: independent design/source/privacy reviewer requested by the user.
+Baseline: `188beda1fa477d21d576c233dd8c7f4c4c267d23`, interface 46.
+Review date: 2026-10-14. Scope: the observability plan and existing production
+MCP/catalog/runtime identity plus observer-v3 context. No paid model calls,
+publication, installation changes, or implementation edits were performed.
+
+Current disposition: independent source/privacy review **satisfied** for the exact
+files bound below. R1–R7 are resolved. Earlier pending dispositions describe the
+review stages and are superseded by the final disposition. Overall branch
+acceptance still requires the owner's full Engine/observer-v3/projection and
+publication gates; this review does not assert those unexecuted results.
+
+## Design disposition
+
+Proceed to implementation with the acceptance points below. The proposed ownership
+and disclosure boundaries are appropriate. These are design requirements to verify,
+not findings against implementation that does not exist yet. Implementation/source
+acceptance remains pending.
+
+The plan preserves the Engine's domain authority, opts in outside agent tools,
+distinguishes construction from delivery, disables a failed recorder, rejects
+unsupported retained state, and honestly excludes replay, intent, acquisition,
+token and billing conclusions. No design change to observer-v3 is needed.
+
+## Findings and required evidence
+
+### D1 — Medium: include health emission in the fail-open boundary
+
+Evidence: plan lines 74–77 require one stderr health report and no domain-return
+change; existing stdio delivery is owned at `mcp.py:210–227`. A stderr write or
+flush can itself raise. A failure handler that emits through an unguarded sink can
+replace an otherwise valid response or mask the original interruption.
+
+Disposition: guard health emission and recorder cleanup independently. Keep fixed
+reason codes and in-memory counters; never include exception strings, database
+paths or deployment input in failure text. Test a broken stderr sink as well as a
+failed database, proving identical MCP output and exactly one domain invocation.
+The synchronous implementation cannot promise an end-to-end latency bound for
+arbitrarily blocked filesystem or stderr I/O; state that operational limit rather
+than deriving it from SQLite busy timeout.
+
+### D2 — High: structural privacy needs component-level authority
+
+Evidence: plan lines 56–61 permit schema property names and sanitized pointers,
+while `input_feedback.py:14–19` projects validator issues and
+`validation_feedback.py:25–45` builds instance paths from input locations. It
+already replaces names absent from the schema-path property set with `*`, but
+that set is not position-aware: an arbitrary map key equal to an ancestor's
+declared name can survive. Existing value-free messages therefore do not prove
+every pointer component's exact-location authority. Static property names also
+do not authorize collecting their scalar values or enum choices.
+
+Disposition: admit only property components at their exact schema location;
+unknown keys and `additionalProperties` keys need fixed markers, including inside
+feedback pointers. Redact whole free-form/sensitive subtrees before traversal.
+Never retain raw outputs, message text, raw hashes, selector guesses or client
+identifiers. Admit selector strings only from the selected operation's exact
+canonical definition inventory. Test nested arbitrary keys, escaped pointer
+components, very long/deep objects and arrays, scalar substitutions, and secret
+sentinels across the database, journal, report and health output. Treat absence
+of an independent privacy oracle as pending evidence.
+
+### D3 — Medium: schema identity and migration checks must share writer isolation
+
+Evidence: plan lines 66–73 select multiple processes, `BEGIN IMMEDIATE`, v0 empty
+initialization, intact v1 reopening and rejection of unrelated stores. Reading v0
+or checking table absence before acquiring the transaction permits two initializers
+to decide from stale facts. A `user_version` match alone does not establish the
+application's owned schema.
+
+Disposition: acquire the writer transaction before validating the state used for
+initialization, verify the expected v1 schema/application identity, and preserve
+unrelated/partial/future/corrupt databases. Test simultaneous first-open, v1 reopen,
+spoofed/partial v1, lock exhaustion, quota failure and transactional rollback.
+Do not change foreign database PRAGMAs or create observation tables before proving
+ownership. A version pragma is a ledger, not an ownership proof.
+
+### D4 — Medium: retention must own session growth and in-flight eviction
+
+Evidence: plan lines 47–50 retain session metadata separately and lines 69–70
+bound interactions while explicitly permitting unmatched-start eviction. A limit
+on interactions alone can leave sessions growing through repeated restarts.
+Another process can prune a start while its original owner is still executing.
+
+Disposition: bound and prune orphan session metadata alongside retained records;
+define how finish behaves when its start was already evicted. Such eviction must
+not create a fabricated completion or unnecessarily disable otherwise healthy
+recording. Exercise a live request crossing retention in a second process and
+repeated short-lived sessions. Account for evictions in the persisted/report
+denominators without retaining an unbounded history of evicted identifiers.
+
+### D5 — Medium: bound report work and preserve distinct denominators
+
+Evidence: plan lines 78–83 promise a bounded read-only report with latency, bytes,
+incomplete observations and ordered discovery/failure sequences. Limiting only
+the final serialized JSON does not bound query work or intermediate cardinality.
+The existing direct dispatch returns at `mcp.py:100–107`; stdio writes/flushes at
+`mcp.py:224–225` are a separate observation.
+
+Disposition: declare scan/row/group/output budgets and truncation indicators;
+use one consistent read snapshot. Keep constructed responses, flushed responses,
+notifications, incomplete/evicted starts and protocol/typed/tool errors distinct.
+Define which duration includes recording overhead, use monotonic elapsed time,
+and count JSON UTF-8 bytes under an explicit newline policy. Ordered sequences
+are same-process/operation correlation only, and must not cross unknown retention
+gaps as if a complete history existed. Test read-only behavior against an active
+writer and bounded output with many groups/sessions.
+
+## Source and qualification boundaries
+
+`runtime_identity.py:46–78` already owns process UUID, catalog/schema digests and
+installation-source material digests; its module contract requires a coherent
+installation before startup and does not claim hot reloading. Reuse those values
+without introducing a Git commit or standards snapshot as implementation authority.
+Per-operation digests must identify the complete canonical operation input schema,
+not merely its abbreviated agent-facing presentation. A digest comparison reports
+equality/difference; it cannot reconstruct an absent historical catalog.
+
+`INPUT-DISCOVERY-QUALIFICATION.md:84–132` requires exact source/runtime/catalog,
+chronological starts, isolated thread/turn and retained transcript/readbacks for
+observer-v3 replay. Production shapes deliberately discard the evidence needed for
+those claims. No production report may relabel its sequence counts as observer-v3
+qualification or corrected historical replay. Synthetic tests must remain labeled
+as process/integration evidence, with no model/client acceptance claim.
+
+## Design-stage remaining acceptance
+
+Exact-source review and tests are pending implementation. Required review focuses
+on actual disclosure paths and failure handlers, all persisted fields, transaction
+ordering, delivery/error/interruption paths, report bounds, and unchanged catalogs
+and domain outcomes. No implementation approval, publication approval or deployment
+qualification is implied by this design review.
+
+## First implementation review
+
+Source was uncommitted and changing during this review. The references below name
+the first inspected implementation; final source hashes and dispositions will be
+recorded after fixes. Independent synthetic probes used temporary SQLite databases
+only. They did not invoke a model or modify production code.
+
+| ID | Severity | Evidence and observed consequence | Required disposition |
+| --- | --- | --- | --- |
+| R1 | Medium | `InputShape.feedback` read `pointer`, but production `InputIssue` exposes `instance_pointer`. The real-dispatch privacy test failed because `expected_catalog` disappeared from retained feedback. The unit fixture supplied the nonexistent field. | Consume the actual production diagnostic field and retain a real-dispatch regression. |
+| R2 | Medium | `begin` skipped all dictionaries lacking `id`. A probe dispatched `{'jsonrpc':'1.0','method':'ping'}` and received `-32600`, but retained starts remained zero. | Skip actual notifications while retaining malformed-envelope responses, without changing dispatch behavior. |
+| R3 | Medium | `serve` computed `line.encode('utf-8')` outside the measurement guard even with observations omitted. A StringIO JSON request containing an unpaired surrogate raised `UnicodeEncodeError`; the baseline parser/serializer handles that input. | Make byte measurement optional and fail-open, preserving the omitted path. |
+| R4 | Medium | With one call active, another recorder inserted 16 calls at `max_calls=16`. Finishing the evicted call permanently degraded the first recorder with `start-evicted-by-retention`. | Treat expected retention loss as a dropped terminal; preserve future recorder usability. |
+| R5 | Medium | A two-scope probe with report `limit=1` emitted one group with no group truncation flag, but sequence denominator merged both scopes. Chronology scan and histogram cardinality were unaffected by limit. | Add explicit scan/group/output bounds and truncation; preserve construction/delivery scope in sequence denominators. |
+| R6 | Medium | An empty v0 WAL database was committed to v1/three tables before rejection as `unsupported-journal-mode`; a byte comparison confirmed mutation despite rejected storage capability. | Prove journal/page-budget capability before migration publication and verify rejected retained states remain intact. |
+
+All six findings were sent promptly to the implementation owner. None alleges a
+confirmed secret-value leak in the inspected recorder. R1 concerns missing useful
+feedback; the separate structural privacy projection remains subject to final
+source review and adversarial tests. Acceptance is pending fixes and final review.
+
+## Fix review and additional admission finding
+
+The revised source resolves R1–R6: real `instance_pointer` feedback is consumed;
+malformed no-ID requests are retained; byte encoding is optional/guarded; normal
+in-flight retention loss drops only its terminal; reports separate scope and expose
+bounded sections/scan/truncation/measurement denominators; storage capabilities
+are checked before migration publication. Sanitized shape summaries now include
+successful as well as failed requests. The operating guide accurately distinguishes
+delivery, domain completion, loss counters, idle retention and observer-v3 replay.
+
+Independent command on CPython 3.12.14:
+
+```sh
+python -m unittest tools.standards_engine.tests.test_interaction_observer
+```
+
+Result: 21 tests passed in 27.416 seconds. This is synthetic local/process evidence,
+not real model/client acceptance. Final acceptance remains pending R7 below.
+
+### R7 — High: logically empty v0 can contain unrelated private bytes
+
+The v0 admission predicate at the then-current `interaction_observer.py:293–295`
+checked application/version zero and absence of tables. A database from which a
+table was dropped satisfies that predicate while retaining the table's raw values
+in freelist pages when secure deletion was previously disabled. Enabling secure
+deletion later does not erase those existing remnants.
+
+Independent temporary-database reproduction: disable `secure_delete`, create an
+unrelated table, insert a repeated synthetic private-content sentinel, and drop
+the table. The source has no tables, two pages and one freelist page. The recorder
+accepted it as healthy, recorded a ping, and the sentinel remained in the resulting
+observation file after close. No actual private content was used.
+
+Required disposition: reject non-pristine v0 retained states before mutation.
+Zero-length new/existing files are a sufficient narrow initial support policy;
+supporting additional empty representations requires proof that they cannot carry
+unrelated retained bytes. Preserve rejected files rather than wiping them. Add an
+on-disk sentinel regression and update the documented v0 admission policy.
+
+## Final source disposition
+
+R7 is resolved by requiring both the original file-descriptor stat and current
+physical file length under the writer lock to be zero before initializing v0.
+The trusted-directory precondition excludes hostile path replacement. The guide
+and plan now select zero-length new v0 sources explicitly. Rejected prior files
+are preserved. A first attempted correction used transaction-observed page count
+zero and rejected new files because SQLite materializes a virtual first page at
+`BEGIN IMMEDIATE`; independent probing exposed this and the physical-length check
+replaced it before acceptance.
+
+An independent final temporary-database probe observed a fresh file becoming
+healthy and the dropped-private-content file being rejected as
+`nonpristine-database`, byte-for-byte unchanged with its sentinel still present.
+The final focused command above passed **22 tests in 27.573 seconds** on CPython
+3.12.14. It covers real dispatch/privacy, arbitrary outputs, parse/envelope errors,
+tool exception/interruption/delivery failure, broken stderr, unavailable/locked/full
+SQLite, supported/rejected/migration-interrupted states, retention across processes,
+scope/report truncation, and cold stdio/process replacement. `git diff --check`
+also passed. No real logs, private datasets or model calls were used.
+
+| Finding | Final disposition |
+| --- | --- |
+| R1 | Resolved: production diagnostic field and real-dispatch regression. |
+| R2 | Resolved: invalid no-ID envelopes observed; notifications excluded. |
+| R3 | Resolved: optional guarded request measurement; omitted/enabled surrogate regression. |
+| R4 | Resolved: late terminal loss counted while healthy recording continues. |
+| R5 | Resolved: scoped sequences, bounded scan/sections, truncation flags, sanitized shapes and measurement counts. |
+| R6 | Resolved: journal/page admission precedes migration publication; rejected bytes preserved. |
+| R7 | Resolved: only pristine zero-length v0 is admitted; prior private remnants rejected without wiping. |
+
+No unresolved source/privacy findings remain in this reviewed scope. The design's
+D1–D5 acceptance points are satisfied by the inspected construction and focused
+evidence, within the operating guide's local POSIX/trusted-directory and synchronous
+I/O limits. Same-user access remains outside the confidentiality/tamper boundary;
+shape summaries remain correlation evidence and cannot qualify observer-v3 replay,
+agent intent, schema acquisition, tokens or billing.
+
+Current exact SHA256 binding (including the accepted test-only addendum below):
+
+| File | SHA256 |
+| --- | --- |
+| `tools/standards_engine/standards_engine/interaction_observer.py` | `bc14652d3aa78ea400495d7f89fe9c4662c4de16acf759cac41d8279239a2a78` |
+| `tools/standards_engine/standards_engine/mcp.py` | `3852b7531dbf53d9c3eceec5503c7544de7df48aee714f618542b57754ae8596` |
+| `tools/standards_engine/tests/test_interaction_observer.py` | `3e39503185582b2ebeb13d86d92f65015e6ddb33f9133fee66db469b0e78d9a3` |
+| `tools/standards_engine/MCP-OBSERVABILITY.md` | `852dae36046f4dd54c912c6e0efc7f7c5f722a4d31723eb647956b6903e2743e` |
+| `tools/standards_engine/README.md` | `02fc2e17d6e4479d70fcb62caa189516db4e205ddfdc381a1b6ed721ed9ac15b` |
+
+Any subsequent substantive change to these files invalidates this exact-source
+binding and needs a review scoped to the changed claims. This reviewer performed
+no publication or installation action and wrote only this report.
+
+## Accepted test-only addendum
+
+The sole subsequent source delta adds
+`test_catalog_revision_report_attributes_exact_operation_schema_changes`. Removing
+that method reproduces the preceding accepted test-file SHA256
+`d9071a506955f0f2014d1901eec2f58ab01cb1a412c2ac4e56324480b59be3c3` exactly.
+The production source, operating guide and README still match their accepted
+hashes above.
+
+The new synthetic test creates two recorder sessions with an explicit catalog
+addition, removal and changed input type, then requires the exact corresponding
+report categories and preserved synthetic implementation attribution. Its expected
+categories follow directly from those independently specified fixture changes.
+It proves recorder/report catalog comparison, not authentic runtime provenance or
+model/client qualification. No findings against the delta.
+
+Independent targeted command:
+
+```sh
+python -m unittest tools.standards_engine.tests.test_interaction_observer.ObserverTests.test_catalog_revision_report_attributes_exact_operation_schema_changes
+```
+
+Result: one test passed in 0.003 seconds. The owner's completed focused-suite log
+at `/tmp/issue3-observer-tests.log` was also inspected: 23 tests passed in 25.572
+seconds. That log is supporting owner evidence; the independent complete-suite
+run remains the 22-test result above, with this one-test delta independently run.
+Independent source/privacy review remains satisfied at the updated exact binding.
+
+## Accepted constant-ownership addendum
+
+The final production delta imports the recorder's three default retention/quota
+constants into the MCP CLI and names the existing 8192-byte shape limit
+`MAX_SHAPE_BYTES`. Independent reverse substitution reconstructs the two preceding
+accepted source hashes exactly: observer
+`17751c69e10db84d391120738079b8389dea1cb43626fd08139bb11498b5421f` and MCP
+`153b3dd86023c9e6f5677889474ee224c28cb20e8945ba09de7eb235402d253b`.
+There are no other source changes in this delta.
+
+The shared defaults evaluate to 10000 calls, 604800 seconds and 33554432 bytes,
+identical to the replaced CLI expressions. The named shape limit evaluates to
+8192 and is used in the same comparison. This supplies a construction proof of
+equivalent behavior while giving those policies one owner; no new test is needed
+for the literal-to-constant replacement. Runtime implementation identity naturally
+changes with source material, as its existing contract requires.
+
+Tests, operating guide and README still match their accepted hashes. The current
+table above binds the new production hashes. Independent source/privacy review
+remains satisfied with no new findings. The owner's final combined focused run
+and broader Engine/baseline-failure disposition remain separate evidence gates.
diff --git a/docs/plans/mcp-observability/reports/verification.md b/docs/plans/mcp-observability/reports/verification.md
new file mode 100644
index 00000000..ec0536ac
--- /dev/null
+++ b/docs/plans/mcp-observability/reports/verification.md
@@ -0,0 +1,139 @@
+# Observability local verification and publication handoff
+
+Date: 2026-10-14. Separate checkout: `/workspace/Coding-Standards`.
+Branch: `feat/issue3-mcp-observability`. Baseline:
+`188beda1fa477d21d576c233dd8c7f4c4c267d23`.
+Implementation commit: `e2334869ccae2ee6187292c4493efaea4b80b342`.
+The following evidence commit owns this report; use `git rev-parse HEAD` for the
+final handoff OID. No normative standards, generated agent contracts, Tuldok,
+observer-v3 code/artifacts or existing installation configuration changed.
+
+## Current fixture follow-up result (2026-10-15)
+
+The bounded PR #4 fixture repair is qualified: resolve the policy from the same
+immutable snapshot, assert it exists, use semantic_revision and its successor,
+and keep setup outside the counting mocks. The real result assertions, validator,
+normative content and both count assertions are unchanged. Only test behavior and
+its registered generated-input digest changed; production observer hashes remain
+those accepted by the original source/privacy review.
+
+| Required follow-up check | Result |
+| --- | --- |
+| Formerly failing case | 1 test passed in 3.748 s |
+| Complete analysis module | 23 tests passed in 113.275 s |
+| Full serial Engine suite | 636 tests passed in 2161.675 s; one uninterrupted invocation with PYTHONPATH=. |
+| Independent fixture review | Satisfied; targeted case independently passed in 3.676 s; exact test-file SHA256 recorded |
+| Generated input verification | Public Engine refresh returned 73 suites / 121 checks passed before final frozen-source runs |
+
+See [fixture review](fixture-review.md). The first broader attempts caught the
+stale registered test-input digest and were interrupted; no partial results count
+as acceptance. After the coordinated generated-input refresh, both requested
+broader commands were restarted from zero and passed as stated above.
+
+The existing Git transport successfully published the original head 4887b7af
+without authentication changes; the parent opened draft PR #4. The invalid CLI
+token was specific to that CLI connection, not proof that Git transport failed.
+The verified fixture follow-up is packaged for the approved existing branch push;
+Git/PR refs own the actual publication identity. Parent owns hosted CI and merge.
+No existing installation was enabled and no actual interaction log is published.
+
+## Original observability qualification (before fixture repair)
+
+The evidence below preserves the initial baseline comparison and its historical
+I1 failure. That failure is now resolved by the qualified follow-up above; it has
+not been rewritten as a successful run of the old fixture.
+
+## Evidence and limits
+
+Environment: Linux/POSIX, CPython 3.12.14, SQLite 3.53.1. Installed versions matched
+`tools/standards_contracts/requirements.lock`: attrs 26.1.0, jsonschema 4.26.0,
+jsonschema-specifications 2025.9.1, referencing 0.37.0, rpds-py 2026.6.3,
+typing-extensions 4.16.0. No dependencies were added or upgraded.
+
+| Check | Observed result |
+| --- | --- |
+| Recorder/MCP/runtime identity focused run | 40 tests passed in 42.362 s:23 recorder,13 existing MCP,4 runtime identity |
+| Independent GPT-6.1 Sol High design/source/privacy review | Satisfied; all R1–R7 resolved, exact final source hashes recorded |
+| Observer-v3 qualification/replay suites | 39 tests passed in 19.044 s |
+| Broader Engine coverage | All 636 discovered cases executed;635 pass with CI environment,1 pre-existing fixture failure |
+| Generated contract projection | `projection --check` passed; no generated agent contract changed |
+| Declarative verification | 73 suites,121 checks passed; generated suite-input manifest refreshed through Engine |
+| Four-configuration cold baseline comparison | Same six replies, three tool calls and byte totals with recording off/on |
+| Staged and branch review | Expected text-only write set; diff whitespace checks pass; no raw logs/databases/secrets staged |
+
+The broader run first completed 93 cases serially, then was interrupted during
+the next module so remaining independent modules could run in four isolated
+Python processes. Only fully completed earlier modules count toward the93;
+the interrupted module was re-run completely. The 56 remaining modules executed
+543 cases. This is aggregate coverage, not a claim that one serial full-suite
+invocation was green. The CI environment sets `PYTHONPATH=.`. An initial
+`test_efficiency_trace` child-process error without it occurred on both baseline
+and candidate; re-running each with that documented setting passed (candidate
+45.363 s, baseline 43.920 s). Its corrected pass is included in 635.
+
+The original remaining failure was
+`test_analysis.AnalysisWorkflowTest.test_proposal_analysis_uses_one_projection_and_one_verified_base`:
+`AUTHORING.INVALID_SEMANTIC_REVISION`. It failed identically in a clean Git clone of
+the baseline, because the fixture hardcodes accepted revision 1/proposed 2. The
+recorder never participates in that test. See [I1](../issues.md). It was initially
+deferred, then repaired under explicit follow-up authority as recorded above. The
+initial qualification is not relabeled as a green full-suite run.
+
+The recorder regressions cover opt-in omission, unchanged catalog/outcomes,
+actual validation feedback, malformed envelopes/JSON, arbitrary-output exclusion,
+schema-position privacy and on-disk sentinels, tool exceptions/interruption,
+broken response delivery, failed health output, unavailable/locked/full SQLite,
+transactional initialization/reopening/interruption, unsupported/corrupt/nonpristine
+files, count/age/session retention, in-flight eviction, read-only reports and
+bounded scan/scope/truncation, catalog schema differences, cold stdio replacement
+and concurrent first-open/writers. Cold process byte sums equal actual stdio
+streams excluding notifications from paired-request measurements. Killed requests
+remain unmatched/unknown after replacement. No client receipt, intent, token,
+billing, tamper-proof audit or reconstruction of redacted requests is claimed.
+
+No model qualification turns or real interaction logs were collected. The
+requested independent review used the specified GPT-6.1 Sol High reviewer. All
+process fixtures and observation stores were synthetic, private and temporary;
+their raw contents are not publication deliverables. Historical observer-v3
+verdicts and transcripts remain untouched.
+
+## Reproduction
+
+```sh
+PYTHONPATH=. python3 -m unittest tools.standards_engine.tests.test_interaction_observer tools.standards_engine.tests.test_mcp tools.standards_engine.tests.test_runtime_identity
+PYTHONPATH=. python3 -m unittest tools.standards_engine.tests.test_model_discovery_qualification tools.standards_engine.tests.test_replay_discovery_qualification tools.standards_engine.tests.test_replay_discovery_workflow
+PYTHONPATH=. python3 -m unittest discover -s tools/standards_engine/tests
+python3 -m tools.standards_contracts.standards_contracts.projection --check
+PYTHONPATH=.:tools/standards_verifier python3 tools/standards_verifier/verify.py --complete
+```
+
+The broad command exposed I1 at original head 4887b7af; after the bounded fixture
+repair the current full serial suite passes all 636 tests.
+[Cold comparison](baseline-comparison.md) gives the workload and byte totals.
+[Independent review](independent-review.md) binds final production/test/docs hashes.
+[Operating contract](../../../../tools/standards_engine/MCP-OBSERVABILITY.md) owns
+configuration, safe disclosure, storage/health/recovery/retention and reporting.
+
+## Handoff resources
+
+Retain this clean checkout and feature branch for parent publication and review;
+no rebase, amend or history rewrite has occurred. Synchronization is explicit:
+parent checks current target authority before publication/integration and
+reverifies any material change. Retirement follows the parent's accepted merge,
+rejection or explicit retained-resource decision; this task grants no destructive
+cleanup authority. No other mounted repository was modified.
+
+The original `/workspace/issue3-mcp-observability.bundle` remains preserved at
+pre-fixture head 4887b7af. The qualified fixture follow-up is packaged separately
+as `/workspace/issue3-mcp-observability-fixture-repair.bundle`, outside the
+repository and not staged. It requires the named baseline and preserves its
+exact feature ref/head. In an authorized checkout containing that baseline:
+
+```sh
+git bundle verify /workspace/issue3-mcp-observability-fixture-repair.bundle
+git fetch /workspace/issue3-mcp-observability-fixture-repair.bundle feat/issue3-mcp-observability:feat/issue3-mcp-observability
+```
+
+PR #4 already exists. Parent may inspect the exact qualified branch and its
+publication ref through working authorized access. Merge and installation
+enablement remain separate decisions. No actual telemetry or qualification trace accompanies it.
diff --git a/evaluation/standards-effectiveness/generated/suite-inputs.json b/evaluation/standards-effectiveness/generated/suite-inputs.json
index b391480c..33ae40f2 100644
--- a/evaluation/standards-effectiveness/generated/suite-inputs.json
+++ b/evaluation/standards-effectiveness/generated/suite-inputs.json
@@ -7301,7 +7301,7 @@
       ]
     },
     {
-      "digest": "sha256:ebee8cc5b4d4e7bc2d6b6a89c6ea8d4f5b6ce5d6d135e629dbfb67cbede347a9",
+      "digest": "sha256:02fc2e17d6e4479d70fcb62caa189516db4e205ddfdc381a1b6ed721ed9ac15b",
       "path": "tools/standards_engine/README.md",
       "state": "present",
       "uses": [
@@ -7614,6 +7614,18 @@
         }
       ]
     },
+    {
+      "digest": "sha256:bc14652d3aa78ea400495d7f89fe9c4662c4de16acf759cac41d8279239a2a78",
+      "path": "tools/standards_engine/standards_engine/interaction_observer.py",
+      "state": "present",
+      "uses": [
+        {
+          "check": "python-package-contract",
+          "role": "package-contract-input",
+          "suite": "a1c-public-cutover"
+        }
+      ]
+    },
     {
       "digest": "sha256:bac3a3330851489f35542e775ecf61c2f04822b17d23806803b8da8d2fb4b506",
       "path": "tools/standards_engine/standards_engine/logical_authoring.py",
@@ -7627,7 +7639,7 @@
       ]
     },
     {
-      "digest": "sha256:d6cfe64a594c017f2e3be53dc3f70c561c9fb9d24ebe98f78a7e5ca43ceaa964",
+      "digest": "sha256:3852b7531dbf53d9c3eceec5503c7544de7df48aee714f618542b57754ae8596",
       "path": "tools/standards_engine/standards_engine/mcp.py",
       "state": "present",
       "uses": [
@@ -7817,7 +7829,7 @@
       ]
     },
     {
-      "digest": "sha256:c463c48d95bcbc5b5bfd3690668198897a251c5000e59db71f2cbc5f983d71af",
+      "digest": "sha256:7f51e403f03f5c57be8b1ca5736a79bcf139c87dd9eb4703ce3f7e59f1ba3d01",
       "path": "tools/standards_engine/tests/test_analysis.py",
       "state": "present",
       "uses": [
@@ -14058,7 +14070,7 @@
     "path": "evaluation/standards-effectiveness/suite-registry.toml"
   },
   "repository_index": {
-    "digest": "sha256:8607daad3aec322a6072b9964512416edd6688958032b1a6cb7e2217b9ab34dd",
+    "digest": "sha256:32d5e126dd74c0349f8b402496260d9184db813209fc2e6de80ed4259f66bef9",
     "uses": [
       {
         "check": "python-package-contract",
diff --git a/tools/standards_engine/MCP-OBSERVABILITY.md b/tools/standards_engine/MCP-OBSERVABILITY.md
new file mode 100644
index 00000000..397ece6d
--- /dev/null
+++ b/tools/standards_engine/MCP-OBSERVABILITY.md
@@ -0,0 +1,187 @@
+# Operator MCP interaction observations
+
+This optional local recorder observes the existing MCP boundary. It does not add
+agent tools, modify domain outcomes or workflow authority, make model calls, retry
+operations, or change the interface-46 catalog. The qualification observer remains
+**observer v3**, with its existing raw-event acquisition, assessment and exact-source
+replay. The production recorder does not import the harness or rewrite verdicts.
+
+## Configuration and authority
+
+Recording is off unless the operator explicitly supplies **both** options below.
+The commands illustrate a new operator-owned installation, not instructions to
+enable an existing one without its operator's approval. Replace paths and the
+nonsensitive deployment label with operator-selected values:
+
+```sh
+python3 -m tools.standards_engine.standards_engine.mcp \
+  --repo-root /operator/standards --purpose authoring \
+  --observe-database /operator/private-diagnostics/interactions.sqlite \
+  --observe-deployment qualification-fixture
+```
+
+The database directory must already exist, be operator controlled and exclude
+concurrent untrusted replacement of the directory or its ancestors. This initial
+recorder supports POSIX regular local files and SQLite rollback journals only.
+It refuses final-component symlinks, hardlinked files, existing group/world-readable
+files, unsupported journal modes and unsupported or unrelated database schemas.
+New files use mode 0600. It does not chmod existing files, create directories or
+use an Engine domain database. Do not place it in a repository, tracked directory,
+network share or an agent-visible artifact directory. The operator must select
+appropriate disk access, encrypted storage/backup and disposal arrangements.
+
+**Same-user filesystem access is not a security boundary.** A process with the
+same OS authority can read, modify or remove the recorder database and stderr.
+Tool-catalog invisibility does not create confidentiality or tamper evidence.
+The recorder therefore excludes payload values regardless of filesystem modes.
+The deployment label must not contain account identities, credentials or secrets.
+No network exporter is included. Export/sharing requires separate operator review.
+
+Policy defaults and valid ranges:
+
+| Option | Default | Range / meaning |
+| --- | --- | --- |
+| `--observe-max-calls` | 10000 | 16–100000 retained interaction starts, across all processes |
+| `--observe-max-age-seconds` | 604800 | 60–31536000 seconds since begin |
+| `--observe-max-bytes` | 33554432 | 1048576–1073741824 bytes, SQLite main-file page quota |
+
+All processes sharing a database must use the same privacy/retention policy.
+Each connection enforces its configured page quota; mixed policies do not establish
+a shared minimum. Rollback journals may temporarily add up to roughly another
+main database's size, plus SQLite bookkeeping and bounded report/sort working space.
+The directory's filesystem quota remains the final bound on all files. No WAL,
+background thread, queue or telemetry write retry is introduced. Main files do not
+shrink on deletion: SQLite reuses freed pages. No automatic VACUUM/rebuild occurs.
+
+## Persisted contract and lifecycle
+
+Schema v1 has three owners within one recorder database:
+
+- `sessions`: exact startup runtime metadata (process instance UUID, interface,
+  purpose, catalog/schema/implementation digests), deployment label and per-tool
+  input/output schema digests. These bind the actually loaded source and effective
+  eager/on-demand catalog. They do not prove a client refreshed its catalog.
+- `interactions`: immutable begin evidence, recorder UUID, session-local ordinal,
+  wall-clock begin instant, method/installed operation, sanitized input shape and
+  its shape-only digest, safe discovery target, expected-catalog match state,
+  request byte measurement and delivery scope. This row is the begin event.
+- `events`: at most one independent end/error/incomplete observation per begin,
+  with monotonic elapsed milliseconds, wall-clock finish instant, response bytes,
+  fixed outcome class and sanitized validation locations/constraint keywords.
+
+Each mutation owns `BEGIN IMMEDIATE`/commit/rollback. Schema admission is checked
+under the same lock, including exact owned DDL and application ID. Zero-length new version-0
+SQLite files migrate transactionally once to v1; v1 reopens without rewriting
+observations. Future versions, unrelated/partial schemas and corruption are
+unavailable, never deleted or replaced. Migration interruption rolls back under
+SQLite recovery. Backup a stopped database before any future supported migration;
+no downgrade migration or automatic restoration is currently provided.
+
+Stdio begins are committed before dispatch. End/error evidence is committed only
+after the ordinary response write and flush return. `stdio-flushed` does not prove
+the client received/processed the response. The direct Python `dispatch` path uses
+`constructed` scope, which proves only that a response was produced. Parsing errors
+and malformed requests with responses are observed; valid notifications have no
+request/response pair and are excluded. Client IDs and clientInfo are not retained.
+
+A caught tool exception has a fixed `tool-exception` category; domain outcome may
+be unknown. Typed validation rejections retain `input-invalid` plus safe field
+locations. Other protocol/domain classes use fixed vocabularies, never exception
+messages, tracebacks or arbitrary result codes. Transport/interruption errors
+append `incomplete/unknown` if storage remains available. A killed process or a
+failed terminal write leaves an unmatched begin. Reports label it unknown; startup
+does not complete old calls or presume other live recorders died. Telemetry cannot
+prove a domain effect did or did not commit and cannot authorize a mutation retry.
+
+Age and count retention run transactionally on each new observed request. Expiry
+is event-driven: an idle/stopped database retains expired records until recording
+resumes or the operator disposes of it. Starts and their terminal evidence are
+removed together; unreferenced session manifests are removed in the same
+transaction. Retention can evict an in-flight call from another process. Its late
+terminal is counted as dropped and recording continues; no start is recreated.
+Retention can remove evidence before a later success, so history/repair denominators
+are limited to retained observations. Per-session prune counters themselves expire
+with their session; they are not installation-lifetime loss totals. Secure deletion
+is enabled for recorder deletes; it does not erase backups, filesystem snapshots,
+SQLite journal history or storage-device remnants.
+
+## Privacy projection
+
+No credentials/tokens, scalar input values, raw request IDs, complete prompts,
+model transcripts, standards/repository bodies, domain handles or arbitrary tool
+outputs are stored or hashed. Every scalar input becomes only its type.
+
+At each structural location, only property names declared by the installed input
+schema can appear. Recursive references/unions supply possible property names;
+this is a disclosure projection, not another input validator. Dynamic map keys
+and unknown names are removed and counted. Sensitive subtrees (`body`, `content`,
+`text`, `prompt`, `message`, `rationale`, `reason`, `evidence`, `details`, `metadata`,
+credentials/password/token/secret/authorization/api_key/environment/env) are replaced
+by a redacted marker. Other input fields still retain no scalar values. Shapes
+have at most 64 visited nodes, depth 5, four sampled array items and 8192 serialized
+ASCII bytes. Truncation is explicit. Validation feedback is limited to eight
+issues, depth-five schema-owned pointer components, wildcard indices/dynamic keys
+and fixed constraint keywords; messages and offending values are omitted.
+
+An input-shape digest groups structural samples, not raw values or exact semantic
+variants. Two scalar variants can deliberately have the same shape. Discovery
+operation values are retained only on exact installed-operation membership.
+Expected catalogs become only unspecified/matches/differs. Selector values are
+not retained. There is no legitimate cross-process session/task ID in the current
+MCP server: reports do not infer continuation from prompts or domain handles.
+
+## Health and recovery
+
+Operator stderr receives fixed JSON health at recorder startup, first degradation,
+retention-dropped terminal observations, and orderly close. It carries state,
+fixed reason, starts/terminals recorded, observations dropped and interactions
+pruned in that process. Stderr failure itself is guarded. Health is not advertised
+in the agent catalog or appended to an Engine result.
+
+Unavailable, locked (20 ms busy budget), full, unsafe, unsupported or corrupt
+storage disables recording for that process and reports degraded health. Existing
+begin evidence remains independent; no completion is invented. No further writes
+or retries occur. The MCP operation still returns its original outcome and runs
+once, including a domain result committed before the telemetry failure. A failed
+start means that interaction may have no retained evidence at all. Health counters
+cover attempted observations, not an exact census of lost interactions.
+
+The operator must monitor stderr/process supervision independently of SQLite,
+fix the owned storage condition and restart the MCP process to recover recording.
+Preserve a corrupt/unrecognized database for diagnosis; select a separately
+approved fresh path rather than silently rebuilding it. A report cannot recover
+health messages that were unavailable/lost with stderr or process termination.
+
+## Read-only inspection
+
+```sh
+python3 -m tools.standards_engine.standards_engine.interaction_observer \
+  --database /operator/private-diagnostics/interactions.sqlite --limit 100
+```
+
+This operator CLI opens an existing v1 database in SQLite read-only/query-only
+mode and uses one read snapshot. It never creates, migrates, prunes or modifies
+observations. JSON goes only to the operator's stdout; no output file is created.
+Each group/session/histogram section is bounded by `--limit` (1–1000), with explicit
+truncation flags. Sequence analysis scans at most the latest 10000 retained starts;
+its scan count and truncation are reported. Aggregate queries scan the bounded
+retained database (at most the configured 100000 starts). Sessions show the latest
+bounded set in chronological order; catalog comparisons cover those shown manifests.
+
+The report provides operation usage, error distribution by structural input shape,
+sanitized field/constraint failures, observed latency and serialized bytes with
+measurement counts, sanitized shapes for successful and failed calls, session
+identity and catalog additions/removals/schema changes. Totals include all retained
+starts/terminals; operation/failure/shape rows identify their delivery scope and
+have their own counts. An end is a non-error MCP response, not evidence the domain
+workflow completed. Missing measurements are null and have separate denominators.
+
+Sequence histograms count preceding observed errors/discoveries before a non-error
+tool response, by session, delivery scope and operation. Their denominator is the
+reported number of non-error responses in the bounded scan. This is a diagnostic
+candidate for investigating repair loops, not proof that errors caused discovery,
+that a schema was acquired before use, or that separate tasks share agent intent.
+No raw trace or redacted request can be reconstructed. Bytes are JSON/stdio bytes,
+not model tokens or provider billing. Preserve independent qualification evidence
+for observer-v3 replay and exact behavioral assessment; do not export actual logs
+with code or automatically rewrite contracts from these summaries.
diff --git a/tools/standards_engine/README.md b/tools/standards_engine/README.md
index bb0473ef..b428f361 100644
--- a/tools/standards_engine/README.md
+++ b/tools/standards_engine/README.md
@@ -21,6 +21,10 @@ See [the publication evidence contract](PURPOSE-SEPARATION.md#publication-eviden
 
 The versioned sections below retain implementation history.
 
+Optional operator-only MCP interaction recording is documented in
+[MCP observability](MCP-OBSERVABILITY.md). It is disabled by default and preserves
+the existing qualification observer v3 and agent-visible interface.
+
 ## Purpose separation (interface 31)
 
 Version 0.2.0 / interface 31 requires host-selected `application` or `authoring`
diff --git a/tools/standards_engine/standards_engine/interaction_observer.py b/tools/standards_engine/standards_engine/interaction_observer.py
new file mode 100644
index 00000000..197b060a
--- /dev/null
+++ b/tools/standards_engine/standards_engine/interaction_observer.py
@@ -0,0 +1,600 @@
+"""Opt-in operator observations. Never an Engine store or an agent tool.
+
+The recorder owns sanitization, SQLite transactions, retention and fail-open
+health. No caller-supplied scalar is persisted. A missing terminal event means
+unknown outcome; restarting a recorder never manufactures a completion.
+"""
+from __future__ import annotations
+
+import argparse
+from collections import Counter
+from dataclasses import dataclass
+import hashlib
+import json
+import os
+from pathlib import Path
+import re
+import sqlite3
+import stat
+import sys
+import time
+from typing import TextIO
+import uuid
+
+SCHEMA_VERSION = 1
+APPLICATION_ID = 0x53454F42  # SEOB, separate from every Engine domain database.
+DEFAULT_MAX_CALLS = 10_000
+DEFAULT_MAX_AGE_SECONDS = 7 * 24 * 60 * 60
+DEFAULT_MAX_BYTES = 32 * 1024 * 1024
+BUSY_TIMEOUT_MS = 20
+MAX_SHAPE_NODES = 64
+MAX_SHAPE_DEPTH = 5
+MAX_SHAPE_BYTES = 8 * 1024
+MAX_ARRAY_ITEMS = 4
+MAX_FEEDBACK_ISSUES = 8
+MAX_FEEDBACK_POINTER_CHARS = 512
+MAX_REPORT_ROWS = 1000
+MAX_SEQUENCE_SCAN = 10_000
+METHODS = frozenset({'initialize', 'tools/list', 'tools/call', 'ping'})
+SENSITIVE_FIELDS = frozenset({
+    'body', 'content', 'text', 'prompt', 'message', 'rationale', 'reason',
+    'evidence', 'details', 'metadata', 'credentials', 'password', 'token',
+    'secret', 'authorization', 'api_key', 'environment', 'env',
+})
+CONSTRAINTS = frozenset({
+    'type', 'required', 'additionalProperties', 'enum', 'const', 'oneOf',
+    'anyOf', 'allOf', 'not', 'minimum', 'maximum', 'minItems', 'maxItems',
+    'minLength', 'maxLength', 'pattern', 'format', 'uniqueItems',
+})
+OUTCOMES = frozenset({'invalid', 'unsupported', 'unavailable', 'stale',
+                      'rejected', 'recovery-required', 'needs-action',
+                      'requires-change', 'complete', 'ready'})
+
+
+SCHEMA_DDL = (
+    'CREATE TABLE sessions (id TEXT PRIMARY KEY, started REAL NOT NULL, runtime TEXT NOT NULL, deployment TEXT NOT NULL, catalog TEXT NOT NULL, pruned INTEGER NOT NULL DEFAULT 0)',
+    'CREATE TABLE interactions (id TEXT PRIMARY KEY, session TEXT NOT NULL REFERENCES sessions(id), ordinal INTEGER NOT NULL, started REAL NOT NULL, method TEXT NOT NULL, operation TEXT NOT NULL, input_shape TEXT NOT NULL, shape_digest TEXT NOT NULL, discovery_operation TEXT, catalog_state TEXT NOT NULL, request_bytes INTEGER, scope TEXT NOT NULL, UNIQUE(session, ordinal))',
+    "CREATE TABLE events (interaction TEXT PRIMARY KEY REFERENCES interactions(id) ON DELETE CASCADE, phase TEXT NOT NULL CHECK(phase IN ('end','error','incomplete')), finished REAL NOT NULL, duration_ms REAL NOT NULL, response_bytes INTEGER, outcome TEXT NOT NULL, feedback TEXT NOT NULL)",
+    'CREATE INDEX interaction_age ON interactions(started)',
+)
+
+
+def _json(value: object) -> str:
+    return json.dumps(value, ensure_ascii=True, sort_keys=True, separators=(',', ':'))
+
+
+def _digest(value: object) -> str:
+    return 'sha256:' + hashlib.sha256(_json(value).encode()).hexdigest()
+
+
+def serialized_bytes(value: object) -> int:
+    """JSON representation bytes, not tokens; avoid retaining a second payload."""
+    return sum(len(part.encode('utf-8')) for part in json.JSONEncoder().iterencode(value))
+
+
+def _kind(value: object) -> str:
+    if value is None:
+        return 'null'
+    if isinstance(value, bool):
+        return 'boolean'
+    if isinstance(value, str):
+        return 'string'
+    if isinstance(value, (int, float)):
+        return 'number'
+    return 'object' if isinstance(value, dict) else 'array' if isinstance(value, list) else 'unknown'
+
+
+class InputShape:
+    """Value-free disclosure projection, not a second input validator.
+
+    Static property names come only from the installed schema. Unknown names and
+    every scalar value disappear before SQLite sees anything. Sensitive children
+    are not traversed even if their properties occur elsewhere in the schema.
+    """
+    def __init__(self, schema: dict) -> None:
+        self.schema = schema
+        self.definitions = schema.get('$defs', {})
+
+    def _expanded(self, nodes: list[dict]) -> list[dict]:
+        result, pending, seen = [], list(nodes), set()
+        while pending:
+            node = pending.pop()
+            if not isinstance(node, dict) or id(node) in seen:
+                continue
+            seen.add(id(node))
+            result.append(node)
+            reference = node.get('$ref', '')
+            if reference.startswith('#/$defs/'):
+                pending.append(self.definitions.get(reference[len('#/$defs/'):], {}))
+            for union in ('oneOf', 'anyOf', 'allOf'):
+                pending.extend(node.get(union, []))
+        return result
+
+    def _properties(self, nodes: list[dict]) -> dict[str, list[dict]]:
+        properties: dict[str, list[dict]] = {}
+        for node in self._expanded(nodes):
+            for name, child in node.get('properties', {}).items():
+                properties.setdefault(name, []).append(child)
+        return properties
+
+    def _items(self, nodes: list[dict]) -> list[dict]:
+        return [node['items'] for node in self._expanded(nodes)
+                if isinstance(node.get('items'), dict)]
+
+    def project(self, value: object) -> dict:
+        budget = MAX_SHAPE_NODES
+        def visit(item, nodes, depth):
+            nonlocal budget
+            budget -= 1
+            if depth >= MAX_SHAPE_DEPTH or budget <= 0:
+                return {'type': _kind(item), 'truncated': True}
+            if isinstance(item, dict):
+                properties = self._properties(nodes)
+                fields, unknown, examined = {}, 0, 0
+                for name, child in item.items():
+                    if examined >= MAX_SHAPE_NODES or budget <= 0:
+                        break
+                    examined += 1
+                    if name not in properties:
+                        unknown += 1
+                        continue
+                    fields[name] = ('redacted' if name.lower() in SENSITIVE_FIELDS
+                                    else visit(child, properties[name], depth + 1))
+                return {'type': 'object', 'fields': fields,
+                        'unknown_fields': unknown, 'truncated': len(item) > examined}
+            if isinstance(item, list):
+                values, children = [], self._items(nodes)
+                for child in item[:MAX_ARRAY_ITEMS]:
+                    if budget <= 0:
+                        break
+                    values.append(visit(child, children, depth + 1))
+                return {'type': 'array', 'items': values,
+                        'truncated': len(item) > len(values)}
+            return {'type': _kind(item)}
+        shape = visit(value, [self.schema], 0)
+        if len(_json(shape)) > MAX_SHAPE_BYTES:
+            return {'type': _kind(value), 'truncated': True}
+        return shape
+
+    def feedback(self, value: object) -> list[dict]:
+        if not isinstance(value, dict) or not isinstance(value.get('issues'), list):
+            return []
+        result = []
+        for issue in value['issues'][:MAX_FEEDBACK_ISSUES]:
+            if not isinstance(issue, dict):
+                continue
+            pointer = issue.get('instance_pointer', '')
+            parts = pointer[:MAX_FEEDBACK_POINTER_CHARS].split('/', MAX_SHAPE_DEPTH + 1)[1:MAX_SHAPE_DEPTH + 1] if isinstance(pointer, str) else []
+            safe, nodes = [], [self.schema]
+            for part in parts:
+                name = part.replace('~1', '/').replace('~0', '~')
+                properties, items = self._properties(nodes), self._items(nodes)
+                if name in properties:
+                    # Emit the canonical schema name/escaping, not caller text.
+                    safe.append(name.replace('~', '~0').replace('/', '~1'))
+                    nodes = [] if name.lower() in SENSITIVE_FIELDS else properties[name]
+                else:
+                    safe.append('*')
+                    nodes = items if part.isdecimal() else []
+            keyword = issue.get('keyword')
+            result.append({'pointer': '/' + '/'.join(safe) if safe else '',
+                           'keyword': keyword if isinstance(keyword, str) and keyword in CONSTRAINTS else 'other'})
+        return result
+
+
+@dataclass(frozen=True)
+class RecorderConfig:
+    database: Path
+    deployment: str
+    max_calls: int = DEFAULT_MAX_CALLS
+    max_age_seconds: int = DEFAULT_MAX_AGE_SECONDS
+    max_bytes: int = DEFAULT_MAX_BYTES
+
+    def __post_init__(self):
+        if not re.fullmatch(r'[A-Za-z0-9][A-Za-z0-9_.-]{0,63}', self.deployment):
+            raise ValueError('deployment must be a nonsensitive label of 1-64 characters')
+        if not 16 <= self.max_calls <= 100_000:
+            raise ValueError('max_calls must be in 16..100000')
+        if not 60 <= self.max_age_seconds <= 365 * 24 * 60 * 60:
+            raise ValueError('max_age_seconds must be in 60..31536000')
+        if not 1024 * 1024 <= self.max_bytes <= 1024 * 1024 * 1024:
+            raise ValueError('max_bytes must be in 1 MiB..1 GiB')
+
+
+@dataclass(frozen=True)
+class Observation:
+    identity: str
+    started: float
+    operation: str
+
+
+class StorageRejected(Exception):
+    """Fixed operator code; never includes paths, SQL or payloads."""
+
+
+class InteractionRecorder:
+    def __init__(self, config: RecorderConfig, runtime: dict, catalog: list[dict],
+                 *, diagnostics: TextIO | None = None) -> None:
+        self.config = config
+        self.runtime = dict(runtime)
+        self._diagnostics = diagnostics if diagnostics is not None else sys.stderr
+        self._connection: sqlite3.Connection | None = None
+        self._active: dict[str, Observation] = {}
+        self._sequence = 0
+        self._closed = False
+        self.health = {'state': 'starting', 'reason': None, 'recorded_starts': 0,
+                       'recorded_terminals': 0, 'dropped_observations': 0,
+                       'pruned_interactions': 0}
+        self._shapes = {tool['name']: InputShape(tool['inputSchema']) for tool in catalog}
+        self._catalog = {tool['name']: {
+            'input_digest': _digest(tool['inputSchema']),
+            'output_digest': (tool.get('_meta', {}).get('standards-engine/output-schema-digest')
+                              or _digest(tool.get('outputSchema'))),
+        } for tool in catalog}
+        try:
+            self._open()
+            self.health['state'] = 'healthy'
+            self._emit()
+        except Exception as error:
+            self._degrade(error)
+
+    def _emit(self) -> None:
+        try:
+            print('standards-engine observer ' + _json(self.health), file=self._diagnostics, flush=True)
+        except Exception:
+            # Operator stderr can itself be broken. Never change a domain result.
+            pass
+
+    def _degrade(self, error: Exception) -> None:
+        reason = 'storage-unavailable'
+        if isinstance(error, StorageRejected):
+            reason = str(error)
+        elif isinstance(error, sqlite3.Error):
+            reason = {sqlite3.SQLITE_BUSY: 'storage-locked', sqlite3.SQLITE_LOCKED: 'storage-locked',
+                      sqlite3.SQLITE_FULL: 'storage-full', sqlite3.SQLITE_CORRUPT: 'storage-corrupt',
+                      sqlite3.SQLITE_NOTADB: 'storage-corrupt'}.get(
+                          getattr(error, 'sqlite_errorcode', 0) & 255, 'storage-error')
+        self.health.update(state='degraded', reason=reason)
+        self.health['dropped_observations'] += 1
+        if self._connection is not None:
+            try:
+                self._connection.close()
+            except Exception:
+                pass
+            self._connection = None
+        self._emit()
+
+    def _open(self) -> None:
+        path = self.config.database
+        if os.name != 'posix':
+            raise StorageRejected('unsupported-filesystem')
+        # The operator owns the existing directory; no recursive mkdir or changes
+        # to existing paths/permissions. O_NOFOLLOW protects the final component.
+        descriptor = os.open(path, os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW, 0o600)
+        try:
+            info = os.fstat(descriptor)
+            if not stat.S_ISREG(info.st_mode) or info.st_mode & 0o077 or info.st_nlink != 1:
+                raise StorageRejected('unsafe-database-file')
+        finally:
+            os.close(descriptor)
+        db = sqlite3.connect(path, timeout=BUSY_TIMEOUT_MS / 1000, isolation_level=None)
+        self._connection = db
+        db.execute('PRAGMA foreign_keys=ON')
+        db.execute(f'PRAGMA busy_timeout={BUSY_TIMEOUT_MS}')
+        db.execute('BEGIN IMMEDIATE')
+        try:
+            # Reject unsupported capabilities before publishing a migration.
+            if db.execute('PRAGMA journal_mode').fetchone()[0] != 'delete':
+                raise StorageRejected('unsupported-journal-mode')
+            pages = self.config.max_bytes // db.execute('PRAGMA page_size').fetchone()[0]
+            if db.execute('PRAGMA page_count').fetchone()[0] > pages:
+                raise StorageRejected('database-over-budget')
+            db.execute(f'PRAGMA max_page_count={pages}')
+            version = db.execute('PRAGMA user_version').fetchone()[0]
+            app = db.execute('PRAGMA application_id').fetchone()[0]
+            tables = db.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall()
+            if version == 0 and app == 0 and not tables:
+                # Table absence cannot prove privacy: dropped SQLite content may
+                # remain in freelist pages or sqlite_master slack. Only a truly
+                # new zero-length source is admitted; never wipe/reuse old bytes.
+                if info.st_size != 0 or path.stat().st_size != 0:
+                    raise StorageRejected('nonpristine-database')
+                for statement in SCHEMA_DDL:
+                    db.execute(statement)
+                db.execute(f'PRAGMA application_id={APPLICATION_ID}')
+                db.execute(f'PRAGMA user_version={SCHEMA_VERSION}')
+            elif version != SCHEMA_VERSION or app != APPLICATION_ID:
+                raise StorageRejected('unsupported-database-schema')
+            actual = {row[0]: row[1] for row in db.execute(
+                "SELECT name,sql FROM sqlite_master WHERE name NOT LIKE 'sqlite_%'")}
+            expected = {statement.split()[2]: statement for statement in SCHEMA_DDL}
+            if actual != expected:
+                raise StorageRejected('invalid-database-schema')
+            db.execute('COMMIT')
+        except BaseException:
+            db.execute('ROLLBACK')
+            raise
+        db.execute('PRAGMA synchronous=FULL')
+        db.execute('PRAGMA secure_delete=ON')
+
+    def _transaction(self, action):
+        db = self._connection
+        if db is None:
+            self.health['dropped_observations'] += 1
+            return False
+        try:
+            db.execute('BEGIN IMMEDIATE')
+            action(db)
+            db.execute('COMMIT')
+            return True
+        except Exception as error:
+            try:
+                db.execute('ROLLBACK')
+            except sqlite3.Error:
+                pass
+            self._degrade(error)
+            return False
+
+    def begin(self, message: object, *, request_bytes: int | None, scope: str,
+              parse_error: bool = False) -> Observation | None:
+        if self._closed:
+            return None
+        if self._active:
+            self._degrade(StorageRejected('overlapping-recorder-call'))
+            return None
+        # Notifications have no response pair. Invalid envelopes and parse errors do.
+        if (isinstance(message, dict) and 'id' not in message and not parse_error
+                and message.get('jsonrpc') == '2.0' and isinstance(message.get('method'), str)):
+            return None
+        if self._connection is None:
+            self.health['dropped_observations'] += 1
+            return None
+        try:
+            params = message.get('params', {}) if isinstance(message, dict) else {}
+            params = params if isinstance(params, dict) else {}
+            method = message.get('method') if isinstance(message, dict) else None
+            method = method if isinstance(method, str) and method in METHODS else 'invalid-json' if parse_error else 'other'
+            name = params.get('name')
+            operation = name if method == 'tools/call' and isinstance(name, str) and name in self._shapes else 'other'
+            shape = self._shapes[operation].project(params.get('arguments', {})) if operation in self._shapes else {}
+            arguments = params.get('arguments', {})
+            discovery = arguments.get('operation') if isinstance(arguments, dict) else None
+            discovery = discovery if operation in {'describe_input', 'describe_output'} and isinstance(discovery, str) and discovery in self._shapes else None
+            expected = arguments.get('expected_catalog') if isinstance(arguments, dict) else None
+            catalog_state = ('unspecified' if expected is None else 'matches'
+                             if expected == self.runtime['catalog_digest'] else 'differs')
+            now = time.time()
+            observation = Observation(str(uuid.uuid4()), time.monotonic(), operation)
+            self._sequence += 1
+            pruned = 0
+            def write(db):
+                nonlocal pruned
+                db.execute('INSERT OR IGNORE INTO sessions(id,started,runtime,deployment,catalog) VALUES(?,?,?,?,?)',
+                           (self.runtime['instance_id'], now, _json(self.runtime), self.config.deployment, _json(self._catalog)))
+                pruned += db.execute('DELETE FROM interactions WHERE started < ?',
+                                     (now - self.config.max_age_seconds,)).rowcount
+                excess = db.execute('SELECT max(0,count(*)-?) FROM interactions',
+                                    (self.config.max_calls - 1,)).fetchone()[0]
+                if excess:
+                    pruned += db.execute('DELETE FROM interactions WHERE id IN '
+                                         '(SELECT id FROM interactions ORDER BY started,id LIMIT ?)', (excess,)).rowcount
+                db.execute('DELETE FROM sessions WHERE id != ? AND NOT EXISTS '
+                           '(SELECT 1 FROM interactions WHERE session=sessions.id)', (self.runtime['instance_id'],))
+                db.execute('UPDATE sessions SET pruned=pruned+? WHERE id=?', (pruned, self.runtime['instance_id']))
+                db.execute('INSERT INTO interactions VALUES(?,?,?,?,?,?,?,?,?,?,?,?)',
+                           (observation.identity, self.runtime['instance_id'], self._sequence, now,
+                            method, operation, _json(shape), _digest(shape), discovery, catalog_state,
+                            request_bytes, scope))
+            if not self._transaction(write):
+                return None
+            self.health['recorded_starts'] += 1
+            self.health['pruned_interactions'] += pruned
+            self._active[observation.identity] = observation
+            return observation
+        except Exception as error:
+            self._degrade(error)
+            return None
+
+    def finish(self, observation: Observation | None, response: object = None,
+               *, response_bytes: int | None = None, incomplete: bool = False) -> None:
+        if observation is None:
+            return
+        self._active.pop(observation.identity, None)
+        try:
+            phase, outcome, feedback = 'end', 'response', []
+            if incomplete:
+                phase, outcome = 'incomplete', 'unknown'
+            elif isinstance(response, dict):
+                result = response.get('result', {})
+                if 'error' in response:
+                    phase = 'error'
+                    code = response['error'].get('code')
+                    outcome = {-32700: 'parse-error', -32600: 'invalid-envelope',
+                               -32601: 'unsupported-method', -32602: 'invalid-parameters',
+                               -32000: 'not-ready'}.get(code, 'protocol-error')
+                elif isinstance(result, dict):
+                    value = result.get('structuredContent', {})
+                    if result.get('isError'):
+                        phase, outcome = 'error', 'tool-exception' if not value else 'rejected'
+                    if isinstance(value, dict):
+                        if isinstance(value.get('outcome'), str) and value['outcome'] in OUTCOMES:
+                            outcome = value['outcome']
+                        elif isinstance(value.get('status'), str) and value['status'] in OUTCOMES:
+                            outcome = value['status']
+                        if observation.operation in self._shapes:
+                            feedback = self._shapes[observation.operation].feedback(value.get('input_feedback'))
+                            if 'input_feedback' in value:
+                                phase, outcome = 'error', 'input-invalid'
+            evicted = False
+            def write(db):
+                nonlocal evicted
+                # Retention in another process may have evicted this start. Do not
+                # recreate it or claim a terminal event was retained.
+                if not db.execute('SELECT 1 FROM interactions WHERE id=?', (observation.identity,)).fetchone():
+                    evicted = True
+                    return
+                db.execute('INSERT INTO events VALUES(?,?,?,?,?,?,?)',
+                           (observation.identity, phase, time.time(),
+                            max(0, (time.monotonic() - observation.started) * 1000),
+                            response_bytes, outcome, _json(feedback)))
+            if self._transaction(write):
+                if evicted:
+                    self.health['dropped_observations'] += 1
+                    self._emit()
+                else:
+                    self.health['recorded_terminals'] += 1
+        except Exception as error:
+            self._degrade(error)
+
+    def close(self) -> None:
+        if self._closed:
+            return
+        for observation in tuple(self._active.values()):
+            self.finish(observation, incomplete=True)
+        self._closed = True
+        if self._connection is not None:
+            try:
+                self._connection.close()
+            except Exception as error:
+                self._degrade(error)
+            self._connection = None
+        self._emit()
+
+
+def open_report(database: Path) -> sqlite3.Connection:
+    """Operator-only, read-only connection. Never creates/migrates a database."""
+    db = sqlite3.connect(database.resolve(strict=True).as_uri() + '?mode=ro', uri=True,
+                         timeout=BUSY_TIMEOUT_MS / 1000, isolation_level=None)
+    try:
+        if (db.execute('PRAGMA application_id').fetchone()[0] != APPLICATION_ID
+                or db.execute('PRAGMA user_version').fetchone()[0] != SCHEMA_VERSION):
+            raise StorageRejected('unsupported-database-schema')
+        db.execute('PRAGMA query_only=ON')
+        return db
+    except BaseException:
+        db.close()
+        raise
+
+
+def report(database: Path, *, limit: int = 100) -> dict:
+    if not 1 <= limit <= MAX_REPORT_ROWS:
+        raise ValueError('report limit must be in 1..1000')
+    db = open_report(database)
+    db.row_factory = sqlite3.Row
+    try:
+        db.execute('BEGIN')  # One consistent read snapshot, including denominators.
+        total = db.execute('SELECT count(*) FROM interactions').fetchone()[0]
+        terminals = db.execute('SELECT count(*) FROM events').fetchone()[0]
+        operations = [dict(row) for row in db.execute(
+            'SELECT i.scope,i.method,i.operation,count(*) AS starts,count(e.interaction) AS terminals,'
+            'sum(CASE WHEN e.phase=\'error\' THEN 1 ELSE 0 END) AS errors,'
+            'sum(CASE WHEN e.interaction IS NULL OR e.phase=\'incomplete\' THEN 1 ELSE 0 END) AS incomplete,'
+            'sum(i.request_bytes) AS request_bytes,count(i.request_bytes) AS request_measurements,'
+            'sum(e.response_bytes) AS response_bytes,count(e.response_bytes) AS response_measurements,'
+            'count(e.duration_ms) AS duration_measurements,'
+            'avg(e.duration_ms) AS observed_duration_ms FROM interactions i LEFT JOIN events e '
+            'ON e.interaction=i.id GROUP BY i.scope,i.method,i.operation ORDER BY starts DESC,i.scope,i.method,i.operation LIMIT ?', (limit + 1,))]
+        operations_truncated = len(operations) > limit
+        operations = operations[:limit]
+        failures = [dict(row) for row in db.execute(
+            'SELECT i.scope,i.operation,i.shape_digest,e.outcome,e.feedback,count(*) AS count FROM interactions i '
+            'JOIN events e ON e.interaction=i.id WHERE e.phase=\'error\' '
+            'GROUP BY i.scope,i.operation,i.shape_digest,e.outcome,e.feedback ORDER BY count DESC,i.scope,i.operation,i.shape_digest,e.outcome,e.feedback LIMIT ?', (limit + 1,))]
+        failures_truncated = len(failures) > limit
+        failures = failures[:limit]
+        shapes = [dict(row) for row in db.execute(
+            'SELECT i.scope,i.operation,i.shape_digest,i.input_shape,count(*) AS starts,'
+            'sum(CASE WHEN e.phase=\'error\' THEN 1 ELSE 0 END) AS errors FROM interactions i '
+            'LEFT JOIN events e ON e.interaction=i.id WHERE i.method=\'tools/call\' '
+            'GROUP BY i.scope,i.operation,i.shape_digest,i.input_shape '
+            'ORDER BY starts DESC,i.scope,i.operation,i.shape_digest LIMIT ?', (limit + 1,))]
+        shapes_truncated = len(shapes) > limit
+        shapes = shapes[:limit]
+        for shape in shapes:
+            shape['input_shape'] = json.loads(shape['input_shape'])
+        for failure in failures:
+            failure['feedback'] = json.loads(failure['feedback'])
+        sessions = [dict(row) for row in db.execute(
+            'SELECT * FROM (SELECT * FROM sessions ORDER BY started DESC,id DESC LIMIT ?) ORDER BY started,id', (limit,))]
+        for session in sessions:
+            session['runtime'] = json.loads(session['runtime'])
+            session['catalog'] = json.loads(session['catalog'])
+        # Chronology within a process only: these are observed attempts, not proof
+        # of agent intent, a repair relation, or an acquired definition.
+        pending = Counter()
+        discoveries = Counter()
+        success_counts = Counter()
+        failure_histograms = {}
+        discovery_histograms = {}
+        scanned = 0
+        for row in db.execute(
+            'SELECT * FROM (SELECT i.session,i.ordinal,i.scope,i.operation,i.discovery_operation,e.phase '
+            'FROM interactions i LEFT JOIN events e ON e.interaction=i.id '
+            'ORDER BY i.started DESC,i.id DESC LIMIT ?) ORDER BY session,ordinal', (MAX_SEQUENCE_SCAN,)):
+            scanned += 1
+            scope = row['scope']
+            key = (row['session'], scope, row['operation'])
+            if row['discovery_operation'] and row['phase'] == 'end':
+                discoveries[(row['session'], scope, row['discovery_operation'])] += 1
+            elif row['operation'] not in {'other', 'describe_input', 'describe_output'}:
+                if row['phase'] == 'error':
+                    pending[key] += 1
+                elif row['phase'] == 'end':
+                    success_counts[scope] += 1
+                    failure_histograms.setdefault(scope, Counter())[pending.pop(key, 0)] += 1
+                    discovery_histograms.setdefault(scope, Counter())[discoveries.pop(key, 0)] += 1
+        sequences = []
+        for scope, successes in sorted(success_counts.items()):
+            errors = failure_histograms[scope]
+            discovery = discovery_histograms[scope]
+            sequences.append({'scope': scope, 'successful_responses': successes,
+                              'preceding_errors_histogram': dict(sorted(errors.items())[:limit]),
+                              'preceding_discoveries_histogram': dict(sorted(discovery.items())[:limit]),
+                              'histograms_truncated': len(errors) > limit or len(discovery) > limit})
+        revisions = {}
+        for session in sessions:
+            revisions.setdefault(session['runtime']['catalog_digest'], session['catalog'])
+        comparisons = []
+        digests = list(revisions)
+        for before, after in zip(digests, digests[1:]):
+            left, right = revisions[before], revisions[after]
+            comparisons.append({'before': before, 'after': after,
+                                'added': sorted(right.keys() - left.keys()),
+                                'removed': sorted(left.keys() - right.keys()),
+                                'changed': sorted(name for name in left.keys() & right.keys() if left[name] != right[name])})
+        return {'schema_version': SCHEMA_VERSION, 'retained_starts': total,
+                'retained_terminals': terminals,
+                'unmatched_starts': total - terminals,
+                'operations': operations, 'operations_truncated': operations_truncated,
+                'failures': failures, 'failures_truncated': failures_truncated, 'shapes': shapes,
+                'shapes_truncated': shapes_truncated, 'sessions': sessions,
+                'session_rows_truncated': db.execute('SELECT count(*) FROM sessions').fetchone()[0] > limit,
+                'catalog_comparisons': comparisons,
+                'sequences': sequences, 'sequence_scan_starts': scanned,
+                'sequence_scan_truncated': total > scanned,
+                'limits': {'group_rows': limit, 'sequence_scan_limit': MAX_SEQUENCE_SCAN,
+                           'meaning': 'Retained observations only; unmatched or incomplete outcomes are unknown. '
+                           'End means response produced/delivered per scope, not domain completion. '
+                           'Sequence denominator is end responses per session/scope/operation in the bounded scan. No cross-process '
+                           'continuation or intent is inferred. Retention can remove prior attempts. '
+                           'Bytes are serialized JSON/stdio bytes, never tokens or billing. '
+                           'Redacted requests cannot be replayed; use independent qualification evidence.'}}
+    finally:
+        db.close()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description='Read-only operator MCP observation report; no agent tools.')
+    parser.add_argument('--database', type=Path, required=True)
+    parser.add_argument('--limit', type=int, default=100)
+    arguments = parser.parse_args()
+    try:
+        print(json.dumps(report(arguments.database, limit=arguments.limit), sort_keys=True, indent=2))
+    except (OSError, sqlite3.Error, StorageRejected, ValueError):
+        print('Observation report unavailable; check database access and schema.', file=sys.stderr)
+        return 1
+    return 0
+
+
+if __name__ == '__main__':
+    raise SystemExit(main())
diff --git a/tools/standards_engine/standards_engine/mcp.py b/tools/standards_engine/standards_engine/mcp.py
index 25c344bd..cf256425 100644
--- a/tools/standards_engine/standards_engine/mcp.py
+++ b/tools/standards_engine/standards_engine/mcp.py
@@ -13,7 +13,7 @@
 from pathlib import Path
 import sys
 import traceback
-from typing import TextIO
+from typing import Callable, TextIO
 
 from .tools import AgentToolFacade
 from .compiled_cache import CompiledSnapshotCache
@@ -21,6 +21,10 @@
 from .contract_discovery import ContractDiscovery
 from .context_projection import Purpose
 from .mcp_catalog import OutputSchemaDelivery, tool_catalog
+from .interaction_observer import (
+    DEFAULT_MAX_AGE_SECONDS, DEFAULT_MAX_BYTES, DEFAULT_MAX_CALLS,
+    InteractionRecorder, RecorderConfig, serialized_bytes,
+)
 
 
 PROTOCOL_VERSION = "2025-11-25"
@@ -35,7 +39,8 @@ def __init__(self, code: int, message: str, data: dict | None = None) -> None:
 
 class MCPServer:
     def __init__(self, root: Path, *, purpose: Purpose | str, advanced: bool = False,
-                 output_schemas: OutputSchemaDelivery | str = OutputSchemaDelivery.EAGER) -> None:
+                 output_schemas: OutputSchemaDelivery | str = OutputSchemaDelivery.EAGER,
+                 observation: RecorderConfig | None = None) -> None:
         self.root = root.resolve()
         self._purpose = Purpose(purpose)
         self.advanced = advanced
@@ -52,6 +57,8 @@ def __init__(self, root: Path, *, purpose: Purpose | str, advanced: bool = False
             catalog_digest=self._runtime_identity.metadata()["catalog_digest"],
         )
         self._compiled_cache = CompiledSnapshotCache(self.root, self.purpose)
+        self._observer = (InteractionRecorder(observation, self._runtime_identity.metadata(), self.tools)
+                          if observation is not None else None)
         self._closed = False
         self.initialized = False
         self.ready = False
@@ -62,6 +69,8 @@ def purpose(self) -> Purpose:
 
     def close(self) -> None:
         """Release only process-owned pure resources; each call closes its store."""
+        if self._observer is not None:
+            self._observer.close()
         self._compiled_cache.close()
         self._interface = None
         self._runtime_identity = None
@@ -71,6 +80,44 @@ def close(self) -> None:
         self._closed = True
 
     def dispatch(self, message: object) -> dict | None:
+        return self._handle(message)
+
+    def _handle(self, message: object, *, deliver: Callable[[dict], int | None] | None = None,
+                request_bytes: int | None = None, parse_error: bool = False) -> dict | None:
+        observer = self._observer
+        ticket = None
+        if observer is not None:
+            # Measurement failures are diagnostics failures, never protocol failures.
+            if request_bytes is None and deliver is None:
+                try:
+                    request_bytes = serialized_bytes(message)
+                except Exception:
+                    pass
+            ticket = observer.begin(message, request_bytes=request_bytes,
+                                    scope='stdio-flushed' if deliver else 'constructed',
+                                    parse_error=parse_error)
+        try:
+            response = ({'jsonrpc': '2.0', 'id': None,
+                         'error': {'code': -32700, 'message': 'Invalid JSON.'}}
+                        if parse_error else self._dispatch(message))
+            response_bytes = None
+            if response is not None:
+                if deliver is not None:
+                    response_bytes = deliver(response)
+                elif observer is not None:
+                    try:
+                        response_bytes = serialized_bytes(response)
+                    except Exception:
+                        pass
+            if observer is not None:
+                observer.finish(ticket, response, response_bytes=response_bytes)
+            return response
+        except BaseException:
+            if observer is not None:
+                observer.finish(ticket, incomplete=True)
+            raise
+
+    def _dispatch(self, message: object) -> dict | None:
         identifier = None
         try:
             if not isinstance(message, dict):
@@ -208,21 +255,27 @@ def _tool_result(self, value: dict) -> dict:
 
 
 def serve(server: MCPServer, source: TextIO, destination: TextIO) -> None:
+    def deliver(response: dict) -> int | None:
+        serialized = json.dumps(response) + "\n"
+        destination.write(serialized)
+        destination.flush()
+        return len(serialized.encode('utf-8')) if server._observer is not None else None
+
     try:
         for line in source:
             try:
                 message = json.loads(line)
+                parse_error = False
             except (ValueError, RecursionError):
-                response = {
-                    "jsonrpc": "2.0",
-                    "id": None,
-                    "error": {"code": -32700, "message": "Invalid JSON."},
-                }
-            else:
-                response = server.dispatch(message)
-            if response is not None:
-                destination.write(json.dumps(response) + "\n")
-                destination.flush()
+                message, parse_error = None, True
+            request_bytes = None
+            if server._observer is not None:
+                try:
+                    request_bytes = len(line.encode('utf-8'))
+                except UnicodeError:
+                    pass
+            server._handle(message, deliver=deliver, request_bytes=request_bytes,
+                           parse_error=parse_error)
     finally:
         server.close()
 
@@ -243,10 +296,26 @@ def main() -> int:
         default=OutputSchemaDelivery.EAGER.value,
         help="Advertise complete output schemas (eager, default), or retrieve exact contracts through describe_output (on-demand).",
     )
+    parser.add_argument('--observe-database', type=Path,
+                        help='Opt in to operator-only local SQLite observations in a trusted directory.')
+    parser.add_argument('--observe-deployment', help='Required nonsensitive deployment label when recording.')
+    parser.add_argument('--observe-max-calls', type=int, default=DEFAULT_MAX_CALLS)
+    parser.add_argument('--observe-max-age-seconds', type=int, default=DEFAULT_MAX_AGE_SECONDS)
+    parser.add_argument('--observe-max-bytes', type=int, default=DEFAULT_MAX_BYTES)
     arguments = parser.parse_args()
+    if (arguments.observe_database is None) != (arguments.observe_deployment is None):
+        parser.error('--observe-database and --observe-deployment must be supplied together')
+    observation = None
+    if arguments.observe_database is not None:
+        try:
+            observation = RecorderConfig(arguments.observe_database, arguments.observe_deployment,
+                                         arguments.observe_max_calls, arguments.observe_max_age_seconds,
+                                         arguments.observe_max_bytes)
+        except ValueError as error:
+            parser.error(str(error))
     serve(
         MCPServer(arguments.repo_root, purpose=arguments.purpose, advanced=arguments.advanced,
-                  output_schemas=arguments.output_schemas),
+                  output_schemas=arguments.output_schemas, observation=observation),
         sys.stdin,
         sys.stdout,
     )
diff --git a/tools/standards_engine/tests/test_analysis.py b/tools/standards_engine/tests/test_analysis.py
index 237f5f8b..0eeaf032 100644
--- a/tools/standards_engine/tests/test_analysis.py
+++ b/tools/standards_engine/tests/test_analysis.py
@@ -2189,6 +2189,11 @@ def quarantine(request):
         self.assertIsInstance(self.engine.resolve(submission), (PendingResult, CompleteResult))
 
     def test_proposal_analysis_uses_one_projection_and_one_verified_base(self) -> None:
+        accepted = self.engine._compiled_snapshot(
+            self.engine._snapshot_id(self.snapshot)
+        )
+        policy = accepted.corpus.resolve_policy_unit(POLICY)
+        self.assertIsNotNone(policy)
         capture = self.engine._snapshots.load_content(self.engine._snapshot_id(self.snapshot))
         planning = next(item.content.decode("utf-8") for item in capture.files
                         if str(item.path) == "workflows/planning.md")
@@ -2196,7 +2201,8 @@ def test_proposal_analysis_uses_one_projection_and_one_verified_base(self) -> No
         created = self.engine.create_proposal(CreateProposalCall.from_value({
             "kind": "create-proposal", "base_snapshot": self.snapshot.as_contract(),
             "change_set": _policy_change_set(policy=POLICY, title=WRITTEN_PLAN_TITLE,
-                body=body, accepted_revision=1, proposed_revision=2,
+                body=body, accepted_revision=policy.semantic_revision,
+                proposed_revision=policy.semantic_revision + 1,
                 purpose="Exercise exact compiled proposal inputs."),
         }))
         self.assertIsInstance(created, CreateProposalResult)
diff --git a/tools/standards_engine/tests/test_interaction_observer.py b/tools/standards_engine/tests/test_interaction_observer.py
new file mode 100644
index 00000000..72cd0f80
--- /dev/null
+++ b/tools/standards_engine/tests/test_interaction_observer.py
@@ -0,0 +1,547 @@
+"""Synthetic operator evidence; never records a real agent/model session."""
+from __future__ import annotations
+
+from contextlib import redirect_stderr
+import io
+import json
+import os
+from pathlib import Path
+import sqlite3
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from unittest.mock import patch
+
+from tools.standards_engine.standards_engine.interaction_observer import (
+    APPLICATION_ID, InputShape, InteractionRecorder, RecorderConfig, SCHEMA_VERSION,
+    report,
+)
+from tools.standards_engine.standards_engine.mcp import MCPServer, serve
+from tools.standards_engine.standards_engine.tools import AgentToolFacade
+from tools.standards_engine.tests.test_mcp import ROOT, initialize, request
+
+MODULE = 'tools.standards_engine.standards_engine.interaction_observer'
+MCP_MODULE = 'tools.standards_engine.standards_engine.mcp'
+
+
+class ObserverTests(unittest.TestCase):
+    def setUp(self):
+        self.temporary = tempfile.TemporaryDirectory()
+        self.addCleanup(self.temporary.cleanup)
+        self.database = Path(self.temporary.name) / 'observations.sqlite'
+        self.diagnostics = io.StringIO()
+
+    def server(self, **options):
+        with redirect_stderr(self.diagnostics):
+            server = MCPServer(ROOT, purpose='authoring',
+                               observation=RecorderConfig(self.database, 'synthetic', **options))
+        self.addCleanup(server.close)
+        return server
+
+    def rows(self, sql):
+        with sqlite3.connect(self.database) as db:
+            db.row_factory = sqlite3.Row
+            return [dict(row) for row in db.execute(sql)]
+
+    def test_opt_in_does_not_change_catalog_or_results(self):
+        plain = MCPServer(ROOT, purpose='authoring')
+        self.addCleanup(plain.close)
+        observed = self.server()
+        self.assertEqual(plain.tools, observed.tools)
+        for server in (plain, observed):
+            initialize(server)
+        # A fixed process identity makes these full public responses comparable.
+        observed._runtime_identity._instance = plain._runtime_identity._instance
+        for message in (request('tools/list'), request('tools/call', {'name': 'runtime_info'}),
+                        request('tools/call', {'name': 'describe_input', 'arguments': {'operation': 'apply'}}),
+                        request('tools/call', {'name': 'runtime_info', 'arguments': {'secret': 'private'}}),
+                        request('tools/call', {'name': 'missing-operation'})):
+            self.assertEqual(plain.dispatch(message), observed.dispatch(message))
+        result = report(self.database)
+        self.assertEqual(result['retained_starts'], 6)
+        self.assertEqual(result['retained_terminals'], 6)
+        self.assertEqual(result['unmatched_starts'], 0)
+        self.assertEqual(result['sessions'][0]['runtime']['interface_version'], 46)
+        self.assertEqual(result['sessions'][0]['runtime']['catalog_digest'],
+                         plain._runtime_identity.metadata()['catalog_digest'])
+        self.assertNotIn('observer', json.dumps(observed.tools))
+
+    def test_shape_and_feedback_disclose_no_values_or_dynamic_keys(self):
+        schema = {'type': 'object', 'properties': {
+            'body': {'type': 'object', 'properties': {'token': {'type': 'string'}}},
+            'facts': {'type': 'object', 'additionalProperties': {
+                'type': 'object', 'properties': {'body': {'type': 'string'}}}},
+            'items': {'type': 'array', 'items': {'type': 'object', 'properties': {
+                'selector': {'type': 'string'}}}},
+        }}
+        shape = InputShape(schema)
+        value = {'body': {'token': 'CREDENTIAL'}, 'facts': {'facts': {'body': 'RAW-STANDARDS'}},
+                 'items': [{'selector': 'PROMPT'}] * 100, 'SECRET-KEY': 'RAW-REPO'}
+        encoded = json.dumps(shape.project(value))
+        for secret in ('CREDENTIAL', 'RAW-STANDARDS', 'PROMPT', 'SECRET-KEY', 'RAW-REPO'):
+            self.assertNotIn(secret, encoded)
+        self.assertEqual(shape.project(value)['fields']['body'], 'redacted')
+        self.assertEqual(shape.project(value)['fields']['facts']['fields'], {})
+        feedback = shape.feedback({'issues': [
+            {'instance_pointer': '/facts/facts/body', 'keyword': 'type', 'message': 'CREDENTIAL'},
+            {'instance_pointer': '/items/8/selector', 'keyword': 'required'},
+            {'instance_pointer': '/body/token', 'keyword': 'CREDENTIAL'},
+        ]})
+        self.assertEqual(feedback, [{'pointer': '/facts/*/*', 'keyword': 'type'},
+                                    {'pointer': '/items/*/selector', 'keyword': 'required'},
+                                    {'pointer': '/body/*', 'keyword': 'other'}])
+        # Deep/large adversarial data is bounded without traversing unknown maps.
+        projected = shape.project({'items': [{'selector': 'x' * 1_000_000}] * 100_000})
+        self.assertLess(len(json.dumps(projected)), 8192)
+
+    def test_privacy_at_real_dispatch_and_disk_boundary(self):
+        server = self.server()
+        initialize(server)
+        secrets = ['SYNTHETIC_PASSWORD_847', 'SYNTHETIC_PROMPT_738', 'SYNTHETIC_REPO_339']
+        server.dispatch(request('tools/call', {'name': 'runtime_info', 'arguments': {
+            'expected_catalog': secrets[0], secrets[1]: secrets[2]}}, identifier=secrets[0]))
+        server.dispatch(request('tools/call', {'name': secrets[1], 'arguments': {'body': secrets[2]}}))
+        server.dispatch(request('initialize', {'clientInfo': {'name': secrets[0]}}))
+        server.close()
+        for path in self.database.parent.iterdir():
+            contents = path.read_bytes()
+            for secret in secrets:
+                self.assertNotIn(secret.encode(), contents)
+        summary = json.dumps(report(self.database))
+        for secret in secrets:
+            self.assertNotIn(secret, summary)
+        failures = self.rows("SELECT outcome,feedback FROM events WHERE phase='error'")
+        self.assertIn('input-invalid', {row['outcome'] for row in failures})
+        self.assertTrue(any('expected_catalog' in row['feedback'] for row in failures))
+
+    def test_malformed_envelopes_and_parse_errors_do_not_disable_recorder(self):
+        server = self.server()
+        for value in (None, [], {}, {'method': 'ping'}, {'id': 1, 'method': [], 'jsonrpc': '2.0'},
+                      request('tools/call', {'name': []}), request('ping', [])):
+            server.dispatch(value)
+        output = io.StringIO()
+        serve(server, io.StringIO('{BAD JSON\n'), output)
+        self.assertEqual(json.loads(output.getvalue())['error']['code'], -32700)
+        self.assertEqual(server._observer.health['state'], 'healthy')
+        self.assertEqual(report(self.database)['retained_starts'], 8)
+        self.assertIn('parse-error', {row['outcome'] for row in self.rows('SELECT outcome FROM events')})
+
+    def test_tool_exception_interruption_and_delivery_failure(self):
+        server = self.server()
+        initialize(server)
+        with patch.object(AgentToolFacade, 'open_repository') as opened, redirect_stderr(io.StringIO()):
+            opened.return_value.__enter__.side_effect = RuntimeError('SENSITIVE_EXCEPTION')
+            response = server.dispatch(request('tools/call', {'name': 'apply', 'arguments': {}}))
+        self.assertTrue(response['result']['isError'])
+        with patch.object(server, '_request', side_effect=KeyboardInterrupt('SENSITIVE_INTERRUPT')):
+            with self.assertRaises(KeyboardInterrupt):
+                server.dispatch(request('ping'))
+        def broken(response):
+            raise BrokenPipeError('SENSITIVE_PIPE')
+        with self.assertRaises(BrokenPipeError):
+            server._handle(request('ping'), deliver=broken, request_bytes=22)
+        events = self.rows('SELECT phase,outcome,response_bytes FROM events')
+        self.assertEqual([r['outcome'] for r in events], ['response', 'tool-exception', 'unknown', 'unknown'])
+        self.assertEqual(sum(r['phase'] == 'incomplete' for r in events), 2)
+        self.assertNotIn('SENSITIVE_', self.database.read_bytes().decode(errors='ignore'))
+
+    def test_unavailable_locked_and_failed_health_sink_preserve_responses(self):
+        unavailable = Path(self.temporary.name) / 'missing' / 'db'
+        with redirect_stderr(io.StringIO()):
+            server = MCPServer(ROOT, purpose='authoring', observation=RecorderConfig(unavailable, 'synthetic'))
+        self.addCleanup(server.close)
+        self.assertEqual(server.dispatch(request('ping'))['result'], {})
+        self.assertEqual(server._observer.health['state'], 'degraded')
+        observed = self.server()
+        with sqlite3.connect(self.database, isolation_level=None) as lock:
+            lock.execute('BEGIN IMMEDIATE')
+            before = time.monotonic()
+            response = observed.dispatch(request('ping'))
+            elapsed = time.monotonic() - before
+            lock.execute('ROLLBACK')
+        self.assertEqual(response['result'], {})
+        self.assertEqual(observed._observer.health['reason'], 'storage-locked')
+        self.assertLess(elapsed, 1, '20ms SQLite busy policy must not wait indefinitely')
+        class BrokenDiagnostics:
+            def write(self, value):
+                raise OSError('private')
+            def flush(self):
+                raise OSError('private')
+        recorder = InteractionRecorder(RecorderConfig(unavailable, 'synthetic'), {}, [], diagnostics=BrokenDiagnostics())
+        recorder.close()
+        self.assertEqual(recorder.health['state'], 'degraded')
+
+    def test_full_storage_after_domain_result_never_repeats_work(self):
+        server = self.server()
+        initialize(server)
+        original = server._request
+        calls = []
+        def domain(method, params):
+            calls.append(method)
+            result = original(method, params)
+            # Fill remaining permitted SQLite pages after the begin commit. This
+            # reaches real SQLITE_FULL while inserting the terminal evidence.
+            db = server._observer._connection
+            db.execute('UPDATE events SET feedback=?', ('x' * 3950,))
+            pages = db.execute('PRAGMA page_count').fetchone()[0]
+            db.execute(f'PRAGMA max_page_count={pages}')
+            return result
+        with patch.object(server, '_request', side_effect=domain):
+            response = server.dispatch(request('ping'))
+        self.assertEqual(response['result'], {})
+        self.assertEqual(calls, ['ping'])
+        self.assertEqual(server._observer.health['reason'], 'storage-full')
+        self.assertEqual(report(self.database)['unmatched_starts'], 1)
+
+    def test_migration_reopen_unsupported_corrupt_and_spoofed(self):
+        server = self.server()
+        server.dispatch(request('ping'))
+        server.close()
+        with sqlite3.connect(self.database) as db:
+            self.assertEqual(db.execute('PRAGMA user_version').fetchone()[0], SCHEMA_VERSION)
+            self.assertEqual(db.execute('PRAGMA application_id').fetchone()[0], APPLICATION_ID)
+        again = self.server()
+        again.dispatch(request('ping'))
+        again.close()
+        self.assertEqual(report(self.database)['retained_starts'], 2)
+        for name, setup in (
+            ('future', 'PRAGMA user_version=999'),
+            ('unrelated', 'CREATE TABLE business (secret TEXT)'),
+            ('spoofed', f'PRAGMA user_version=1; PRAGMA application_id={APPLICATION_ID}; CREATE TABLE sessions(id TEXT)'),
+        ):
+            target = Path(self.temporary.name) / name
+            with sqlite3.connect(target) as db:
+                db.executescript(setup)
+            target.chmod(0o600)
+            before = target.read_bytes()
+            with redirect_stderr(io.StringIO()):
+                recorder = InteractionRecorder(RecorderConfig(target, 'synthetic'), {}, [])
+            self.assertEqual(recorder.health['state'], 'degraded')
+            self.assertEqual(target.read_bytes(), before)
+        corrupt = Path(self.temporary.name) / 'corrupt'
+        corrupt.write_bytes(b'not sqlite sensitive synthetic')
+        corrupt.chmod(0o600)
+        before = corrupt.read_bytes()
+        recorder = InteractionRecorder(RecorderConfig(corrupt, 'synthetic'), {}, [], diagnostics=io.StringIO())
+        self.assertEqual(recorder.health['state'], 'degraded')
+        self.assertEqual(corrupt.read_bytes(), before)
+
+    def test_unsupported_journal_and_page_budget_never_publish_migration(self):
+        for name, pragma in (('wal', 'PRAGMA journal_mode=WAL'),):
+            path = Path(self.temporary.name) / name
+            with sqlite3.connect(path) as db:
+                db.execute(pragma)
+            path.chmod(0o600)
+            before = path.read_bytes()
+            recorder = InteractionRecorder(RecorderConfig(path, 'synthetic', max_bytes=1024*1024),
+                                           {}, [], diagnostics=io.StringIO())
+            self.assertEqual(recorder.health['state'], 'degraded')
+            self.assertEqual(path.read_bytes(), before)
+            with sqlite3.connect(path) as db:
+                self.assertEqual(db.execute('PRAGMA user_version').fetchone()[0], 0)
+                self.assertEqual(db.execute('SELECT name FROM sqlite_master').fetchall(), [])
+
+    def test_dropped_private_content_cannot_be_admitted_as_an_empty_recorder(self):
+        secret = 'SYNTHETIC_PRIOR_STORE_SECRET'
+        with sqlite3.connect(self.database) as db:
+            db.execute('PRAGMA secure_delete=OFF')
+            db.execute('CREATE TABLE prior_private(payload TEXT)')
+            db.execute('INSERT INTO prior_private VALUES(?)', (secret,))
+            db.execute('DROP TABLE prior_private')
+        self.database.chmod(0o600)
+        before = self.database.read_bytes()
+        self.assertIn(secret.encode(), before)
+        recorder = InteractionRecorder(RecorderConfig(self.database, 'synthetic'), {}, [], diagnostics=io.StringIO())
+        self.assertEqual(recorder.health['reason'], 'nonpristine-database')
+        self.assertEqual(self.database.read_bytes(), before)
+        with sqlite3.connect(self.database) as db:
+            self.assertEqual(db.execute('PRAGMA user_version').fetchone()[0], 0)
+            self.assertEqual(db.execute('SELECT name FROM sqlite_master').fetchall(), [])
+
+    def test_oversized_empty_store_is_rejected_before_migration(self):
+        with sqlite3.connect(self.database) as db:
+            db.execute('CREATE TABLE staging(payload BLOB)')
+            db.execute('INSERT INTO staging VALUES(zeroblob(2097152))')
+            db.execute('DROP TABLE staging')
+        self.database.chmod(0o600)
+        before = self.database.read_bytes()
+        recorder = InteractionRecorder(RecorderConfig(self.database, 'synthetic', max_bytes=1048576),
+                                       {}, [], diagnostics=io.StringIO())
+        self.assertEqual(recorder.health['reason'], 'database-over-budget')
+        self.assertEqual(self.database.read_bytes(), before)
+
+    def test_interrupted_initial_migration_reopens_empty_source_transactionally(self):
+        script = """
+import os, sqlite3, sys
+connection=sqlite3.connect(sys.argv[1],isolation_level=None)
+connection.execute('BEGIN IMMEDIATE')
+connection.execute('CREATE TABLE sessions(id TEXT PRIMARY KEY)')
+os._exit(0)
+"""
+        process = subprocess.run([sys.executable, '-c', script, str(self.database)],
+                                 capture_output=True, text=True, timeout=10)
+        self.assertEqual(process.returncode, 0, process.stderr)
+        self.database.chmod(0o600)
+        server = self.server()
+        self.assertEqual(server._observer.health['state'], 'healthy')
+        server.dispatch(request('ping'))
+        self.assertEqual(report(self.database)['retained_terminals'], 1)
+
+    def test_private_regular_file_and_no_symlink_or_hardlink(self):
+        target = Path(self.temporary.name) / 'target'
+        target.write_text('unchanged')
+        target.chmod(0o600)
+        for name in ('symlink', 'hardlink'):
+            path = Path(self.temporary.name) / name
+            path.symlink_to(target) if name == 'symlink' else os.link(target, path)
+            recorder = InteractionRecorder(RecorderConfig(path, 'synthetic'), {}, [], diagnostics=io.StringIO())
+            self.assertEqual(recorder.health['state'], 'degraded')
+        self.assertEqual(target.read_text(), 'unchanged')
+        public = Path(self.temporary.name) / 'public'
+        public.touch()
+        public.chmod(0o644)
+        recorder = InteractionRecorder(RecorderConfig(public, 'synthetic'), {}, [], diagnostics=io.StringIO())
+        self.assertEqual(recorder.health['reason'], 'unsafe-database-file')
+        self.assertEqual(public.stat().st_mode & 0o777, 0o644)
+
+    def test_count_age_and_session_retention_and_read_only_reports(self):
+        # Every process-like recorder has its own session; old empty sessions must
+        # disappear along with their interactions rather than accumulating forever.
+        for _ in range(19):
+            server = self.server(max_calls=16)
+            server.dispatch(request('ping'))
+            server.close()
+        summary = report(self.database, limit=1)
+        self.assertEqual(summary['retained_starts'], 16)
+        self.assertEqual(len(self.rows('SELECT id FROM sessions')), 16)
+        self.assertTrue(summary['session_rows_truncated'])
+        with sqlite3.connect(self.database) as db:
+            db.execute('UPDATE interactions SET started=0')
+        server = self.server(max_calls=16, max_age_seconds=60)
+        server.dispatch(request('ping'))
+        self.assertEqual(report(self.database)['retained_starts'], 1)
+        self.assertEqual(server._observer.health['pruned_interactions'], 16)
+        before = self.database.read_bytes()
+        report(self.database)
+        self.assertEqual(self.database.read_bytes(), before)
+        for limit in (0, 1001):
+            with self.assertRaises(ValueError):
+                report(self.database, limit=limit)
+        missing = Path(self.temporary.name) / 'absent.sqlite'
+        with self.assertRaises(FileNotFoundError):
+            report(missing)
+        self.assertFalse(missing.exists())
+
+    def test_retention_can_evict_inflight_evidence_without_disabling_a_live_process(self):
+        first = self.server(max_calls=16)
+        second = self.server(max_calls=16)
+        ticket = first._observer.begin(request('ping'), request_bytes=50, scope='constructed')
+        for _ in range(16):
+            second.dispatch(request('ping'))
+        first._observer.finish(ticket, {'result': {}}, response_bytes=14)
+        self.assertEqual(first._observer.health['state'], 'healthy')
+        self.assertEqual(first._observer.health['dropped_observations'], 1)
+        first.dispatch(request('ping'))
+        self.assertEqual(first._observer.health['recorded_terminals'], 1)
+        summary = report(self.database)
+        self.assertEqual(summary['retained_starts'], 16)
+        self.assertEqual(summary['retained_terminals'], 16)
+
+    def test_observed_discovery_and_error_sequences_have_explicit_denominators(self):
+        server = self.server()
+        initialize(server)
+        for name, arguments in (('describe_input', {'operation': 'runtime_info'}),
+                                ('runtime_info', {'extra': 'bad'}), ('runtime_info', {}),
+                                ('runtime_info', {})):
+            server.dispatch(request('tools/call', {'name': name, 'arguments': arguments}))
+        summary = report(self.database)
+        self.assertEqual(summary['sequences'][0]['successful_responses'], 2)
+        self.assertEqual(summary['sequences'][0]['preceding_errors_histogram'], {1: 1, 0: 1})
+        self.assertEqual(summary['sequences'][0]['preceding_discoveries_histogram'], {1: 1, 0: 1})
+        self.assertEqual(summary['failures'][0]['outcome'], 'input-invalid')
+
+    def test_catalog_revision_report_attributes_exact_operation_schema_changes(self):
+        def runtime(instance, catalog):
+            return {'instance_id': instance, 'interface_version': 46, 'purpose': 'authoring',
+                    'catalog_digest': catalog, 'schema_digest': 'synthetic-schema',
+                    'implementation_digest': 'synthetic-source'}
+        first_catalog = [{'name': 'shared', 'inputSchema': {'type': 'string'}},
+                         {'name': 'retired', 'inputSchema': {}}]
+        second_catalog = [{'name': 'shared', 'inputSchema': {'type': 'number'}},
+                          {'name': 'added', 'inputSchema': {}}]
+        for instance, digest, catalog in (('first', 'synthetic-catalog-1', first_catalog),
+                                          ('second', 'synthetic-catalog-2', second_catalog)):
+            recorder = InteractionRecorder(RecorderConfig(self.database, 'synthetic'),
+                                           runtime(instance, digest), catalog, diagnostics=io.StringIO())
+            ticket = recorder.begin(request('tools/call', {'name': 'shared', 'arguments': {}}),
+                                    request_bytes=50, scope='constructed')
+            recorder.finish(ticket, {'result': {}}, response_bytes=14)
+            recorder.close()
+        summary = report(self.database)
+        self.assertEqual(summary['catalog_comparisons'], [{
+            'before': 'synthetic-catalog-1', 'after': 'synthetic-catalog-2',
+            'added': ['added'], 'removed': ['retired'], 'changed': ['shared']}])
+        self.assertEqual({row['runtime']['implementation_digest'] for row in summary['sessions']},
+                         {'synthetic-source'})
+
+    def test_report_scope_truncation_and_bounded_scan_are_explicit(self):
+        server = self.server()
+        initialize(server)
+        message = request('tools/call', {'name': 'runtime_info', 'arguments': {}})
+        server.dispatch(message)
+        server._handle(message, deliver=lambda response: 10, request_bytes=10)
+        summary = report(self.database, limit=1)
+        self.assertTrue(summary['operations_truncated'])
+        self.assertTrue(summary['shapes_truncated'])
+        self.assertEqual({row['scope'] for row in summary['sequences']}, {'constructed', 'stdio-flushed'})
+        with patch(MODULE + '.MAX_SEQUENCE_SCAN', 1):
+            bounded = report(self.database)
+        self.assertEqual(bounded['sequence_scan_starts'], 1)
+        self.assertTrue(bounded['sequence_scan_truncated'])
+        self.assertIn('input_shape', summary['shapes'][0])
+
+    def test_arbitrary_outputs_are_never_retained(self):
+        server = self.server()
+        initialize(server)
+        secret = 'SYNTHETIC_OUTPUT_CREDENTIAL'
+        result = {'structuredContent': {'body': secret, 'outcome': secret,
+                  'input_feedback': {'issues': [{'instance_pointer': '/' + secret,
+                                                'keyword': secret, 'message': secret}]}}, 'isError': True}
+        with patch.object(server, '_request', return_value=result):
+            response = server.dispatch(request('tools/call', {'name': 'runtime_info', 'arguments': {}}))
+        self.assertEqual(response['result'], result)
+        server.close()
+        self.assertNotIn(secret.encode(), self.database.read_bytes())
+        self.assertNotIn(secret, json.dumps(report(self.database)))
+
+    def test_unencodable_text_measurement_never_changes_stdio_behavior(self):
+        for observed in (False, True):
+            server = self.server() if observed else MCPServer(ROOT, purpose='authoring')
+            self.addCleanup(server.close)
+            output = io.StringIO()
+            payload = '{"jsonrpc":"2.0","id":"\ud800","method":"ping"}\n'
+            serve(server, io.StringIO(payload), output)
+            self.assertEqual(json.loads(output.getvalue())['result'], {})
+
+
+
+class ProcessTests(unittest.TestCase):
+    def test_cold_stdio_restart_bytes_identity_and_operator_cli(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            database = Path(temporary) / 'observer.sqlite'
+            messages = [request('initialize', {'protocolVersion': '2025-11-25', 'capabilities': {}, 'clientInfo': {}}),
+                        {'jsonrpc': '2.0', 'method': 'notifications/initialized'}, request('tools/list'),
+                        request('tools/call', {'name': 'describe_input', 'arguments': {'operation': 'runtime_info'}}),
+                        request('tools/call', {'name': 'runtime_info', 'arguments': {'unexpected': 'SYNTHETIC_SECRET'}}),
+                        request('tools/call', {'name': 'runtime_info', 'arguments': {}})]
+            payload = ''.join(json.dumps(message) + '\n' for message in messages)
+            command = [sys.executable, '-m', MCP_MODULE, '--repo-root', str(ROOT), '--purpose', 'authoring',
+                       '--output-schemas', 'on-demand', '--observe-database', str(database), '--observe-deployment', 'cold-test']
+            identities = []
+            for _ in range(2):
+                process = subprocess.run(command, cwd=ROOT, input=payload, capture_output=True, text=True, timeout=30)
+                self.assertEqual(process.returncode, 0, process.stderr)
+                responses = [json.loads(line) for line in process.stdout.splitlines()]
+                identity = responses[0]['result']['_meta']['standards-engine/runtime']
+                identities.append(identity)
+                with sqlite3.connect(database) as db:
+                    row = db.execute('SELECT runtime FROM sessions WHERE id=?', (identity['instance_id'],)).fetchone()
+                    self.assertEqual(json.loads(row[0]), identity)
+                    counts = db.execute('SELECT sum(request_bytes) FROM interactions WHERE session=?', (identity['instance_id'],)).fetchone()[0]
+                    notification_bytes = len((json.dumps(messages[1]) + '\n').encode())
+                    self.assertEqual(counts, len(payload.encode()) - notification_bytes)
+                    output_bytes = db.execute('SELECT sum(e.response_bytes) FROM events e JOIN interactions i '
+                                             'ON i.id=e.interaction WHERE session=?', (identity['instance_id'],)).fetchone()[0]
+                    self.assertEqual(output_bytes, len(process.stdout.encode()))
+                self.assertIn('"state":"healthy"', process.stderr)
+            self.assertNotEqual(identities[0]['instance_id'], identities[1]['instance_id'])
+            for key in ('implementation_digest', 'schema_digest', 'catalog_digest', 'interface_version'):
+                self.assertEqual(identities[0][key], identities[1][key])
+            summary = report(database)
+            self.assertEqual(summary['retained_starts'], 10)
+            self.assertEqual(summary['retained_terminals'], 10)
+            self.assertEqual({row['scope'] for row in summary['operations']}, {'stdio-flushed'})
+            before = database.read_bytes()
+            result = subprocess.run([sys.executable, '-m', MODULE, '--database', str(database), '--limit', '2'],
+                                    cwd=ROOT, capture_output=True, text=True, timeout=30)
+            self.assertEqual(result.returncode, 0, result.stderr)
+            self.assertEqual(json.loads(result.stdout)['retained_starts'], 10)
+            self.assertEqual(database.read_bytes(), before)
+
+    def test_process_kill_leaves_unmatched_start_and_replacement_does_not_complete_it(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            database = Path(temporary) / 'observer.sqlite'
+            script = '''
+import sys, threading
+from pathlib import Path
+from tools.standards_engine.standards_engine.mcp import MCPServer, serve
+from tools.standards_engine.standards_engine.interaction_observer import RecorderConfig
+server=MCPServer(Path(sys.argv[1]), purpose='authoring', observation=RecorderConfig(Path(sys.argv[2]), 'kill-test'))
+def blocked(method, params):
+    print('entered-domain', file=sys.stderr, flush=True)
+    threading.Event().wait()
+server._request=blocked
+serve(server,sys.stdin,sys.stdout)
+'''
+            process = subprocess.Popen([sys.executable, '-c', script, str(ROOT), str(database)], cwd=ROOT,
+                                       stdin=subprocess.PIPE, stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True)
+            try:
+                process.stdin.write(json.dumps(request('ping')) + '\n')
+                process.stdin.flush()
+                self.assertIn('healthy', process.stderr.readline())
+                self.assertEqual(process.stderr.readline().strip(), 'entered-domain')
+                process.kill()
+                process.communicate(timeout=10)
+            finally:
+                if process.poll() is None:
+                    process.kill()
+                    process.communicate(timeout=10)
+            replacement = subprocess.run([sys.executable, '-m', MCP_MODULE, '--purpose', 'authoring',
+                                          '--observe-database', str(database), '--observe-deployment', 'replacement'],
+                                         cwd=ROOT, input=json.dumps(request('ping')) + '\n',
+                                         text=True, capture_output=True, timeout=30)
+            self.assertEqual(replacement.returncode, 0, replacement.stderr)
+            summary = report(database)
+            self.assertEqual(summary['retained_starts'], 2)
+            self.assertEqual(summary['retained_terminals'], 1)
+            self.assertEqual(summary['unmatched_starts'], 1)
+
+    def test_concurrent_processes_and_first_open_never_overwrite_or_duplicate(self):
+        with tempfile.TemporaryDirectory() as temporary:
+            database = Path(temporary) / 'observer.sqlite'
+            command = [sys.executable, '-m', MCP_MODULE, '--purpose', 'authoring',
+                       '--observe-database', str(database), '--observe-deployment', 'concurrent']
+            workers = [subprocess.Popen(command, cwd=ROOT, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
+                                        stderr=subprocess.PIPE, text=True) for _ in range(4)]
+            try:
+                outputs = []
+                for worker in workers:
+                    worker.stdin.write(''.join(json.dumps(request('ping', identifier=i)) + '\n' for i in range(5)))
+                    worker.stdin.flush()
+                for worker in workers:
+                    output, errors = worker.communicate(timeout=30)
+                    self.assertEqual(worker.returncode, 0, errors)
+                    self.assertEqual(len(output.splitlines()), 5)
+                    outputs.append(errors)
+            finally:
+                for worker in workers:
+                    if worker.poll() is None:
+                        worker.kill()
+                        worker.communicate(timeout=10)
+            # Contention may intentionally degrade a process after the 20ms budget.
+            # A failed terminal write may leave an unmatched begin; no duplicate or
+            # fabricated terminal is permitted. Domain replies still total 20.
+            summary = report(database)
+            self.assertGreater(summary['retained_starts'], 0)
+            self.assertLessEqual(summary['retained_starts'], 20)
+            self.assertLessEqual(summary['retained_terminals'], summary['retained_starts'])
+            with sqlite3.connect(database) as db:
+                self.assertEqual(db.execute('PRAGMA integrity_check').fetchone()[0], 'ok')
+                self.assertEqual(db.execute('PRAGMA foreign_key_check').fetchall(), [])
+            self.assertTrue(all('healthy' in output or 'degraded' in output for output in outputs))
+
+
+if __name__ == '__main__':
+    unittest.main()