Add Forgejo CI smoke workflow (enablement template)

Honest first-pass maturity vector grounded in README/docs/tests present
in this repo; no invented evidence. Flagged for human review before
publish. See reuse-surface history/2026-07-06-coverage-classification.md.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

.claude/rules/agents.md

@@ -0,0 +1,20 @@
+## Kaizen Agents
+
+Specialized agent personas available on demand via the state-hub MCP.
+
+**Discover:** `list_kaizen_agents()` — returns all agents with name, description, category
+**Load:** `get_kaizen_agent("tdd-workflow")` — returns full instructions; read and follow them
+
+Common agents:
+
+| Agent | Category | When to use |
+|-------|----------|-------------|
+| `tdd-workflow` | testing | Step-by-step TDD8 workflow for any feature |
+| `code-refactoring` | quality | Code quality analysis and safe refactoring |
+| `test-maintenance` | testing | Diagnose and fix failing tests |
+| `requirements-engineering` | process | Prevent interface/mock mismatches upfront |
+| `keepaTodofile` | process | Maintain TODO.md during work |
+| `project-management` | process | Track status, determine next steps |
+| `datamodel-optimization` | quality | Optimize dataclasses and data structures |
+
+All 17 agents: call `list_kaizen_agents()` for the full list.

.claude/rules/architecture.md

@@ -0,0 +1,8 @@
+## Architecture
+
+<!-- TODO: Describe the key design decisions and component structure.
+     Key modules, data flows, external integrations, state machines, etc. -->
+
+## Quick Reference
+
+`~/state-hub/mcp_server/TOOLS.md` — MCP tool reference

.claude/rules/credential-routing.md

@@ -0,0 +1,50 @@
+# Credential and access routing
+
+**Audience:** Codex, Claude Code, Grok, and custodian agents that call **llm-connect**
+for inference. Run this check **before** requesting secrets, API keys, SSH access,
+login tokens, or database passwords — in any repo, not only `ops-warden`.
+
+ops-warden **issues SSH certificates only** (`warden sign`, `cert_command`). Every
+other credential need belongs to another subsystem. **Do not** message
+`ops-warden` on State Hub expecting a secret value; the reply is a pointer, not a key.
+
+### Lookup (do this first)
+
+```bash
+warden route find "<describe your need>" --json
+warden route show <catalog-id> --json
+```
+
+Requires the `warden` CLI from `~/ops-warden` (`uv tool install .` or `uv run warden`).
+
+| Agent runtime | How to orient |
+| --- | --- |
+| **Codex / Grok** (shell, HTTP State Hub) | `warden route` commands above; inbox `to_agent=user-engine` is for coordination, not secret vending |
+| **Claude Code** (MCP when available) | `get_domain_summary("custodian")` for workstreams; **still** use `warden route` for credential ownership |
+| **llm-connect** (inference service) | Never put secret retrieval in prompts; route custody to OpenBao/operator paths surfaced by `warden route` |
+
+### Quick routing table
+
+| I need… | Owner | ops-warden executes? |
+| --- | --- | --- |
+| SSH cert (`adm`/`agt`/`atm`) | ops-warden | **Yes** — `warden sign` |
+| API key, DB password, provider token | OpenBao (`railiance-platform`) | No — route only |
+| Login / OIDC / MFA | key-cape / Keycloak | No — route only |
+| Authorization decision | flex-auth | No — route only |
+| activity-core → issue-core emission | activity-core + issue-core | No — `warden route show activity-core-issue-sink` |
+| SSH tunnel | ops-bridge (+ `cert_command` from warden) | No — route only |
+
+### Anti-patterns (do not do these)
+
+- `POST /messages/` to `ops-warden` asking for `ISSUE_CORE_API_KEY`, `OPENROUTER_API_KEY`, etc.
+- Inventing `warden secret`, `warden login`, `warden bao`, `warden tunnel` — they do not exist
+- Pasting secrets into Git, State Hub, workplans, logs, or chat
+
+### Other capabilities (reuse-surface)
+
+Non-credential capabilities are usually discovered through **reuse-surface** federation
+(`reuse-surface` registry / `capability.*` indexes). Credential routing is inlined in
+every repo's agent instructions because it is high-frequency, high-risk, and easy to
+get wrong.
+
+**Canon:** `~/ops-warden/wiki/CredentialRouting.md` · catalog `~/ops-warden/registry/routing/catalog.yaml`

.claude/rules/first-session.md

@@ -0,0 +1,38 @@
+## First Session Protocol
+
+Triggered when `get_domain_summary("communication")` shows **no workstreams**.
+The project is registered but work has not yet been structured.
+
+**Step 1 — Read, don't write**
+- `~/the-custodian/canon/projects/communication/project_charter_v0.1.md` — purpose, scope
+- `~/the-custodian/canon/projects/communication/roadmap_v0.1.md` — planned phases
+- Scan repo root: README, directory structure, existing code or docs
+
+**Step 2 — Survey in-progress work**
+Look for TODOs, open branches, half-finished files. Note done vs. started but incomplete.
+
+**Step 3 — Propose workstreams to Bernd**
+Propose 1–3 workstreams — each a coherent strand, weeks to months, anchored to a
+roadmap phase. **Wait for approval before creating.**
+
+**Step 4 — Create workplan file first, then DB record (ADR-001)**
+```
+workplans/USER-WP-NNNN-<slug>.md   ← write this first
+```
+Then register in the hub:
+```
+create_workstream(topic_id="a6c6e745-bf54-4465-9340-1534a2be493e", title="...", owner="...", description="...")
+create_task(workstream_id="<id>", title="...", priority="high|medium|low")
+```
+
+**Step 5 — Record the setup**
+```
+add_progress_event(
+    summary="First session: structured communication into N workstreams, M tasks",
+    event_type="milestone",
+    topic_id="a6c6e745-bf54-4465-9340-1534a2be493e",
+    detail={"workstreams": [...], "tasks_created": M}
+)
+```
+
+<!-- Delete or archive this file once past first session -->

.claude/rules/repo-boundary.md

@@ -0,0 +1,8 @@
+## Repo boundary
+
+This repo owns **user-engine** only. It does not own:
+
+<!-- TODO: List what belongs in adjacent repos, e.g.:
+- SSH key management → railiance-infra/
+- State hub code     → state-hub/
+-->

.claude/rules/repo-identity.md

@@ -0,0 +1,5 @@
+**Purpose:** Headless user-domain/profile service for accounts, preferences, memberships, catalogs, projections, audit, and events.
+
+**Domain:** communication
+**Repo slug:** user-engine
+**Topic ID:** a6c6e745-bf54-4465-9340-1534a2be493e

.claude/rules/session-protocol.md

@@ -0,0 +1,85 @@
+## Session Protocol
+
+Dev Hub (State Hub API): http://127.0.0.1:8000
+MCP server name in `~/.claude.json`: `dev-hub`
+
+**Step 1 — Orient**
+
+Read the offline-safe brief first — it works without a live hub connection:
+```bash
+cat .custodian-brief.md
+```
+Then call the MCP tool for richer cross-domain context when MCP tools are exposed:
+```
+get_domain_summary("communication")
+```
+If MCP tools are unavailable in the current agent session, use the REST API:
+```bash
+curl -s "http://127.0.0.1:8000/state/summary" | python3 -m json.tool
+```
+If the hub is offline: `cd ~/state-hub && make api`
+
+**Step 2 — Check inbox**
+With MCP tools:
+```
+get_messages(to_agent="user-engine", unread_only=True)
+```
+Mark read with `mark_message_read(message_id)`. Reply or act on coordination
+requests before proceeding.
+
+Without MCP tools:
+```bash
+curl -s "http://127.0.0.1:8000/messages/?to_agent=user-engine&unread_only=true" \
+  | python3 -m json.tool
+curl -s -X PATCH "http://127.0.0.1:8000/messages/<id>/read" \
+  -H "Content-Type: application/json" -d '{}'
+```
+
+**Step 3 — Scan workplans**
+```bash
+ls workplans/
+```
+For each file with `status: ready`, `active`, or `blocked`, note pending
+`wait`/`todo`/`progress` tasks.
+
+**Step 4 — Present brief**
+
+1. **Active workstreams** for `communication` — title, task counts, blocking decisions
+2. **Pending tasks** from `workplans/` + any `[repo:user-engine]` hub tasks
+3. **Goal guidance** — if `goal_guidance` in summary:
+   - `needs_workplan`: surface as top action — *"Repo goal '{title}' has no workplan yet"*
+   - `alignment_warnings`: flag if active work is not aligned with current goal
+4. **Suggested next action** — highest-priority open item
+5. **SBOM status** — flag if `last_sbom_at` is unset for this repo
+
+If no workstreams: follow First Session Protocol (`first-session.md`).
+
+**During work:** `record_decision()` · `add_progress_event()` · `resolve_decision()`
+
+> State Hub is a *read model*. Bootstrap tools (`create_workstream`, `create_task`)
+> are First Session Protocol only. Work structure belongs in repo files (ADR-001).
+
+**Session close:**
+With MCP tools:
+```
+add_progress_event(summary="...", topic_id="a6c6e745-bf54-4465-9340-1534a2be493e", workstream_id="<uuid>")
+```
+Without MCP tools:
+```bash
+curl -s -X POST http://127.0.0.1:8000/progress/ \
+  -H "Content-Type: application/json" \
+  -d '{"topic_id":"a6c6e745-bf54-4465-9340-1534a2be493e","workstream_id":"<uuid>","event_type":"note","summary":"what changed","author":"codex"}'
+```
+If workplan files were modified, ensure the local copy is up to date first:
+```bash
+git -C <repo_path> pull --ff-only
+cd ~/state-hub && make fix-consistency REPO=user-engine
+```
+For repos where implementation runs on a remote machine (e.g. CoulombCore),
+use the combined target which pulls before fixing:
+```bash
+cd ~/state-hub && make fix-consistency-remote REPO=user-engine
+```
+**C-15** (DB task ahead of file) is normal in multi-machine workflows — writeback
+will sync the file to match DB.  **C-16** (repo behind remote) blocks all writes
+until you pull — intentional to prevent clobbering remote progress.

.claude/rules/stack-and-commands.md

@@ -0,0 +1,19 @@
+## Stack
+
+<!-- TODO: Fill in language, frameworks, and key dependencies -->
+- **Language:**
+- **Key deps:**
+
+## Dev Commands
+
+```bash
+# TODO: Fill in the standard commands for this repo
+
+# Install dependencies
+
+# Run tests
+
+# Lint / type check
+
+# Build / package (if applicable)
+```

.claude/rules/workplan-convention.md

@@ -0,0 +1,40 @@
+## Workplan Convention (ADR-001)
+
+File location: `workplans/USER-WP-NNNN-<slug>.md`
+ID prefix: `USER-WP-`
+
+Work items originate as files in this repo **before** being registered in the hub.
+
+Canonical workplan/workstream frontmatter statuses are:
+`proposed`, `ready`, `active`, `blocked`, `backlog`, `finished`, `archived`.
+Use `proposed` for a newly drafted plan, `ready` after review against current
+repo state, and `finished` when implementation is complete. `stalled` and
+`needs_review` are derived health labels, not stored statuses.
+
+Closed workplans may be moved to `workplans/archived/` with a completion-date
+prefix: `YYMMDD-USER-WP-NNNN-<slug>.md`. The frontmatter id remains
+unchanged; the prefix is only for quick visual reference.
+
+Small opportunistic tasks discovered during another session use **Ad Hoc Tasks**:
+`workplans/ADHOC-YYYY-MM-DD.md`, workstream slug `adhoc-YYYY-MM-DD`, and task ids
+`ADHOC-YYYY-MM-DD-T01`, `T02`, etc. Use adhocs only for low-risk work completed
+directly. Promote anything requiring analysis, design, approval, dependencies, or
+multiple planned phases into a normal workplan.
+
+Ecosystem todos from other agents arrive as `[repo:user-engine]` hub tasks —
+visible at session start. Pick one up by creating the workplan file, then registering
+the workstream.
+
+Task blocks use this shape:
+
+```task
+id: USER-WP-NNNN-T01
+status: wait | todo | progress | done | cancel
+priority: high | medium | low
+state_hub_task_id: "<uuid>"         # written by fix-consistency — do not edit
+```
+
+Status progression is `todo` → `progress` → `done`; use `wait` for waiting or
+blocked work and `cancel` for stopped work.
+
+<!-- Ralph Loop rules and HEUREKA sequence: ~/.claude/CLAUDE.md — do not duplicate here -->

