Compare commits

...

10 commits

Author SHA1 Message Date
0616d5354d Add Forgejo CI smoke workflow (enablement template)
All checks were successful
CI Smoke / host-smoke (push) Successful in 0s
CI Smoke / container-smoke (push) Successful in 2s
2026-07-08 12:38:49 +02:00
b5b5a25370 Draft capability entry (reuse-surface REUSE-WP-0017-T04, cohort 3)
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>
2026-07-06 19:50:54 +02:00
9c6ad74f6b Normalize agent instructions and workplan frontmatter (STATE-WP-0067)
- Align agent files with on-disk workplan prefixes (infer from workplan ids)
- Set workplan domain to registered domain_slug; add topic_slug where applicable
- Repair frontmatter delimiter formatting; migrate legacy task status literals
- Regenerate AGENTS.md, CLAUDE.md, and .claude/rules from State Hub templates
2026-06-22 23:16:28 +02:00
3544a1b9d6 Add .repo-classification.yaml (CUST-WP-0050 T11 agent first-pass) 2026-06-22 17:47:43 +02:00
d13bc3ad8a chore: sync user wp 0019 statehub ids 2026-06-16 07:34:49 +02:00
a1692c62e3 test: add provider postgres conformance 2026-06-16 07:33:34 +02:00
1f2ac6666f chore: sync user wp 0018 statehub ids 2026-06-16 07:16:42 +02:00
0d50ad294d feat: add postgres user engine store 2026-06-16 07:14:37 +02:00
c494511a2e chore: sync user wp 0017 statehub ids 2026-06-16 03:45:31 +02:00
6810d9a3aa feat: add durable store record serialization 2026-06-16 03:43:55 +02:00
45 changed files with 2572 additions and 24 deletions

20
.claude/rules/agents.md Normal file
View file

@ -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.

View file

@ -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

View file

@ -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`

View file

@ -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 13 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 -->

View file

@ -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/
-->

View file

@ -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

View file

@ -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.

View file

@ -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)
```

View file

@ -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 -->

View file

@ -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}"

16
.repo-classification.yaml Normal file
View file

@ -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

219
AGENTS.md Normal file
View file

@ -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/`)

12
CLAUDE.md Normal file
View file

@ -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

View file

@ -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, idempotent migration, core save/read/query behavior, transaction rollback,
outbox ordering, and diagnostics that expose counts without raw factor or outbox ordering, and diagnostics that expose counts without raw factor or
profile values. 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.

View file

@ -10,13 +10,14 @@ tested immediately in local and agent environments.
```text ```text
src/user_engine/ 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 domain/ transport- and persistence-neutral domain schemas
errors.py typed service exceptions for callers and future transports errors.py typed service exceptions for callers and future transports
migrations.py ordered durable-store migration manifest migrations.py ordered durable-store migration manifest
ports.py adapter protocols for identity, authorization, events, audit, ports.py adapter protocols for identity, authorization, events, audit,
membership export, application bindings, and secrets membership export, application bindings, and secrets
service.py headless service API for the isolated MVP 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 testing/ local fixtures for tests and examples
tests/ standard-library unittest suite tests/ standard-library unittest suite
``` ```
@ -36,6 +37,19 @@ The command runs:
PYTHONPATH=src python3 -m unittest discover -s tests -p 'test_*.py' 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 ## Implementation Rule
Add new behavior in this order: Add new behavior in this order:

View file

@ -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. reusable harness that future adapters can run with their own store factory.
The standard local suite runs that harness against `InMemoryUserEngineStore`. 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: 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, outbox
- Add provider-backed conformance tests for locking, uniqueness races, claiming, redacted diagnostics, and restore validation.
migration readiness, outbox claiming, redacted diagnostics, and restore
validation.
- Add conformance tests that run against both in-memory and Postgres stores.
- Integrate the adapter with the future NetKingdom Postgres provider repo. - Integrate the adapter with the future NetKingdom Postgres provider repo.

View file

@ -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`

View file

@ -1,4 +1,19 @@
version: 1 version: 1
updated: '2026-06-16' updated: '2026-07-06'
domain: helix_forge 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

View file

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

View file

@ -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()
)

View file

@ -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}

View file

@ -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()

View file

@ -89,6 +89,11 @@ def assert_user_engine_store_conformance(
_assert_diagnostics_contract(testcase, store_factory) _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: def _assert_readiness_contract(testcase: TestCase, store_factory: StoreFactory) -> None:
store = store_factory() store = store_factory()
if store.schema_version is None: if store.schema_version is None:

View file

@ -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()

View file

@ -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()

View file

@ -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()

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

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

View file

@ -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.

View file

@ -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.

View file

@ -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.