.forgejo/workflows/ci-smoke.yaml

@@ -0,0 +1,29 @@
+# Canonical CI smoke template (tier 1 routing drill).
+# Copy to: .forgejo/workflows/ci-smoke.yaml in consumer repos.
+name: CI Smoke
+
+on:
+  push:
+    branches:
+      - main
+  workflow_dispatch:
+
+jobs:
+  host-smoke:
+    runs-on: self-hosted
+    steps:
+      - name: Routing probe (host runner)
+        run: |
+          set -eu
+          echo "repository=${GITHUB_REPOSITORY:-unknown}"
+          echo "sha=${GITHUB_SHA:-unknown}"
+          echo "runner=${RUNNER_NAME:-unknown}"
+          uname -a
+
+  container-smoke:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Routing probe (container label)
+        run: |
+          set -eu
+          echo "container-smoke ok for ${GITHUB_REPOSITORY:-unknown}"

.repo-classification.yaml

@@ -0,0 +1,16 @@
+repo_classification:
+  standard: Repo Classification Standard
+  version: '1.0'
+  classified_at: '2026-06-22'
+  classified_by: agent
+  category: project
+  domain: communication
+  secondary_domains: []
+  capability_tags: []
+  business_stake:
+  - product
+  - experience
+  - technology
+  business_mechanics:
+  - coordination
+  - operation

AGENTS.md

@@ -0,0 +1,219 @@
+# user-engine — Agent Instructions
+
+## Repo Identity
+
+**Purpose:** Headless user-domain/profile service for accounts, preferences, memberships, catalogs, projections, audit, and events.
+
+**Domain:** communication
+**Repo slug:** user-engine
+**Topic ID:** `a6c6e745-bf54-4465-9340-1534a2be493e`
+**Workplan prefix:** `USER-WP-`
+
+---
+
+## State Hub Integration
+
+The Custodian State Hub tracks work across all domains. Interact via HTTP REST —
+there is no MCP server for Codex agents.
+
+| Context | URL |
+|---------|-----|
+| Local workstation | `http://127.0.0.1:8000` |
+| Remote via tunnel | `http://127.0.0.1:18000` |
+
+### Orient at session start
+
+```bash
+# Offline brief — works without hub connection
+cat .custodian-brief.md
+
+# Active workstreams for this domain
+curl -s "http://127.0.0.1:8000/workstreams/?topic_id=a6c6e745-bf54-4465-9340-1534a2be493e&status=active" \
+  | python3 -m json.tool
+
+# Check inbox
+curl -s "http://127.0.0.1:8000/messages/?to_agent=user-engine&unread_only=true" \
+  | python3 -m json.tool
+```
+
+Mark a message read:
+```bash
+curl -s -X PATCH "http://127.0.0.1:8000/messages/<id>/read" \
+  -H "Content-Type: application/json" -d '{}'
+```
+
+### Log progress (required at session close)
+
+```bash
+curl -s -X POST http://127.0.0.1:8000/progress/ \
+  -H "Content-Type: application/json" \
+  -d '{
+    "summary": "what was done",
+    "event_type": "note",
+    "author": "codex",
+    "workstream_id": "<uuid>",
+    "task_id": "<uuid>"
+  }'
+```
+
+Omit `workstream_id` / `task_id` when not applicable.
+
+### Update task status
+
+```bash
+curl -s -X PATCH "http://127.0.0.1:8000/tasks/<task_id>" \
+  -H "Content-Type: application/json" \
+  -d '{"status": "progress"}'
+# values: wait | todo | progress | done | cancel
+```
+
+### Flag a task for human review
+
+```bash
+curl -s -X PATCH "http://127.0.0.1:8000/tasks/<task_id>" \
+  -H "Content-Type: application/json" \
+  -d '{"needs_human": true, "intervention_note": "reason"}'
+```
+
+---
+
+## Session Protocol
+
+**Start:**
+1. `cat .custodian-brief.md` — domain goal and open workstreams (offline-safe)
+2. Check inbox: `GET /messages/?to_agent=user-engine&unread_only=true`; mark read
+3. Scan workplans: `ls workplans/` — note `status: ready`, `active`, or `blocked` files and open tasks
+4. Check human-needed tasks: `GET /tasks/?needs_human=true`
+
+**During work:**
+- Update task statuses in workplan files as tasks progress
+- Record significant decisions via `POST /decisions/`
+
+**Close:**
+1. Update workplan file task statuses to reflect progress
+2. Log: `POST /progress/` with a summary of what changed
+3. Note for the custodian operator: after workplan file changes, run from
+   `~/state-hub`:
+   ```bash
+   make fix-consistency REPO=user-engine
+   ```
+   This syncs task status from files into the hub DB.
+
+---
+
+## Credential and access routing
+
+**Audience:** Codex, Claude Code, Grok, and custodian agents that call **llm-connect**
+for inference. Run this check **before** requesting secrets, API keys, SSH access,
+login tokens, or database passwords — in any repo, not only `ops-warden`.
+
+ops-warden **issues SSH certificates only** (`warden sign`, `cert_command`). Every
+other credential need belongs to another subsystem. **Do not** message
+`ops-warden` on State Hub expecting a secret value; the reply is a pointer, not a key.
+
+### Lookup (do this first)
+
+```bash
+warden route find "<describe your need>" --json
+warden route show <catalog-id> --json
+```
+
+Requires the `warden` CLI from `~/ops-warden` (`uv tool install .` or `uv run warden`).
+
+| Agent runtime | How to orient |
+| --- | --- |
+| **Codex / Grok** (shell, HTTP State Hub) | `warden route` commands above; inbox `to_agent=user-engine` is for coordination, not secret vending |
+| **Claude Code** (MCP when available) | `get_domain_summary("custodian")` for workstreams; **still** use `warden route` for credential ownership |
+| **llm-connect** (inference service) | Never put secret retrieval in prompts; route custody to OpenBao/operator paths surfaced by `warden route` |
+
+### Quick routing table
+
+| I need… | Owner | ops-warden executes? |
+| --- | --- | --- |
+| SSH cert (`adm`/`agt`/`atm`) | ops-warden | **Yes** — `warden sign` |
+| API key, DB password, provider token | OpenBao (`railiance-platform`) | No — route only |
+| Login / OIDC / MFA | key-cape / Keycloak | No — route only |
+| Authorization decision | flex-auth | No — route only |
+| activity-core → issue-core emission | activity-core + issue-core | No — `warden route show activity-core-issue-sink` |
+| SSH tunnel | ops-bridge (+ `cert_command` from warden) | No — route only |
+
+### Anti-patterns (do not do these)
+
+- `POST /messages/` to `ops-warden` asking for `ISSUE_CORE_API_KEY`, `OPENROUTER_API_KEY`, etc.
+- Inventing `warden secret`, `warden login`, `warden bao`, `warden tunnel` — they do not exist
+- Pasting secrets into Git, State Hub, workplans, logs, or chat
+
+### Other capabilities (reuse-surface)
+
+Non-credential capabilities are usually discovered through **reuse-surface** federation
+(`reuse-surface` registry / `capability.*` indexes). Credential routing is inlined in
+every repo's agent instructions because it is high-frequency, high-risk, and easy to
+get wrong.
+
+**Canon:** `~/ops-warden/wiki/CredentialRouting.md` · catalog `~/ops-warden/registry/routing/catalog.yaml`
+
+<!-- REPO-AGENTS-EXTENSIONS -->
+<!-- Append repo-specific agent instructions below this marker.
+     The state-hub template sync preserves content after this line. -->
+
+---
+
+## Workplan Convention (ADR-001)
+
+Work items originate as files in this repo — not in the hub. The hub is a
+read/cache/index layer that rebuilds from files.
+
+**File location:** `workplans/USER-WP-NNNN-<slug>.md`
+
+**Archived location:** finished workplans may move to
+`workplans/archived/YYMMDD-USER-WP-NNNN-<slug>.md`. The `YYMMDD` prefix is
+the completion/archive date; the frontmatter `id` does not change.
+
+**Ad Hoc Tasks:** small opportunistic fixes discovered during a session use
+`workplans/ADHOC-YYYY-MM-DD.md` with task ids `ADHOC-YYYY-MM-DD-T01`, etc. Use
+this only for low-risk work completed directly; create a normal workplan for
+anything needing analysis, design, approval, dependencies, or multiple phases.
+
+**Frontmatter:**
+
+```yaml
+---
+id: USER-WP-NNNN
+type: workplan
+title: "..."
+domain: communication
+repo: user-engine
+status: proposed | ready | active | blocked | backlog | finished | archived
+owner: codex
+topic_slug: ...
+created: "YYYY-MM-DD"
+updated: "YYYY-MM-DD"
+state_hub_workstream_id: "<uuid>"   # written by fix-consistency — do not edit
+---
+```
+
+Use `proposed` for a new draft, `ready` after review against current repo
+state, and `finished` after implementation. `stalled` and `needs_review` are
+derived health labels, not frontmatter statuses.
+
+**Task block format** (one per `##` section):
+
+```
+## Task Title
+
+` ` `task
+id: USER-WP-NNNN-T01
+status: wait | todo | progress | done | cancel
+priority: high | medium | low
+state_hub_task_id: "<uuid>"         # written by fix-consistency — do not edit
+` ` `
+
+Task description text.
+```
+
+Status progression: `todo` → `progress` → `done`; use `wait` for waiting/blocked work and `cancel` for stopped work.
+
+To create a new workplan:
+1. Write the file following the format above
+2. Notify the custodian operator to run `make fix-consistency REPO=user-engine`
+   (or send a message to the hub agent via `POST /messages/`)

CLAUDE.md

@@ -0,0 +1,12 @@
+# user-engine — Claude Code Instructions
+
+@SCOPE.md
+@.claude/rules/repo-identity.md
+@.claude/rules/session-protocol.md
+@.claude/rules/first-session.md
+@.claude/rules/workplan-convention.md
+@.claude/rules/stack-and-commands.md
+@.claude/rules/architecture.md
+@.claude/rules/repo-boundary.md
+@.claude/rules/credential-routing.md
+@.claude/rules/agents.md

docs/contracts.md

@@ -249,3 +249,21 @@ with a factory that returns a fresh store. The harness covers readiness,
 idempotent migration, core save/read/query behavior, transaction rollback,
 outbox ordering, and diagnostics that expose counts without raw factor or
 profile values.
+
+`user_engine.store_records` defines the JSONB serialization contract for the
+generic record table. `store_record_for` turns supported domain dataclasses
+into `StoreRecord` envelopes with deterministic keys and index metadata, while
+`domain_record_from_store_record` restores those payloads to domain objects.
+These payloads are durable state and may contain sensitive values, so they must
+not be emitted as diagnostics.
+
+`user_engine.adapters.postgres.PostgresUserEngineStore` is the optional
+Postgres implementation. It accepts a provider-owned DB-API or psycopg-like
+connection, applies the bootstrap SQL in `migrate`, and persists generic
+records, audit records, and pending outbox events without depending on a
+specific driver package.
+
+`user_engine.testing.postgres_provider` provides env-gated live conformance
+helpers for provider repositories. They require a dedicated test DSN plus
+`USER_ENGINE_POSTGRES_TEST_RESET=1` before deleting rows from bootstrap-owned
+tables.

docs/development.md

@@ -10,13 +10,14 @@ tested immediately in local and agent environments.
 
 ```text
 src/user_engine/
-  adapters/    local standalone adapters and deterministic test doubles
+  adapters/    local standalone adapters, Postgres adapter, and test doubles
   domain/      transport- and persistence-neutral domain schemas
   errors.py    typed service exceptions for callers and future transports
   migrations.py ordered durable-store migration manifest
   ports.py     adapter protocols for identity, authorization, events, audit,
                membership export, application bindings, and secrets
   service.py   headless service API for the isolated MVP
+  store_records.py JSON-safe durable-store record serialization
   testing/     local fixtures for tests and examples
 tests/         standard-library unittest suite
 ```
@@ -36,6 +37,19 @@ The command runs:
 PYTHONPATH=src python3 -m unittest discover -s tests -p 'test_*.py'
 ```
 
+Live Postgres conformance tests are skipped by default. To run them against a
+dedicated disposable database, install `psycopg` or `psycopg2` in the active
+environment and set:
+
+```bash
+USER_ENGINE_POSTGRES_TEST_DSN='postgresql://...' \
+USER_ENGINE_POSTGRES_TEST_RESET=1 \
+make test
+```
+
+The reset flag is required because those tests delete rows from the
+bootstrap-owned `user_engine_*` tables.
+
 ## Implementation Rule
 
 Add new behavior in this order:

docs/postgres-durable-store-consumer-requirements.md

@@ -289,11 +289,34 @@ bootstrap schema, and `user_engine.testing.store_conformance` exposes a
 reusable harness that future adapters can run with their own store factory.
 The standard local suite runs that harness against `InMemoryUserEngineStore`.
 
+USER-WP-0017 adds the provider-neutral serialization layer. Future Postgres
+adapters should use `user_engine.store_records.store_record_for` before writing
+to `user_engine_records` and `domain_record_from_store_record` after reading
+JSONB payloads back. The `StoreRecord` envelope maps directly to the generic
+record table columns: `record_type`, `record_key`, `tenant`, `user_id`,
+`application_id`, `scope_type`, `scope_id`, and `payload`.
+
+Durable payloads are raw state, not diagnostics. They can include factor
+values, profile values, prepared-account matches, and access-profile defaults.
+Adapters must avoid logging payloads and should use `record_counts` or other
+redacted diagnostics for observability.
+
+USER-WP-0018 adds `PostgresUserEngineStore`, a dependency-free adapter that
+accepts a provider-supplied DB-API or psycopg-like connection. It writes generic
+records through `StoreRecord`, keeps audit and outbox payloads in their
+dedicated bootstrap tables, applies the bootstrap SQL through `migrate`, and
+uses the shared conformance harness with a fake Postgres connection for local
+unit coverage.
+
+USER-WP-0019 adds optional provider-backed conformance tests. They are skipped
+by default and run only when a dedicated test database is supplied through
+`USER_ENGINE_POSTGRES_TEST_DSN` and destructive cleanup is acknowledged with
+`USER_ENGINE_POSTGRES_TEST_RESET=1`. The helper supports either `psycopg` or
+`psycopg2` when a provider repository installs one of them. Cleanup touches
+only the bootstrap-owned `user_engine_*` tables.
+
 Likely future follow-up work should be:
 
-- Add a Postgres adapter behind the existing store boundary.
-- Add provider-backed conformance tests for locking, uniqueness races,
-  migration readiness, outbox claiming, redacted diagnostics, and restore
-  validation.
-- Add conformance tests that run against both in-memory and Postgres stores.
+- Add provider-backed conformance tests for locking, uniqueness races, outbox
+  claiming, redacted diagnostics, and restore validation.
 - Integrate the adapter with the future NetKingdom Postgres provider repo.

registry/capabilities/capability.identity.user-engine.md

@@ -0,0 +1,123 @@
+---
+id: capability.identity.user-engine
+name: Headless Multi-Tenant User Engine
+summary: Headless, multi-application, multi-tenant user management engine covering registration, identity/factor
+  models, entitlement claims, hats/realms/services/assets access profiles, and onboarding journeys.
+owner: user-engine
+status: draft
+domain: communication
+tags:
+- identity
+- user-management
+- multi-tenant
+maturity:
+  discovery:
+    current: D4
+    target: D5
+    confidence: high
+    rationale: 'Unusually large documentation set for its size: docs/development.md, configuration.md,
+      contracts.md, canon-mapping.md, canon-interface-card.yaml, family-dataspace-onboarding.md, registration-identity-and-factor-model.md,
+      prepared-accounts-and-entitlement-claims.md, hats-realms-services-assets-access-profiles.md, onboarding-journeys-and-welcome-protocols.md,
+      and more — scope and contracts are thoroughly documented.'
+  availability:
+    current: A1
+    target: A3
+    confidence: medium
+    rationale: Python package with `make test`; no packaged distribution or hosted service confirmed in
+      this sweep.
+external_evidence:
+  completeness:
+    level: C2
+    confidence: low
+    basis: scope_vs_intent_and_consumer_expectations
+    satisfied_expectations:
+    - registration/identity/factor model documented
+    - entitlement claims and access-profile model (hats/realms/services/assets) documented
+    - canon mapping and interface card present
+    broken_expectations: []
+    out_of_scope_expectations: []
+  reliability:
+    level: R0
+    confidence: low
+    basis: consumer_quality_signals
+    known_reliability_risks:
+    - breadth of documented scope not independently verified against implementation completeness in this
+      sweep
+discovery:
+  intent: Provide a headless, multi-application, multi-tenant user-management engine covering the full
+    registration-to-entitlement lifecycle, reusable across NetKingdom-family applications.
+  includes:
+  - multi-tenant registration and identity/factor model
+  - entitlement claims and hats/realms/services/assets access profiles
+  - onboarding journeys and welcome protocols
+  excludes:
+  - IAM/OIDC protocol implementation itself (see key-cape, net-kingdom)
+  assumptions: []
+  use_cases: []
+  research_memos: []
+availability:
+  current_level: A1
+  target_level: A3
+  current_artifacts:
+  - Python package (`user-engine`)
+  target_artifacts: []
+  consumption_modes:
+  - library import
+relations:
+  depends_on: []
+  supports: []
+  related_to: []
+evidence:
+  documentation:
+  - docs/contracts.md
+  - docs/registration-identity-and-factor-model.md
+  - docs/canon-mapping.md
+  tests:
+  - tests/
+  consumer_feedback: []
+  bug_reports: []
+  incidents: []
+consumer_guidance:
+  recommended_for:
+  - multi-tenant applications needing a headless user/identity/entitlement engine
+  not_recommended_for:
+  - needs for the IAM/OIDC protocol layer itself (see key-cape/net-kingdom)
+  known_limitations:
+  - large documented scope; implementation completeness not independently verified in this sweep
+promotion_history: []
+---
+
+# Headless Multi-Tenant User Engine
+
+## Overview
+
+`user-engine` is a headless, multi-application, multi-tenant user-management engine: registration, identity/factor models, prepared accounts and entitlement claims, hats/realms/services/assets access profiles, and onboarding journeys, documented in unusual depth for its size.
+
+## Assessment notes
+
+### Discovery
+
+Unusually large documentation set for its size: docs/development.md, configuration.md, contracts.md, canon-mapping.md, canon-interface-card.yaml, family-dataspace-onboarding.md, registration-identity-and-factor-model.md, prepared-accounts-and-entitlement-claims.md, hats-realms-services-assets-access-profiles.md, onboarding-journeys-and-welcome-protocols.md, and more — scope and contracts are thoroughly documented.
+
+### Availability
+
+Python package with `make test`; no packaged distribution or hosted service confirmed in this sweep.
+
+### Completeness
+
+First-pass honest assessment from the REUSE-WP-0017 coverage campaign
+(reuse-surface). No external consumer feedback exists yet; levels reflect
+scope-vs-intent documentation quality, not internal code quality.
+
+### Reliability
+
+No production consumer telemetry exists yet; reliability level is
+intentionally conservative pending REUSE-WP-0019 reuse-telemetry evidence.
+
+## Promotion checklist
+
+- [x] ID follows `capability.<domain>.<name>` pattern
+- [x] Maturity enums match `specs/CapabilityMaturityStandard.md`
+- [x] `external_evidence` is populated separately from `maturity`
+- [ ] Relations reference valid capability IDs (none yet)
+- [x] Index entry added in `registry/indexes/capabilities.yaml`

registry/indexes/capabilities.yaml

@@ -1,4 +1,19 @@
 version: 1
-updated: '2026-06-16'
+updated: '2026-07-06'
 domain: helix_forge
-capabilities: []
+capabilities:
+- id: capability.identity.user-engine
+  name: Headless Multi-Tenant User Engine
+  summary: Headless, multi-application, multi-tenant user management engine covering registration, identity/factor
+    models, entitlement claims, hats/realms/services/assets access profiles, and onboarding journeys.
+  vector: D4 / A1 / C2 / R0
+  domain: communication
+  status: draft
+  owner: user-engine
+  path: registry/capabilities/capability.identity.user-engine.md
+  tags:
+  - identity
+  - user-management
+  - multi-tenant
+  consumption_modes:
+  - library import

src/user_engine/adapters/__init__.py

@@ -4,8 +4,10 @@ from user_engine.adapters.local import (
     InMemoryUserEngineStore,
     LocalAuthorizationCheckPort,
 )
+from user_engine.adapters.postgres import PostgresUserEngineStore
 
 __all__ = [
     "InMemoryUserEngineStore",
     "LocalAuthorizationCheckPort",
+    "PostgresUserEngineStore",
 ]

src/user_engine/adapters/postgres.py

@@ -0,0 +1,638 @@
+"""Postgres-backed store adapter.
+
+The adapter is dependency-free: callers provide a DB-API or psycopg-like
+connection object. Provider repositories remain responsible for creating,
+pooling, securing, and observing those connections.
+"""
+
+from __future__ import annotations
+
+import json
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any, Iterable, Iterator, Mapping, Protocol, cast
+
+from user_engine.domain import (
+    Account,
+    AccessProfile,
+    ActiveAccessContext,
+    Application,
+    ApplicationBinding,
+    AuditRecord,
+    Catalog,
+    ExternalIdentity,
+    FamilyInvitation,
+    IdentityFactor,
+    Membership,
+    OnboardingJourney,
+    OutboxEvent,
+    PreparedAccount,
+    ProfileValue,
+    RegistrationSession,
+    TenantAccount,
+    User,
+    WelcomeProtocol,
+)
+from user_engine.migrations import LATEST_SCHEMA_VERSION, USER_ENGINE_RECORD_COUNT_KEYS
+from user_engine.store_records import (
+    StoreRecord,
+    composite_record_key,
+    domain_record_from_store_record,
+    store_record_for,
+)
+
+_RECORD_COLUMNS = (
+    "record_type, record_key, tenant, user_id, application_id, "
+    "scope_type, scope_id, payload"
+)
+_RECORD_COUNT_KEY_BY_TYPE = {
+    "application_bindings": "bindings",
+}
+
+
+class PostgresCursor(Protocol):
+    def execute(self, sql: str, params: Iterable[Any] | None = None) -> Any:
+        """Execute a SQL statement."""
+
+    def fetchone(self) -> Any | None:
+        """Fetch one row from the previous query."""
+
+    def fetchall(self) -> Iterable[Any]:
+        """Fetch all rows from the previous query."""
+
+    def close(self) -> Any:
+        """Close the cursor."""
+
+
+class PostgresConnection(Protocol):
+    def cursor(self) -> PostgresCursor:
+        """Return a DB-API-like cursor."""
+
+    def commit(self) -> Any:
+        """Commit the current transaction."""
+
+    def rollback(self) -> Any:
+        """Roll back the current transaction."""
+
+
+class PostgresUserEngineStore:
+    """Postgres implementation of the `UserEngineStore` protocol."""
+
+    def __init__(self, connection: PostgresConnection) -> None:
+        self.connection = connection
+
+    @property
+    def schema_version(self) -> str | None:
+        return LATEST_SCHEMA_VERSION if self._has_latest_schema() else None
+
+    @property
+    def ready(self) -> bool:
+        return self.schema_version == LATEST_SCHEMA_VERSION
+
+    def migrate(self) -> None:
+        with self._cursor() as cursor:
+            for statement in _bootstrap_sql_statements():
+                cursor.execute(statement)
+        self.connection.commit()
+
+    @contextmanager
+    def transaction(self) -> Iterator[None]:
+        begin = getattr(self.connection, "begin", None)
+        if callable(begin):
+            begin()
+        try:
+            yield
+        except Exception:
+            self.connection.rollback()
+            raise
+        else:
+            self.connection.commit()
+
+    def save_user(self, user: User) -> None:
+        self._upsert_record(user)
+
+    def user(self, user_id: str) -> User | None:
+        return cast(User | None, self._get_record("users", user_id))
+
+    def save_account(self, account: Account) -> None:
+        self._upsert_record(account)
+
+    def user_account(self, user_id: str) -> Account | None:
+        return cast(Account | None, self._get_record("accounts", user_id))
+
+    def save_identity(self, identity: ExternalIdentity) -> None:
+        self._upsert_record(identity)
+
+    def find_identity(self, issuer: str, subject: str) -> ExternalIdentity | None:
+        key = composite_record_key(issuer, subject)
+        return cast(ExternalIdentity | None, self._get_record("external_identities", key))
+
+    def identities_for_user(self, user_id: str) -> tuple[ExternalIdentity, ...]:
+        return cast(
+            tuple[ExternalIdentity, ...],
+            self._query_records("external_identities", user_id=user_id),
+        )
+
+    def save_tenant_account(self, account: TenantAccount) -> None:
+        self._upsert_record(account)
+
+    def tenant_account(self, tenant: str, user_id: str) -> TenantAccount | None:
+        key = composite_record_key(tenant, user_id)
+        return cast(TenantAccount | None, self._get_record("tenant_accounts", key))
+
+    def save_membership(self, membership: Membership) -> None:
+        self._upsert_record(membership)
+
+    def memberships_for_user(
+        self, user_id: str, *, tenant: str | None = None
+    ) -> tuple[Membership, ...]:
+        return cast(
+            tuple[Membership, ...],
+            self._query_records("memberships", user_id=user_id, tenant=tenant),
+        )
+
+    def memberships_for_tenant(self, tenant: str) -> tuple[Membership, ...]:
+        return cast(
+            tuple[Membership, ...],
+            self._query_records("memberships", tenant=tenant),
+        )
+
+    def save_application(self, application: Application) -> None:
+        self._upsert_record(application)
+
+    def application(self, application_id: str) -> Application | None:
+        return cast(Application | None, self._get_record("applications", application_id))
+
+    def save_binding(self, binding: ApplicationBinding) -> None:
+        self._upsert_record(binding)
+
+    def binding(self, application_id: str) -> ApplicationBinding | None:
+        return cast(
+            ApplicationBinding | None,
+            self._get_record("application_bindings", application_id),
+        )
+
+    def save_catalog(self, catalog: Catalog) -> None:
+        self._upsert_record(catalog)
+
+    def catalog(self, catalog_id: str) -> Catalog | None:
+        return cast(Catalog | None, self._get_record("catalogs", catalog_id))
+
+    def all_catalogs(self) -> tuple[Catalog, ...]:
+        return cast(tuple[Catalog, ...], self._query_records("catalogs"))
+
+    def save_family_invitation(self, invitation: FamilyInvitation) -> None:
+        self._upsert_record(invitation)
+
+    def family_invitation(self, invitation_id: str) -> FamilyInvitation | None:
+        return cast(
+            FamilyInvitation | None,
+            self._get_record("family_invitations", invitation_id),
+        )
+
+    def family_invitations_for_user(
+        self, user_id: str
+    ) -> tuple[FamilyInvitation, ...]:
+        return cast(
+            tuple[FamilyInvitation, ...],
+            self._query_records("family_invitations", user_id=user_id),
+        )
+
+    def save_registration_session(self, session: RegistrationSession) -> None:
+        self._upsert_record(session)
+
+    def registration_session(
+        self, registration_id: str
+    ) -> RegistrationSession | None:
+        return cast(
+            RegistrationSession | None,
+            self._get_record("registration_sessions", registration_id),
+        )
+
+    def all_registration_sessions(self) -> tuple[RegistrationSession, ...]:
+        return cast(
+            tuple[RegistrationSession, ...],
+            self._query_records("registration_sessions"),
+        )
+
+    def save_identity_factor(self, factor: IdentityFactor) -> None:
+        self._upsert_record(factor)
+
+    def identity_factor(self, factor_id: str) -> IdentityFactor | None:
+        return cast(
+            IdentityFactor | None,
+            self._get_record("identity_factors", factor_id),
+        )
+
+    def factors_for_registration(
+        self, registration_id: str
+    ) -> tuple[IdentityFactor, ...]:
+        return cast(
+            tuple[IdentityFactor, ...],
+            self._query_records(
+                "identity_factors",
+                scope_type="registration",
+                scope_id=registration_id,
+            ),
+        )
+
+    def factors_for_user(self, user_id: str) -> tuple[IdentityFactor, ...]:
+        return cast(
+            tuple[IdentityFactor, ...],
+            self._query_records("identity_factors", user_id=user_id),
+        )
+
+    def save_prepared_account(self, account: PreparedAccount) -> None:
+        self._upsert_record(account)
+
+    def prepared_account(self, prepared_account_id: str) -> PreparedAccount | None:
+        return cast(
+            PreparedAccount | None,
+            self._get_record("prepared_accounts", prepared_account_id),
+        )
+
+    def prepared_accounts_for_tenant(
+        self, tenant: str
+    ) -> tuple[PreparedAccount, ...]:
+        return cast(
+            tuple[PreparedAccount, ...],
+            self._query_records("prepared_accounts", tenant=tenant),
+        )
+
+    def save_access_profile(self, profile: AccessProfile) -> None:
+        self._upsert_record(profile)
+
+    def access_profile(self, access_profile_id: str) -> AccessProfile | None:
+        return cast(
+            AccessProfile | None,
+            self._get_record("access_profiles", access_profile_id),
+        )
+
+    def access_profiles_for_tenant(self, tenant: str) -> tuple[AccessProfile, ...]:
+        return cast(
+            tuple[AccessProfile, ...],
+            self._query_records("access_profiles", tenant=tenant),
+        )
+
+    def save_active_access_context(self, context: ActiveAccessContext) -> None:
+        self._upsert_record(context)
+
+    def active_access_context(
+        self, user_id: str, tenant: str
+    ) -> ActiveAccessContext | None:
+        key = composite_record_key(user_id, tenant)
+        return cast(
+            ActiveAccessContext | None,
+            self._get_record("active_access_contexts", key),
+        )
+
+    def active_access_contexts_for_tenant(
+        self, tenant: str
+    ) -> tuple[ActiveAccessContext, ...]:
+        return cast(
+            tuple[ActiveAccessContext, ...],
+            self._query_records("active_access_contexts", tenant=tenant),
+        )
+
+    def save_welcome_protocol(self, protocol: WelcomeProtocol) -> None:
+        self._upsert_record(protocol)
+
+    def welcome_protocol(self, protocol_id: str) -> WelcomeProtocol | None:
+        return cast(
+            WelcomeProtocol | None,
+            self._get_record("welcome_protocols", protocol_id),
+        )
+
+    def welcome_protocols_for_tenant(
+        self, tenant: str
+    ) -> tuple[WelcomeProtocol, ...]:
+        return cast(
+            tuple[WelcomeProtocol, ...],
+            self._query_records("welcome_protocols", tenant=tenant),
+        )
+
+    def save_onboarding_journey(self, journey: OnboardingJourney) -> None:
+        self._upsert_record(journey)
+
+    def onboarding_journey(self, journey_id: str) -> OnboardingJourney | None:
+        return cast(
+            OnboardingJourney | None,
+            self._get_record("onboarding_journeys", journey_id),
+        )
+
+    def onboarding_journeys_for_user(
+        self, user_id: str, *, tenant: str | None = None
+    ) -> tuple[OnboardingJourney, ...]:
+        return cast(
+            tuple[OnboardingJourney, ...],
+            self._query_records("onboarding_journeys", user_id=user_id, tenant=tenant),
+        )
+
+    def onboarding_journeys_for_tenant(
+        self, tenant: str
+    ) -> tuple[OnboardingJourney, ...]:
+        return cast(
+            tuple[OnboardingJourney, ...],
+            self._query_records("onboarding_journeys", tenant=tenant),
+        )
+
+    def save_profile_value(self, value: ProfileValue) -> None:
+        self._upsert_record(value)
+
+    def values_for_user(self, user_id: str) -> tuple[ProfileValue, ...]:
+        return cast(
+            tuple[ProfileValue, ...],
+            self._query_records("profile_values", user_id=user_id),
+        )
+
+    def append_audit(self, record: AuditRecord) -> None:
+        store_record = store_record_for(record)
+        with self._cursor() as cursor:
+            cursor.execute(
+                """
+                INSERT INTO user_engine_audit_records (
+                    audit_id, tenant, actor_issuer, actor_subject, action,
+                    subject, correlation_id, summary, payload
+                )
+                VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s::jsonb)
+                ON CONFLICT (audit_id) DO NOTHING
+                """,
+                (
+                    record.audit_id,
+                    record.tenant,
+                    record.actor.issuer,
+                    record.actor.subject,
+                    record.action,
+                    record.subject,
+                    record.correlation_id,
+                    record.summary,
+                    json.dumps(store_record.payload),
+                ),
+            )
+
+    def audit_log(self) -> tuple[AuditRecord, ...]:
+        with self._cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT payload
+                FROM user_engine_audit_records
+                ORDER BY recorded_at, audit_id
+                """
+            )
+            return tuple(
+                cast(AuditRecord, self._decode_payload_row("audit_records", row))
+                for row in cursor.fetchall()
+            )
+
+    def append_outbox(self, event: OutboxEvent) -> None:
+        store_record = store_record_for(event)
+        with self._cursor() as cursor:
+            cursor.execute(
+                """
+                INSERT INTO user_engine_outbox_events (
+                    event_id, tenant, event_type, aggregate_id, correlation_id,
+                    payload, occurred_at
+                )
+                VALUES (%s, %s, %s, %s, %s, %s::jsonb, %s)
+                ON CONFLICT (event_id) DO NOTHING
+                """,
+                (
+                    event.event_id,
+                    event.tenant,
+                    event.event_type,
+                    event.aggregate_id,
+                    event.correlation_id,
+                    json.dumps(store_record.payload),
+                    event.occurred_at,
+                ),
+            )
+
+    def pending_outbox(self) -> tuple[OutboxEvent, ...]:
+        with self._cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT payload
+                FROM user_engine_outbox_events
+                WHERE claimed_at IS NULL AND delivered_at IS NULL
+                ORDER BY occurred_at, event_id
+                """
+            )
+            return tuple(
+                cast(OutboxEvent, self._decode_payload_row("outbox_events", row))
+                for row in cursor.fetchall()
+            )
+
+    def record_counts(self) -> Mapping[str, int]:
+        counts = {key: 0 for key in USER_ENGINE_RECORD_COUNT_KEYS}
+        with self._cursor() as cursor:
+            cursor.execute(
+                """
+                SELECT record_type, COUNT(*)
+                FROM user_engine_records
+                GROUP BY record_type
+                """
+            )
+            for record_type, count in cursor.fetchall():
+                key = _RECORD_COUNT_KEY_BY_TYPE.get(record_type, record_type)
+                if key in counts:
+                    counts[key] = int(count)
+
+            cursor.execute("SELECT COUNT(*) FROM user_engine_audit_records")
+            counts["audit_records"] = int(_first_column(cursor.fetchone()) or 0)
+            cursor.execute(
+                """
+                SELECT COUNT(*)
+                FROM user_engine_outbox_events
+                WHERE claimed_at IS NULL AND delivered_at IS NULL
+                """
+            )
+            counts["pending_outbox_events"] = int(
+                _first_column(cursor.fetchone()) or 0
+            )
+        return counts
+
+    def _upsert_record(self, value: Any) -> None:
+        record = store_record_for(value)
+        with self._cursor() as cursor:
+            cursor.execute(
+                f"""
+                INSERT INTO user_engine_records (
+                    record_type, record_key, tenant, user_id, application_id,
+                    scope_type, scope_id, payload
+                )
+                VALUES (%s, %s, %s, %s, %s, %s, %s, %s::jsonb)
+                ON CONFLICT (record_type, record_key) DO UPDATE SET
+                    tenant = EXCLUDED.tenant,
+                    user_id = EXCLUDED.user_id,
+                    application_id = EXCLUDED.application_id,
+                    scope_type = EXCLUDED.scope_type,
+                    scope_id = EXCLUDED.scope_id,
+                    payload = EXCLUDED.payload,
+                    updated_at = now()
+                """,
+                (
+                    record.record_type,
+                    record.record_key,
+                    record.tenant,
+                    record.user_id,
+                    record.application_id,
+                    record.scope_type,
+                    record.scope_id,
+                    json.dumps(record.payload),
+                ),
+            )
+
+    def _get_record(self, record_type: str, record_key: str) -> Any | None:
+        with self._cursor() as cursor:
+            cursor.execute(
+                f"""
+                SELECT {_RECORD_COLUMNS}
+                FROM user_engine_records
+                WHERE record_type = %s AND record_key = %s
+                """,
+                (record_type, record_key),
+            )
+            row = cursor.fetchone()
+        if row is None:
+            return None
+        return domain_record_from_store_record(_store_record_from_row(row))
+
+    def _query_records(
+        self,
+        record_type: str,
+        *,
+        tenant: str | None = None,
+        user_id: str | None = None,
+        application_id: str | None = None,
+        scope_type: str | None = None,
+        scope_id: str | None = None,
+    ) -> tuple[Any, ...]:
+        filters = {
+            "tenant": tenant,
+            "user_id": user_id,
+            "application_id": application_id,
+            "scope_type": scope_type,
+            "scope_id": scope_id,
+        }
+        clauses = ["record_type = %s"]
+        params: list[Any] = [record_type]
+        for column, value in filters.items():
+            if value is not None:
+                clauses.append(f"{column} = %s")
+                params.append(value)
+        where_clause = " AND ".join(clauses)
+        with self._cursor() as cursor:
+            cursor.execute(
+                f"""
+                SELECT {_RECORD_COLUMNS}
+                FROM user_engine_records
+                WHERE {where_clause}
+                ORDER BY record_key
+                """,
+                tuple(params),
+            )
+            rows = cursor.fetchall()
+        return tuple(
+            domain_record_from_store_record(_store_record_from_row(row))
+            for row in rows
+        )
+
+    def _decode_payload_row(self, record_type: str, row: Any) -> Any:
+        payload = _first_column(row)
+        if isinstance(payload, str):
+            payload = json.loads(payload)
+        record_key = str(cast(Mapping[str, Any], payload).get("event_id") or "")
+        if record_type == "audit_records":
+            record_key = str(cast(Mapping[str, Any], payload).get("audit_id") or "")
+        return domain_record_from_store_record(
+            StoreRecord(record_type=record_type, record_key=record_key, payload=payload)
+        )
+
+    def _has_latest_schema(self) -> bool:
+        try:
+            with self._cursor() as cursor:
+                cursor.execute(
+                    """
+                    SELECT 1
+                    FROM user_engine_schema_versions
+                    WHERE version = %s
+                    """,
+                    (LATEST_SCHEMA_VERSION,),
+                )
+                return cursor.fetchone() is not None
+        except Exception:
+            self.connection.rollback()
+            return False
+
+    @contextmanager
+    def _cursor(self) -> Iterator[PostgresCursor]:
+        cursor = self.connection.cursor()
+        try:
+            yield cursor
+        finally:
+            close = getattr(cursor, "close", None)
+            if callable(close):
+                close()
+
+
+def _store_record_from_row(row: Any) -> StoreRecord:
+    if isinstance(row, Mapping):
+        payload = row["payload"]
+        if isinstance(payload, str):
+            payload = json.loads(payload)
+        return StoreRecord(
+            record_type=str(row["record_type"]),
+            record_key=str(row["record_key"]),
+            tenant=cast(str | None, row.get("tenant")),
+            user_id=cast(str | None, row.get("user_id")),
+            application_id=cast(str | None, row.get("application_id")),
+            scope_type=cast(str | None, row.get("scope_type")),
+            scope_id=cast(str | None, row.get("scope_id")),
+            payload=payload,
+        )
+
+    (
+        record_type,
+        record_key,
+        tenant,
+        user_id,
+        application_id,
+        scope_type,
+        scope_id,
+        payload,
+    ) = row
+    if isinstance(payload, str):
+        payload = json.loads(payload)
+    return StoreRecord(
+        record_type=str(record_type),
+        record_key=str(record_key),
+        tenant=tenant,
+        user_id=user_id,
+        application_id=application_id,
+        scope_type=scope_type,
+        scope_id=scope_id,
+        payload=payload,
+    )
+
+
+def _first_column(row: Any) -> Any:
+    if row is None:
+        return None
+    if isinstance(row, Mapping):
+        return next(iter(row.values()))
+    return row[0]
+
+
+def _load_bootstrap_sql() -> str:
+    repo_root = Path(__file__).resolve().parents[3]
+    return (repo_root / "migrations/postgres/0001_user_engine_store.sql").read_text(
+        encoding="utf-8"
+    )
+
+
+def _bootstrap_sql_statements() -> tuple[str, ...]:
+    return tuple(
+        f"{statement.strip()};"
+        for statement in _load_bootstrap_sql().split(";")
+        if statement.strip()
+    )

src/user_engine/store_records.py

@@ -0,0 +1,357 @@
+"""JSONB-oriented durable store record serialization."""
+
+from __future__ import annotations
+
+import json
+from collections.abc import Mapping as MappingABC
+from dataclasses import dataclass, fields, is_dataclass
+from datetime import datetime
+from enum import Enum
+from types import NoneType, UnionType
+from typing import Any, Callable, Mapping, Union, get_args, get_origin, get_type_hints
+
+from user_engine.domain import (
+    Account,
+    AccessProfile,
+    ActiveAccessContext,
+    Application,
+    ApplicationBinding,
+    AuditRecord,
+    Catalog,
+    ExternalIdentity,
+    FamilyInvitation,
+    IdentityFactor,
+    Membership,
+    OnboardingJourney,
+    OutboxEvent,
+    PreparedAccount,
+    ProfileValue,
+    RegistrationSession,
+    TenantAccount,
+    User,
+    WelcomeProtocol,
+)
+from user_engine.migrations import USER_ENGINE_STORE_RECORD_TYPES
+
+
+@dataclass(frozen=True)
+class StoreRecord:
+    """One generic durable-store row for a domain object payload."""
+
+    record_type: str
+    record_key: str
+    payload: Mapping[str, Any]
+    tenant: str | None = None
+    user_id: str | None = None
+    application_id: str | None = None
+    scope_type: str | None = None
+    scope_id: str | None = None
+
+
+@dataclass(frozen=True)
+class StoreRecordCodec:
+    """Codec rule for one user-engine store record type."""
+
+    record_type: str
+    model_type: type[Any]
+    record_key: Callable[[Any], str]
+    metadata: Callable[[Any], Mapping[str, str | None]]
+
+
+def store_record_for(value: Any) -> StoreRecord:
+    """Return a generic durable-store record for a supported domain object."""
+    codec = _CODECS_BY_MODEL.get(type(value))
+    if codec is None:
+        raise TypeError(f"unsupported store record type: {type(value).__name__}")
+
+    metadata = dict(codec.metadata(value))
+    return StoreRecord(
+        record_type=codec.record_type,
+        record_key=codec.record_key(value),
+        payload=_encode_dataclass(value),
+        tenant=metadata.get("tenant"),
+        user_id=metadata.get("user_id"),
+        application_id=metadata.get("application_id"),
+        scope_type=metadata.get("scope_type"),
+        scope_id=metadata.get("scope_id"),
+    )
+
+
+def composite_record_key(*parts: str | None) -> str:
+    """Return the deterministic composite key used by durable store records."""
+    return _composite_key(*parts)
+
+
+def domain_record_from_store_record(record: StoreRecord) -> Any:
+    """Decode a durable-store record payload into its domain dataclass."""
+    codec = _CODECS_BY_RECORD_TYPE.get(record.record_type)
+    if codec is None:
+        raise ValueError(f"unsupported store record type: {record.record_type}")
+    return _decode_dataclass(codec.model_type, record.payload)
+
+
+def validate_store_record_codecs() -> tuple[str, ...]:
+    """Return codec coverage errors against the durable-store manifest."""
+    errors: list[str] = []
+    manifest_types = set(USER_ENGINE_STORE_RECORD_TYPES)
+    codec_types = set(_CODECS_BY_RECORD_TYPE)
+    missing = sorted(manifest_types - codec_types)
+    extra = sorted(codec_types - manifest_types)
+    if missing:
+        errors.append(f"missing codecs for: {', '.join(missing)}")
+    if extra:
+        errors.append(f"extra codecs for: {', '.join(extra)}")
+    return tuple(errors)
+
+
+def _encode_dataclass(value: Any) -> Mapping[str, Any]:
+    if not is_dataclass(value):
+        raise TypeError(f"expected dataclass, got {type(value).__name__}")
+    return {
+        field.name: _encode_value(getattr(value, field.name))
+        for field in fields(value)
+    }
+
+
+def _encode_value(value: Any) -> Any:
+    if is_dataclass(value):
+        return _encode_dataclass(value)
+    if isinstance(value, datetime):
+        return value.isoformat()
+    if isinstance(value, Enum):
+        return value.value
+    if isinstance(value, tuple):
+        return [_encode_value(item) for item in value]
+    if isinstance(value, list):
+        return [_encode_value(item) for item in value]
+    if isinstance(value, MappingABC):
+        return {str(key): _encode_value(item) for key, item in value.items()}
+    return value
+
+
+def _decode_dataclass(model_type: type[Any], payload: Mapping[str, Any]) -> Any:
+    hints = get_type_hints(model_type)
+    kwargs = {
+        field.name: _decode_value(payload[field.name], hints[field.name])
+        for field in fields(model_type)
+        if field.name in payload
+    }
+    return model_type(**kwargs)
+
+
+def _decode_value(value: Any, type_hint: Any) -> Any:
+    if value is None:
+        return None
+    if type_hint is Any:
+        return value
+    origin = get_origin(type_hint)
+    args = get_args(type_hint)
+
+    if origin in (UnionType, Union):
+        non_none_args = tuple(arg for arg in args if arg is not NoneType)
+        if len(non_none_args) == 1 and len(non_none_args) != len(args):
+            return _decode_value(value, non_none_args[0])
+        for arg in non_none_args:
+            try:
+                return _decode_value(value, arg)
+            except (TypeError, ValueError):
+                continue
+        return value
+
+    if type_hint is datetime:
+        return datetime.fromisoformat(str(value))
+    if isinstance(type_hint, type) and issubclass(type_hint, Enum):
+        return type_hint(value)
+    if isinstance(type_hint, type) and is_dataclass(type_hint):
+        return _decode_dataclass(type_hint, value)
+    if origin is tuple:
+        if not args:
+            return tuple(value)
+        item_hint = args[0] if len(args) == 2 and args[1] is Ellipsis else None
+        if item_hint is not None:
+            return tuple(_decode_value(item, item_hint) for item in value)
+        return tuple(
+            _decode_value(item, args[index])
+            for index, item in enumerate(value)
+        )
+    if origin is list:
+        item_hint = args[0] if args else Any
+        return [_decode_value(item, item_hint) for item in value]
+    if origin in (dict, Mapping, MappingABC):
+        return dict(value)
+    if type_hint in (str, int, float, bool):
+        return type_hint(value)
+    return value
+
+
+def _single_key(value: str) -> str:
+    return value
+
+
+def _composite_key(*parts: str | None) -> str:
+    return json.dumps(list(parts), separators=(",", ":"), ensure_ascii=True)
+
+
+def _enum_value(value: Any) -> str | None:
+    if value is None:
+        return None
+    if isinstance(value, Enum):
+        return str(value.value)
+    return str(value)
+
+
+_CODECS = (
+    StoreRecordCodec(
+        "users",
+        User,
+        lambda value: _single_key(value.user_id),
+        lambda value: {"user_id": value.user_id},
+    ),
+    StoreRecordCodec(
+        "accounts",
+        Account,
+        lambda value: _single_key(value.user_id),
+        lambda value: {"user_id": value.user_id},
+    ),
+    StoreRecordCodec(
+        "external_identities",
+        ExternalIdentity,
+        lambda value: _composite_key(value.issuer, value.subject),
+        lambda value: {"user_id": value.user_id},
+    ),
+    StoreRecordCodec(
+        "tenant_accounts",
+        TenantAccount,
+        lambda value: _composite_key(value.tenant, value.user_id),
+        lambda value: {"tenant": value.tenant, "user_id": value.user_id},
+    ),
+    StoreRecordCodec(
+        "memberships",
+        Membership,
+        lambda value: _single_key(value.membership_id),
+        lambda value: {
+            "tenant": value.tenant,
+            "user_id": value.user_id,
+            "scope_type": value.scope_type,
+            "scope_id": value.scope_id,
+        },
+    ),
+    StoreRecordCodec(
+        "applications",
+        Application,
+        lambda value: _single_key(value.application_id),
+        lambda value: {"application_id": value.application_id},
+    ),
+    StoreRecordCodec(
+        "application_bindings",
+        ApplicationBinding,
+        lambda value: _single_key(value.application_id),
+        lambda value: {"application_id": value.application_id},
+    ),
+    StoreRecordCodec(
+        "catalogs",
+        Catalog,
+        lambda value: _single_key(value.catalog_id),
+        lambda value: {"application_id": value.owning_application_id},
+    ),
+    StoreRecordCodec(
+        "family_invitations",
+        FamilyInvitation,
+        lambda value: _single_key(value.invitation_id),
+        lambda value: {
+            "tenant": value.tenant,
+            "user_id": value.user_id,
+            "application_id": value.application_id,
+            "scope_type": "family",
+            "scope_id": value.family_scope_id,
+        },
+    ),
+    StoreRecordCodec(
+        "registration_sessions",
+        RegistrationSession,
+        lambda value: _single_key(value.registration_id),
+        lambda value: {"tenant": value.tenant, "user_id": value.user_id},
+    ),
+    StoreRecordCodec(
+        "identity_factors",
+        IdentityFactor,
+        lambda value: _single_key(value.factor_id),
+        lambda value: {
+            "user_id": value.user_id,
+            "scope_type": "registration" if value.registration_id else None,
+            "scope_id": value.registration_id,
+        },
+    ),
+    StoreRecordCodec(
+        "prepared_accounts",
+        PreparedAccount,
+        lambda value: _single_key(value.prepared_account_id),
+        lambda value: {"tenant": value.tenant},
+    ),
+    StoreRecordCodec(
+        "access_profiles",
+        AccessProfile,
+        lambda value: _single_key(value.access_profile_id),
+        lambda value: {
+            "tenant": value.tenant,
+            "scope_type": _enum_value(value.scope_type),
+            "scope_id": value.scope_id,
+        },
+    ),
+    StoreRecordCodec(
+        "active_access_contexts",
+        ActiveAccessContext,
+        lambda value: _composite_key(value.user_id, value.tenant),
+        lambda value: {
+            "tenant": value.tenant,
+            "user_id": value.user_id,
+            "scope_type": _enum_value(value.scope_type),
+            "scope_id": value.scope_id,
+        },
+    ),
+    StoreRecordCodec(
+        "welcome_protocols",
+        WelcomeProtocol,
+        lambda value: _single_key(value.protocol_id),
+        lambda value: {
+            "tenant": value.tenant,
+            "application_id": value.application_id,
+        },
+    ),
+    StoreRecordCodec(
+        "onboarding_journeys",
+        OnboardingJourney,
+        lambda value: _single_key(value.journey_id),
+        lambda value: {"tenant": value.tenant, "user_id": value.user_id},
+    ),
+    StoreRecordCodec(
+        "profile_values",
+        ProfileValue,
+        lambda value: _composite_key(
+            value.user_id,
+            value.attribute_key,
+            _enum_value(value.scope),
+            value.scope_id,
+        ),
+        lambda value: {
+            "user_id": value.user_id,
+            "scope_type": _enum_value(value.scope),
+            "scope_id": value.scope_id,
+        },
+    ),
+    StoreRecordCodec(
+        "audit_records",
+        AuditRecord,
+        lambda value: _single_key(value.audit_id),
+        lambda value: {"tenant": value.tenant, "user_id": value.subject},
+    ),
+    StoreRecordCodec(
+        "outbox_events",
+        OutboxEvent,
+        lambda value: _single_key(value.event_id),
+        lambda value: {"tenant": value.tenant},
+    ),
+)
+
+_CODECS_BY_RECORD_TYPE = {codec.record_type: codec for codec in _CODECS}
+_CODECS_BY_MODEL = {codec.model_type: codec for codec in _CODECS}

src/user_engine/testing/postgres_provider.py

@@ -0,0 +1,82 @@
+"""Opt-in live Postgres conformance helpers for provider repositories."""
+
+from __future__ import annotations
+
+import os
+from dataclasses import dataclass
+from typing import Any, Mapping
+
+from user_engine.adapters.postgres import PostgresConnection, PostgresUserEngineStore
+
+POSTGRES_TEST_DSN_ENV = "USER_ENGINE_POSTGRES_TEST_DSN"
+POSTGRES_TEST_RESET_ENV = "USER_ENGINE_POSTGRES_TEST_RESET"
+_TRUTHY = {"1", "true", "yes", "on"}
+_TABLES = (
+    "user_engine_outbox_events",
+    "user_engine_audit_records",
+    "user_engine_records",
+    "user_engine_schema_versions",
+)
+
+
+@dataclass(frozen=True)
+class PostgresProviderTestConfig:
+    """Configuration for destructive provider-backed Postgres tests."""
+
+    dsn: str
+
+
+def postgres_provider_test_config(
+    environ: Mapping[str, str] | None = None,
+) -> tuple[PostgresProviderTestConfig | None, str | None]:
+    """Return live test config or a skip reason."""
+    env = environ or os.environ
+    dsn = env.get(POSTGRES_TEST_DSN_ENV, "").strip()
+    if not dsn:
+        return None, f"{POSTGRES_TEST_DSN_ENV} is not set"
+    reset_value = env.get(POSTGRES_TEST_RESET_ENV, "").strip().lower()
+    if reset_value not in _TRUTHY:
+        return (
+            None,
+            f"{POSTGRES_TEST_RESET_ENV}=1 is required because tests reset "
+            "user_engine_* tables",
+        )
+    return PostgresProviderTestConfig(dsn=dsn), None
+
+
+def connect_postgres_provider(dsn: str) -> PostgresConnection:
+    """Connect with psycopg3 or psycopg2 when a provider installs either one."""
+    try:
+        import psycopg  # type: ignore[import-not-found]
+
+        return psycopg.connect(dsn)  # type: ignore[no-any-return]
+    except ImportError:
+        pass
+
+    try:
+        import psycopg2  # type: ignore[import-not-found]
+
+        return psycopg2.connect(dsn)  # type: ignore[no-any-return]
+    except ImportError as exc:
+        raise RuntimeError("install psycopg or psycopg2 to run live tests") from exc
+
+
+def reset_user_engine_postgres_tables(connection: PostgresConnection) -> None:
+    """Create then empty user-engine tables in a dedicated provider test DB."""
+    PostgresUserEngineStore(connection).migrate()
+    cursor = connection.cursor()
+    try:
+        for table in _TABLES:
+            cursor.execute(f"DELETE FROM {table}")
+    finally:
+        close = getattr(cursor, "close", None)
+        if callable(close):
+            close()
+    connection.commit()
+
+
+def close_postgres_provider_connection(connection: Any) -> None:
+    """Close provider connections that expose a close method."""
+    close = getattr(connection, "close", None)
+    if callable(close):
+        close()

src/user_engine/testing/store_conformance.py

@@ -89,6 +89,11 @@ def assert_user_engine_store_conformance(
     _assert_diagnostics_contract(testcase, store_factory)
 
 
+def reference_store_records(store: UserEngineStore) -> dict[str, Any]:
+    """Write and return a representative record for every store record type."""
+    return _write_reference_records(store)
+
+
 def _assert_readiness_contract(testcase: TestCase, store_factory: StoreFactory) -> None:
     store = store_factory()
     if store.schema_version is None:

tests/test_postgres_provider_conformance.py

@@ -0,0 +1,107 @@
+import unittest
+
+from user_engine.adapters.postgres import PostgresUserEngineStore
+from user_engine.domain import User
+from user_engine.testing.postgres_provider import (
+    POSTGRES_TEST_DSN_ENV,
+    POSTGRES_TEST_RESET_ENV,
+    close_postgres_provider_connection,
+    connect_postgres_provider,
+    postgres_provider_test_config,
+    reset_user_engine_postgres_tables,
+)
+from user_engine.testing.store_conformance import (
+    assert_user_engine_store_conformance,
+)
+
+
+class ProviderPostgresConfigTests(unittest.TestCase):
+    def test_config_skips_without_dsn(self):
+        config, reason = postgres_provider_test_config({})
+
+        self.assertIsNone(config)
+        self.assertIn(POSTGRES_TEST_DSN_ENV, reason or "")
+
+    def test_config_requires_reset_acknowledgement(self):
+        config, reason = postgres_provider_test_config(
+            {POSTGRES_TEST_DSN_ENV: "postgresql://example.test/db"}
+        )
+
+        self.assertIsNone(config)
+        self.assertIn(POSTGRES_TEST_RESET_ENV, reason or "")
+
+    def test_config_accepts_dsn_and_reset_acknowledgement(self):
+        config, reason = postgres_provider_test_config(
+            {
+                POSTGRES_TEST_DSN_ENV: "postgresql://example.test/db",
+                POSTGRES_TEST_RESET_ENV: "1",
+            }
+        )
+
+        self.assertIsNotNone(config)
+        self.assertIsNone(reason)
+
+
+class ProviderPostgresConformanceTests(unittest.TestCase):
+    def setUp(self):
+        self.config, reason = postgres_provider_test_config()
+        if reason:
+            self.skipTest(reason)
+        self.connections = []
+
+    def tearDown(self):
+        for connection in self.connections:
+            close_postgres_provider_connection(connection)
+        if self.config is not None:
+            cleanup = connect_postgres_provider(self.config.dsn)
+            try:
+                reset_user_engine_postgres_tables(cleanup)
+            finally:
+                close_postgres_provider_connection(cleanup)
+
+    def test_live_postgres_store_satisfies_store_conformance(self):
+        assert_user_engine_store_conformance(self, self._store_factory)
+
+    def test_live_postgres_migration_readiness(self):
+        store = self._store_factory()
+
+        self.assertFalse(store.ready)
+        store.migrate()
+
+        self.assertTrue(store.ready)
+        self.assertEqual(store.schema_version, "0001_initial")
+
+    def test_live_postgres_upsert_keeps_one_logical_record(self):
+        store = self._store_factory()
+        store.migrate()
+        user = User(user_id="usr_live_upsert", display_name="Original")
+        replacement = User(user_id="usr_live_upsert", display_name="Replacement")
+
+        store.save_user(user)
+        store.save_user(replacement)
+
+        self.assertEqual(store.user(user.user_id), replacement)
+        cursor = store.connection.cursor()
+        try:
+            cursor.execute(
+                """
+                SELECT COUNT(*)
+                FROM user_engine_records
+                WHERE record_type = %s AND record_key = %s
+                """,
+                ("users", user.user_id),
+            )
+            self.assertEqual(cursor.fetchone()[0], 1)
+        finally:
+            cursor.close()
+
+    def _store_factory(self) -> PostgresUserEngineStore:
+        assert self.config is not None
+        connection = connect_postgres_provider(self.config.dsn)
+        reset_user_engine_postgres_tables(connection)
+        self.connections.append(connection)
+        return PostgresUserEngineStore(connection)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_postgres_store_adapter.py

@@ -0,0 +1,180 @@
+import copy
+import json
+import unittest
+from typing import Any, Iterable
+
+from user_engine.adapters.postgres import PostgresUserEngineStore
+from user_engine.migrations import LATEST_SCHEMA_VERSION
+from user_engine.store_records import StoreRecord
+from user_engine.testing.store_conformance import (
+    assert_user_engine_store_conformance,
+)
+
+
+class PostgresStoreAdapterTests(unittest.TestCase):
+    def test_fake_postgres_store_satisfies_store_conformance(self):
+        assert_user_engine_store_conformance(
+            self,
+            lambda: PostgresUserEngineStore(_FakePostgresConnection()),
+        )
+
+    def test_ready_is_false_before_migration(self):
+        store = PostgresUserEngineStore(_FakePostgresConnection())
+
+        self.assertFalse(store.ready)
+        self.assertIsNone(store.schema_version)
+
+
+class _FakePostgresConnection:
+    def __init__(self) -> None:
+        self.schema_versions: set[str] = set()
+        self.records: dict[tuple[str, str], StoreRecord] = {}
+        self.audit_payloads: list[dict[str, Any]] = []
+        self.outbox_payloads: list[dict[str, Any]] = []
+        self._snapshot: tuple[
+            set[str],
+            dict[tuple[str, str], StoreRecord],
+            list[dict[str, Any]],
+            list[dict[str, Any]],
+        ] | None = None
+
+    def cursor(self) -> "_FakePostgresCursor":
+        return _FakePostgresCursor(self)
+
+    def begin(self) -> None:
+        self._snapshot = (
+            copy.deepcopy(self.schema_versions),
+            copy.deepcopy(self.records),
+            copy.deepcopy(self.audit_payloads),
+            copy.deepcopy(self.outbox_payloads),
+        )
+
+    def commit(self) -> None:
+        self._snapshot = None
+
+    def rollback(self) -> None:
+        if self._snapshot is None:
+            return
+        (
+            self.schema_versions,
+            self.records,
+            self.audit_payloads,
+            self.outbox_payloads,
+        ) = self._snapshot
+        self._snapshot = None
+
+
+class _FakePostgresCursor:
+    def __init__(self, connection: _FakePostgresConnection) -> None:
+        self.connection = connection
+        self._rows: list[Any] = []
+
+    def execute(self, sql: str, params: Iterable[Any] | None = None) -> None:
+        normalized = " ".join(sql.lower().split())
+        values = tuple(params or ())
+
+        if "insert into user_engine_schema_versions" in normalized:
+            self.connection.schema_versions.add(LATEST_SCHEMA_VERSION)
+            self._rows = []
+            return
+        if "from user_engine_schema_versions" in normalized:
+            self._rows = [(1,)] if values[0] in self.connection.schema_versions else []
+            return
+        if normalized.startswith("insert into user_engine_records"):
+            payload = json.loads(values[7])
+            record = StoreRecord(
+                record_type=values[0],
+                record_key=values[1],
+                tenant=values[2],
+                user_id=values[3],
+                application_id=values[4],
+                scope_type=values[5],
+                scope_id=values[6],
+                payload=payload,
+            )
+            self.connection.records[(record.record_type, record.record_key)] = record
+            self._rows = []
+            return
+        if "from user_engine_records" in normalized:
+            self._select_records(normalized, values)
+            return
+        if normalized.startswith("insert into user_engine_audit_records"):
+            self.connection.audit_payloads.append(json.loads(values[8]))
+            self._rows = []
+            return
+        if normalized.startswith("insert into user_engine_outbox_events"):
+            self.connection.outbox_payloads.append(json.loads(values[5]))
+            self._rows = []
+            return
+        if "from user_engine_audit_records" in normalized:
+            if "count(*)" in normalized:
+                self._rows = [(len(self.connection.audit_payloads),)]
+            else:
+                self._rows = [
+                    (json.dumps(payload),) for payload in self.connection.audit_payloads
+                ]
+            return
+        if "from user_engine_outbox_events" in normalized:
+            if "count(*)" in normalized:
+                self._rows = [(len(self.connection.outbox_payloads),)]
+            else:
+                self._rows = [
+                    (json.dumps(payload),) for payload in self.connection.outbox_payloads
+                ]
+            return
+        self._rows = []
+
+    def fetchone(self) -> Any | None:
+        return self._rows[0] if self._rows else None
+
+    def fetchall(self) -> list[Any]:
+        return self._rows
+
+    def close(self) -> None:
+        return None
+
+    def _select_records(self, normalized: str, values: tuple[Any, ...]) -> None:
+        if "group by record_type" in normalized:
+            counts: dict[str, int] = {}
+            for record_type, _record_key in self.connection.records:
+                counts[record_type] = counts.get(record_type, 0) + 1
+            self._rows = sorted(counts.items())
+            return
+
+        record_type = values[0]
+        filter_columns = [
+            column
+            for column in (
+                "record_key",
+                "tenant",
+                "user_id",
+                "application_id",
+                "scope_type",
+                "scope_id",
+            )
+            if f"{column} = %s" in normalized
+        ]
+        filters = dict(zip(filter_columns, values[1:]))
+        rows = []
+        for (stored_type, _key), record in self.connection.records.items():
+            if stored_type != record_type:
+                continue
+            if any(getattr(record, column) != value for column, value in filters.items()):
+                continue
+            rows.append(
+                (
+                    record.record_type,
+                    record.record_key,
+                    record.tenant,
+                    record.user_id,
+                    record.application_id,
+                    record.scope_type,
+                    record.scope_id,
+                    json.dumps(record.payload),
+                )
+            )
+        self._rows = sorted(rows, key=lambda row: row[1])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_store_records.py

@@ -0,0 +1,93 @@
+import json
+import unittest
+
+from user_engine.adapters.local import InMemoryUserEngineStore
+from user_engine.migrations import USER_ENGINE_STORE_RECORD_TYPES
+from user_engine.store_records import (
+    StoreRecord,
+    domain_record_from_store_record,
+    store_record_for,
+    validate_store_record_codecs,
+)
+from user_engine.testing.store_conformance import (
+    PROFILE_SECRET_VALUE,
+    RAW_FACTOR_VALUE,
+    TENANT,
+    USER_ID,
+    reference_store_records,
+)
+
+
+class StoreRecordSerializationTests(unittest.TestCase):
+    def test_codecs_cover_migration_manifest_record_types(self):
+        self.assertEqual(validate_store_record_codecs(), ())
+
+    def test_reference_records_round_trip_through_json_safe_payloads(self):
+        store = InMemoryUserEngineStore()
+        store.migrate()
+        records = reference_store_records(store)
+        expected_types = set(USER_ENGINE_STORE_RECORD_TYPES)
+
+        encoded_types = set()
+        for value in records.values():
+            record = store_record_for(value)
+            encoded_types.add(record.record_type)
+            json.dumps(record.payload)
+            decoded = domain_record_from_store_record(
+                StoreRecord(
+                    record_type=record.record_type,
+                    record_key=record.record_key,
+                    payload=json.loads(json.dumps(record.payload)),
+                    tenant=record.tenant,
+                    user_id=record.user_id,
+                    application_id=record.application_id,
+                    scope_type=record.scope_type,
+                    scope_id=record.scope_id,
+                )
+            )
+            self.assertEqual(decoded, value)
+
+        self.assertEqual(encoded_types, expected_types)
+
+    def test_record_metadata_supports_provider_indexes(self):
+        store = InMemoryUserEngineStore()
+        store.migrate()
+        records = reference_store_records(store)
+
+        tenant_account = store_record_for(records["tenant_account"])
+        active_context = store_record_for(records["access_context"])
+        profile_value = store_record_for(records["profile_value"])
+        factor = store_record_for(records["factor"])
+
+        self.assertEqual(tenant_account.record_key, f'["{TENANT}","{USER_ID}"]')
+        self.assertEqual(tenant_account.tenant, TENANT)
+        self.assertEqual(tenant_account.user_id, USER_ID)
+        self.assertEqual(active_context.tenant, TENANT)
+        self.assertEqual(active_context.user_id, USER_ID)
+        self.assertEqual(active_context.scope_type, "tenant")
+        self.assertEqual(profile_value.scope_type, "global")
+        self.assertEqual(factor.user_id, USER_ID)
+
+    def test_durable_payloads_are_raw_state_not_diagnostics(self):
+        store = InMemoryUserEngineStore()
+        store.migrate()
+        records = reference_store_records(store)
+
+        factor = store_record_for(records["factor"])
+        access_profile = store_record_for(records["access_profile"])
+
+        self.assertIn(RAW_FACTOR_VALUE, json.dumps(factor.payload))
+        self.assertIn(PROFILE_SECRET_VALUE, json.dumps(access_profile.payload))
+
+    def test_unknown_record_type_fails_closed(self):
+        with self.assertRaises(ValueError):
+            domain_record_from_store_record(
+                StoreRecord(record_type="unknown", record_key="1", payload={})
+            )
+
+        with self.assertRaises(TypeError):
+            store_record_for(object())
+
+
+if __name__ == "__main__":
+    unittest.main()

workplans/USER-WP-0001-preparation-and-interface-adoption.md

@@ -2,7 +2,7 @@
 id: USER-WP-0001
 type: workplan
 title: "User Engine Preparation And Interface Adoption"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0002-isolated-mvp.md

@@ -2,7 +2,7 @@
 id: USER-WP-0002
 type: workplan
 title: "User Engine Isolated MVP"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0003-multi-tenancy.md

@@ -2,7 +2,7 @@
 id: USER-WP-0003
 type: workplan
 title: "User Engine Multi-Tenancy"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0004-multi-application-catalogs.md

@@ -2,7 +2,7 @@
 id: USER-WP-0004
 type: workplan
 title: "User Engine Multi-Application And Catalog Support"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0005-integrated-test-scenarios.md

@@ -2,7 +2,7 @@
 id: USER-WP-0005
 type: workplan
 title: "User Engine Integrated Test Scenarios"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0006-finalization-polish.md

@@ -2,7 +2,7 @@
 id: USER-WP-0006
 type: workplan
 title: "User Engine Implementation Assessment And Polish"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0007-identity-domain-canon-alignment.md

@@ -2,7 +2,7 @@
 id: USER-WP-0007
 type: workplan
 title: "Identity Domain Canon Alignment"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0008-family-dataspace-onboarding.md

@@ -2,7 +2,7 @@
 id: USER-WP-0008
 type: workplan
 title: "Family Dataspace Onboarding"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0009-postgres-durable-store-requirements.md

@@ -2,7 +2,7 @@
 id: USER-WP-0009
 type: workplan
 title: "Postgres Durable Store Consumer Requirements"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0010-registration-identity-and-factor-model.md

@@ -2,7 +2,7 @@
 id: USER-WP-0010
 type: workplan
 title: "Registration Identity And Factor Model"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0011-prepared-accounts-and-entitlement-claims.md

@@ -2,7 +2,7 @@
 id: USER-WP-0011
 type: workplan
 title: "Prepared Accounts And Entitlement Claims"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0012-hats-realms-services-assets-access-profiles.md

@@ -2,7 +2,7 @@
 id: USER-WP-0012
 type: workplan
 title: "Hats, Realms, Services, Assets, And Access Profiles"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0013-onboarding-journeys-and-welcome-protocols.md

@@ -2,7 +2,7 @@
 id: USER-WP-0013
 type: workplan
 title: "Onboarding Journeys And Welcome Protocols"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0014-registration-and-access-management-ui.md

@@ -2,7 +2,7 @@
 id: USER-WP-0014
 type: workplan
 title: "Registration And Access Management UI"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0015-registration-scenario-and-security-conformance.md

@@ -2,7 +2,7 @@
 id: USER-WP-0015
 type: workplan
 title: "Registration Scenario And Security Conformance"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0016-durable-store-migration-and-conformance-harness.md

@@ -2,7 +2,7 @@
 id: USER-WP-0016
 type: workplan
 title: "Durable Store Migration And Conformance Harness"
-domain: netkingdom
+domain: communication
 repo: user-engine
 status: finished
 owner: codex

workplans/USER-WP-0017-durable-store-record-serialization.md

@@ -0,0 +1,116 @@
+---
+id: USER-WP-0017
+type: workplan
+title: "Durable Store Record Serialization"
+domain: communication
+repo: user-engine
+status: finished
+owner: codex
+topic_slug: netkingdom
+planning_priority: medium
+planning_order: 17
+created: "2026-06-16"
+updated: "2026-06-16"
+depends_on:
+  - USER-WP-0016
+state_hub_workstream_id: "bc35882d-5caa-4c2a-9b63-31b0b1c81486"
+---
+
+# USER-WP-0017 - Durable Store Record Serialization
+
+## Goal
+
+Define a dependency-free serialization contract for the generic durable store
+record shape introduced by USER-WP-0016 so a future Postgres adapter can persist
+and restore domain dataclasses through JSONB without embedding ad hoc codecs.
+
+## Scope Direction
+
+This workplan should cover deterministic record keys, adapter metadata columns,
+JSON-safe payload encoding, and round-trip decoding for every logical record
+type in the migration manifest.
+
+## Non-Goals
+
+- Do not add a production Postgres driver.
+- Do not implement connection pooling, migrations, locks, or outbox claiming.
+- Do not redact durable payloads; adapters must avoid logging raw payloads.
+
+## Tasks
+
+```task
+id: USER-WP-0017-T1
+status: done
+priority: high
+state_hub_task_id: "198b3e06-7093-410a-8538-54628e70dfa5"
+```
+
+Add a store-record envelope matching the generic Postgres bootstrap table
+columns.
+
+```task
+id: USER-WP-0017-T2
+status: done
+priority: high
+state_hub_task_id: "d18b5db1-b4c5-4e23-aa12-335cacfa5eb2"
+```
+
+Add deterministic record-key and metadata extraction rules for all manifest
+record types.
+
+```task
+id: USER-WP-0017-T3
+status: done
+priority: high
+state_hub_task_id: "ca1757e6-52c3-44e4-9523-d01c8f1dcc6b"
+```
+
+Add JSON-safe payload encoding for dataclasses, enums, datetimes, tuples, and
+mapping fields.
+
+```task
+id: USER-WP-0017-T4
+status: done
+priority: high
+state_hub_task_id: "33f555f0-2ab3-4f93-ae5a-28b70814bb57"
+```
+
+Add payload decoding back into the original domain dataclasses.
+
+```task
+id: USER-WP-0017-T5
+status: done
+priority: medium
+state_hub_task_id: "93c5bd34-f935-4cc9-ab6c-a0a37865592f"
+```
+
+Document how future Postgres adapters should use the serialization contract.
+
+## Acceptance Criteria
+
+- Every logical record type in the migration manifest has a codec.
+- Encoded payloads can be passed through `json.dumps`.
+- Domain records round-trip through `StoreRecord` without losing enum,
+  datetime, tuple, or nested dataclass structure.
+- Composite keys are deterministic and collision-resistant for scoped records.
+- Documentation warns that durable payloads may contain sensitive values and
+  must not be used as diagnostics output.
+
+## Expected Outputs
+
+- `user_engine.store_records` module.
+- Store-record serialization tests.
+- Durable-store documentation updates.
+
+## Implementation Notes
+
+Implemented on 2026-06-16:
+
+- Added `StoreRecord`, `store_record_for`, `domain_record_from_store_record`,
+  and manifest validation helpers.
+- Added JSON-safe recursive encode/decode support for all current domain
+  dataclasses used by `UserEngineStore`.
+- Added round-trip tests using the same reference records as the conformance
+  harness.
+- Documented the serialization layer as the provider-neutral prerequisite to a
+  live Postgres adapter.

workplans/USER-WP-0018-postgres-store-adapter.md

@@ -0,0 +1,113 @@
+---
+id: USER-WP-0018
+type: workplan
+title: "Postgres Store Adapter"
+domain: communication
+repo: user-engine
+status: finished
+owner: codex
+topic_slug: netkingdom
+planning_priority: medium
+planning_order: 18
+created: "2026-06-16"
+updated: "2026-06-16"
+depends_on:
+  - USER-WP-0016
+  - USER-WP-0017
+state_hub_workstream_id: "192a3a0c-81af-4004-a466-bf670fa99212"
+---
+
+# USER-WP-0018 - Postgres Store Adapter
+
+## Goal
+
+Add a dependency-free Postgres store adapter behind `UserEngineStore` that uses
+the migration and serialization contracts from USER-WP-0016 and USER-WP-0017.
+
+## Scope Direction
+
+The adapter should accept a provider-supplied DB-API or psycopg-like connection
+and avoid owning credentials, pooling, deployment, backups, or platform
+observability.
+
+## Non-Goals
+
+- Do not vendor or require a Postgres driver in the core package.
+- Do not add Docker or live database requirements to the unit test suite.
+- Do not implement outbox claim/ack/retry or provider restore validation yet.
+
+## Tasks
+
+```task
+id: USER-WP-0018-T1
+status: done
+priority: high
+state_hub_task_id: "b1beac60-bc8f-4d00-9387-aecb5c721755"
+```
+
+Add a Postgres adapter that implements the `UserEngineStore` protocol using the
+generic record, audit, and outbox tables.
+
+```task
+id: USER-WP-0018-T2
+status: done
+priority: high
+state_hub_task_id: "794796a0-0851-44c1-a27a-18e037295cb7"
+```
+
+Wire the adapter to `StoreRecord` serialization and deterministic record keys.
+
+```task
+id: USER-WP-0018-T3
+status: done
+priority: high
+state_hub_task_id: "160909a7-f626-4afc-97e2-151eedc4f255"
+```
+
+Support schema readiness, migration execution, transactions, audit reads,
+pending outbox reads, and adapter-neutral record counts.
+
+```task
+id: USER-WP-0018-T4
+status: done
+priority: medium
+state_hub_task_id: "4eeb3043-28a3-46d0-b7da-563686e9e28b"
+```
+
+Add a fake Postgres connection that runs the shared conformance harness without
+requiring production infrastructure.
+
+```task
+id: USER-WP-0018-T5
+status: done
+priority: medium
+state_hub_task_id: "58a6d45d-7f27-477e-9ddf-0d734086db21"
+```
+
+Document the provider boundary and remaining provider-backed conformance work.
+
+## Acceptance Criteria
+
+- The adapter has no hard runtime dependency on a specific Postgres driver.
+- The adapter passes the same store conformance harness as the in-memory store.
+- Migration readiness uses the shared latest schema version.
+- Record counts stay redacted and adapter-neutral.
+- Docs explain that provider repositories still own live-driver, lock, restore,
+  and outbox claiming validation.
+
+## Expected Outputs
+
+- `user_engine.adapters.postgres.PostgresUserEngineStore`.
+- Fake Postgres adapter tests.
+- Durable-store documentation updates.
+
+## Implementation Notes
+
+Implemented on 2026-06-16:
+
+- Added `PostgresUserEngineStore` using provider-supplied DB-API/psycopg-like
+  connections.
+- Reused `StoreRecord` serialization for all generic record writes and reads.
+- Added transaction, migration, readiness, audit, outbox, and record count
+  support.
+- Added fake connection tests that run the shared store conformance harness.

workplans/USER-WP-0019-provider-backed-postgres-conformance.md

@@ -0,0 +1,113 @@
+---
+id: USER-WP-0019
+type: workplan
+title: "Provider Backed Postgres Conformance"
+domain: communication
+repo: user-engine
+status: finished
+owner: codex
+topic_slug: netkingdom
+planning_priority: medium
+planning_order: 19
+created: "2026-06-16"
+updated: "2026-06-16"
+depends_on:
+  - USER-WP-0018
+state_hub_workstream_id: "40dbb193-1cbc-49b2-a08e-044f504c25e6"
+---
+
+# USER-WP-0019 - Provider Backed Postgres Conformance
+
+## Goal
+
+Add opt-in live Postgres conformance tests for `PostgresUserEngineStore` so
+provider repositories can prove the adapter against a real database without
+making ordinary user-engine tests require infrastructure.
+
+## Scope Direction
+
+The suite should be skipped unless an explicit test DSN and destructive reset
+acknowledgement are supplied. It should cover migration readiness, the shared
+store conformance harness, uniqueness/upsert behavior, rollback semantics, and
+record-count diagnostics against a real provider connection.
+
+## Non-Goals
+
+- Do not add a mandatory Postgres driver dependency.
+- Do not run live database tests by default.
+- Do not implement outbox claim/ack/retry or restore validation yet.
+
+## Tasks
+
+```task
+id: USER-WP-0019-T1
+status: done
+priority: high
+state_hub_task_id: "425f0c51-4333-4f63-8397-ddc0bfe8a428"
+```
+
+Add env-gated live Postgres connection helpers that support `psycopg` or
+`psycopg2` when installed.
+
+```task
+id: USER-WP-0019-T2
+status: done
+priority: high
+state_hub_task_id: "f180d009-097d-45b9-aa1f-586f219f7916"
+```
+
+Require an explicit destructive reset acknowledgement before cleaning
+`user_engine_*` provider test tables.
+
+```task
+id: USER-WP-0019-T3
+status: done
+priority: high
+state_hub_task_id: "e2bbf3bd-04a2-4344-a252-330f2a190a0c"
+```
+
+Run the shared store conformance harness against a live provider connection
+when configured.
+
+```task
+id: USER-WP-0019-T4
+status: done
+priority: medium
+state_hub_task_id: "f3b781d0-3a46-4bab-a2c0-f4eb14d70ca9"
+```
+
+Add live checks for migration readiness and deterministic upsert uniqueness.
+
+```task
+id: USER-WP-0019-T5
+status: done
+priority: medium
+state_hub_task_id: "33591917-3099-415f-8d9b-70c5273cf3b3"
+```
+
+Document provider setup, skip behavior, and remaining live conformance gaps.
+
+## Acceptance Criteria
+
+- Standard `make test` skips live Postgres tests unless env vars are present.
+- Live tests fail closed if a DSN is supplied without reset acknowledgement.
+- Provider cleanup only touches `user_engine_*` tables created by the bootstrap.
+- The live suite can use either `psycopg` or `psycopg2` if available.
+- Documentation names the required env vars and remaining follow-up work.
+
+## Expected Outputs
+
+- `user_engine.testing.postgres_provider` helper.
+- Env-gated provider-backed tests.
+- Durable-store documentation updates.
+
+## Implementation Notes
+
+Implemented on 2026-06-16:
+
+- Added optional provider connection and reset helpers for live Postgres tests.
+- Added env-gated tests for migration readiness, shared conformance, and
+  deterministic upsert uniqueness.
+- Hardened adapter readiness before migration when provider tables do not yet
+  exist.
+- Kept the default unit suite dependency-free and infrastructure-free.