feat(repositories): table repositories for the one-owner data model (XWING-2363)

.agents/skills/traust-core-patterns/SKILL.md

@@ -9,8 +9,8 @@ Rules live in `AGENTS.md` → **Design rules**: one owner per capability, one se
 
 | Adding | Recipe | Worked example |
 |---|---|---|
-| An aggregate repository over a contracts table | [repository.md](repository.md#aggregate-repository) | `src/traust_core/v1/repositories/triage.py` (mechanics only; legacy ids) |
-| A table and its repository | [repository.md](repository.md) | `tests/v1/example/repository.py`; over a contracts table: `src/traust_core/v1/repositories/storage.py` |
+| A repository (or a query) over a contracts table | [repository.md](repository.md) | `src/traust_core/v1/repositories/traust/inventory/repository.py`; tests in `tests/v1/repositories/tables/` |
+| Cross-section writes in one transaction | [repository.md](repository.md#unit-of-work) | `src/traust_core/v1/repositories/contract.py` (`SqlTraustUnitOfWork`) |
 | A domain service, policy or interface | [service.md](service.md) | `src/traust_core/v1/services/portfolio.py` (shape) |
 | A tool, source or feed provider | [provider.md](provider.md) | `tests/v1/example/provider.py` |
 | An artifact type the harness/LLM produces | [artifacts.md](artifacts.md) | `tests/v1/example/artifact_publishing.py` (`TRIAGE`) |

.agents/skills/traust-core-patterns/contracts.md

@@ -10,16 +10,17 @@ Bump the pin:
 Write (exact):
 
 - `contracts.validate(name, document)` / `contracts.validate_json(name, payload)` → `DocumentError` with `{path, message}` issues
-- `ArtifactPublisher` and `record_artifact` already call it
+- `ArtifactPublisher` already calls it
 
 Read:
 
-- Named artifact types live in `src/traust_core/v1/models/artifacts.py`; add one line per new artifact: `class XArtifact(Artifact, name="x", schema="x", kind=ResultKind.X): ...`
+- Named artifact types live in `src/traust_core/v1/models/artifacts.py`; add one line per new artifact: `class XArtifact(ArtifactDocument, name="x", schema="x", kind=ResultKind.X): ...`
 - Fields come from the installed schema; never declare them by hand
 - To see what fields exist: `XArtifact.describe()` (types, required/optional) or open `XArtifact.schema_path()`
 - `ctx.analysis_results().read(XArtifact, subject)` / `.read_all(XArtifact, tree)`; `XArtifact.parse(raw)` when you have bytes
 - Enums: `contracts.enum("<enum-file>")` (e.g. `Verdict`, `Severity` in the catalog)
 
 Database:
 
-- `apply_schema(engine, contracts.ddl("storage", dialect))`; check repositories with `schema_drift`
+- `apply_schema(engine, contracts.database_ddl(dialect))` creates every namespace (traust, traust_ops, traust_ledger) in bootstrap order
+- The DDL is contracts' to own and test; core declares each table once as a `TableSpec` and tests repository behaviour (see [repository.md](repository.md))

.agents/skills/traust-core-patterns/repository.md

@@ -1,39 +1,39 @@
 # Repository
 
-1. Model in `models/<subject>.py`: `Entity`/`Model` subclass or frozen dataclass. No database types.
-2. Interface: a `Protocol` over one aggregate, in `repositories/<subject>.py`, using the shared verbs only as needed: `get(id) -> T` (raises `NotFoundError`), `find_*(...) -> T | None`, `list(filter, page) -> Page[T]`, `add(T) -> T` and `update(T) -> T` returning the saved record (database-generated `id`, `created_at`, `updated_at` via `RETURNING`). IDs are `UUID`, named `<entity>_id`. No generic CRUD base, no cross-domain tables.
-3. Unit-of-work protocol: `class XUnitOfWork(UnitOfWork, Protocol): xs: XRepository`.
-4. SQL implementation:
-   - `sa.Table(...)` describing the existing table (DDL is owned by traust-contracts)
-   - `class SqlXRepository(SqlRepository)`; every statement goes through `self._execute("<operation> <id>", stmt)`
-   - `_to_row(model)` / `_to_model(row)` private functions; rows never leave the repository
-   - missing row → `NotFoundError`; duplicates surface as `ConflictError` from `_execute`
-5. `class SqlXUnitOfWork(SqlUnitOfWork)` attaching the repository in `_open_repositories`.
-6. In-memory double implementing the same interface, raising the same errors.
-7. Contract tests parametrized over `["memory", *SQL_BACKENDS]` using `tests/v1/conftest.py::database`. The Postgres case is marked `integration` and runs with `make db-up && make test-integration`.
-
-The domain service takes the unit of work in its constructor and opens it per use case:
+One repository per contract table. The DDL is owned by traust-contracts; you never write DDL, and a table that isn't in the pinned contracts yet is a gap to report, not one to invent.
+
+## Add a table repository
+
+Files mirror the contracts namespaces: `repositories/traust/<section>/<table>.py`, `repositories/traust_ops/<table>.py`, `repositories/traust_ledger/<table>.py`. Copy `repositories/traust/inventory/repository.py`.
+
+1. **Model** in `models/<section>.py`: subclass `Entity` (mutable table) or `AppendOnly` (append-only table); fields are the columns, named as in the DDL; no behaviour, no database types.
+2. **Spec**: `TableSpec(TRAUST | TRAUST_OPS | TRAUST_LEDGER, "<table>", lambda: (<domain columns>), append_only=...)` using `uuid_column` / `text_column` / `timestamp_column` (or `sa.Column` for booleans, integers, numerics). Base columns come from the spec; don't declare them. `TableFactory` builds it per dialect (Postgres schema or SQLite `ops_`/`ledger_` prefix).
+3. **Protocol**: `class XRepository(MutableRecordRepository[X] | RecordRepository[X], Protocol)` listing only this table's queries.
+4. **SQL class**: `class SqlXRepository(SqlMutableRepository[X] | SqlAppendOnlyRepository[X])` with `spec = X_SPEC`, `record = X`. `add`/`get`/`update` are inherited; each query uses `self._one(...)` / `self._many(..., order_by=...)` over `self.table.c`. Missing row → `NotFoundError`; database refusals surface as `ConflictError` from `_execute`.
+5. **Register**: add the spec to the section `__init__.py` `TABLES`, the repository to its section unit-of-work protocol, and the attribute to `SqlTraustUnitOfWork` in `repositories/contract.py`.
+6. **Seed + tests**: add `new_<table>()` / `<table>()` to `Seed` in `tests/v1/repositories/tables/conftest.py` — the per-table CRUD tests then cover it automatically — and a test per query in the section's test module.
+
+New reads go on the owning table's module: no duplicate queries of the same data, no speculative helpers, never reshape a table to fit a caller.
+
+## Unit of work
+
+Services declare the narrow protocol they need (`InventoryUnitOfWork`, `FindingsUnitOfWork`, …, `TraustUnitOfWork` across sections). `SqlTraustUnitOfWork(engine)` opens every repository on one connection, so one `commit()` or `rollback()` covers every section. Services that write together share one instance.
 
 ```python
-class TaskService:
-    def __init__(self, uow: TaskUnitOfWork) -> None:
+class FindingIntake:
+    def __init__(self, uow: TraustUnitOfWork) -> None:
         self._uow = uow
 
-    def complete(self, task_id: UUID) -> Task:
+    def accept(self, assessment_id: UUID, candidates: Sequence[FindingCandidate]) -> IntakeResult:
         with self._uow as uow:
-            done = uow.tasks.update(uow.tasks.get(task_id).complete())  # rule on the model
-            uow.commit()
-        return done
+            ...  # occurrences, findings, decisions
+            uow.commit()  # all or nothing
 ```
 
-## Aggregate repository
+## Tests
 
-One repository per aggregate (the thing the domain asks questions about), not per table. `repositories/triage.py` shows the SQL mechanics (table, mapping, `schema_drift`, idempotency); its string ids and `record_triage` free function are legacy and get replaced in the Findings slice — copy the mechanics, not the shape.
+Behaviour only, in `tests/v1/repositories/tables/`, each test running on SQLite and Postgres (`make test-repositories`; Postgres needs `make db-up`). Use the `uow` and `seed` fixtures; a test states only what it is about: `seed.decision(seed.finding(), severity="high")`.
 
-1. Domain model in `models/<subject>.py` (`TriageVerdict` in `models/findings.py`): fields = the projection's columns, typed.
-2. Protocol with the domain questions only (`for_binding`, `for_source_finding`, `true_positives`); no generic CRUD.
-3. `sa.Table` describing the contracts table (DDL from traust-contracts); match JSON columns' encoding to existing writers (`JSON(none_as_null=True)`, compact serializer from `create_database_engine`).
-4. SQL + in-memory implementations; a unit of work that exposes it alongside any repository it must write with (`evidence` + `triage_verdicts`).
-5. Domain service writes in one transaction: `bind_artifact(...)` then the projection, then `commit()`.
-6. Tests: contract tests on memory/sqlite (+ Postgres `integration`), `schema_drift` against the DDL, idempotency.
-7. Files go by role then subject: `models/<subject>.py`, `repositories/<subject>.py`, `services/<subject>.py`; tests mirror under `tests/v1/<role>/`.
+- Per query: the right rows come back (and `None` / `[]` when they shouldn't).
+- Assert list order only where the query defines it (`occurred_at`, `seq`, `priority`); otherwise compare with `same_rows`.
+- Never restate the DDL (columns, constraints, triggers, indexes): contracts owns and tests that.

AGENTS.md

@@ -7,14 +7,14 @@ Python 3.11+ library. The platform API lives in `traust_core.v1`. Overview: [REA
 - **One owner per capability.** Never implement a capability twice. Inconsistent data always means a second owner: remove it, never sync it.
 - **Less is the goal.** One service per domain owns its use cases (reads and writes). Split a service only when it holds two responsibilities (e.g. `services/findings/`). No speculative helpers, no generic bases.
 - **Place behaviour by responsibility:**
-  - model: rules about one entity's own state (`Disposition.admit`, `Repository.risk_tier`)
+  - model: what a row is (one model per contract table, fields = columns) and rules about one value's own state (`Disposition.admit`); table models carry no behaviour
   - policy: rules needing config or spanning domains (`RatingPolicy`, `IdentityPolicy`)
   - service: use-case orchestration
-  - repository: persistence, reads and writes, its own domain's tables only
+  - repository: one per contract table — save, get, update and the named queries on that table
   - provider: anything outside the process (tools, forges, feeds, ledger, identity)
 - **Domain boundaries.** A service never touches another domain's repositories; it calls that domain's service. Several repositories of your own domain in one service are fine — the line is whose repositories, not how many. Repositories never join across domains. Callers compose across domains by typed ID; composite reads are their own query object (`RepoStateQuery`).
-- **The database owns identity and time.** Every record has `id` (UUIDv4), `created_at`, `updated_at`, generated by the database on insert (`RETURNING`); on update the contracts `<table>_touch` trigger keeps `updated_at` — the app never sets it (no SQLAlchemy `onupdate`). On SQLite, `uuid()` is registered by `create_database_engine`. Append-only tables (ledger events) stop at `id` (UUIDv4) and `created_at` — rows are never updated, so there is no `updated_at`. `id is None` = not saved yet; `add` returns the saved record. The app never mints IDs or timestamps. IDs are plain `UUID`, named `<entity>_id` (fields and parameters, matching the DDL's `<table>_id`); never a `str` ID, never a path as an ID, never a path in a signature. Don't re-declare `id` or add your own created/updated timestamps on an `Entity` or `AppendOnly`. Order comes from `created_at`, never from the id.
-- **Same shapes everywhere.** Repositories: `get`, `find_*`, `list`, `add`, `update` (`add`/`update` return the saved record). Skill doors: `prepare(...) -> <X>Input` and `record(doc, provenance)` (`@overload` per document). Rollups: `summary(filter, group_by)`; series: `timeline(filter, bucket)`. Deterministic document checks: `Lint`.
+- **The database owns identity and time.** Every record has `id` (UUIDv4), `created_at`, `updated_at`, generated by the database on insert (`RETURNING`); on update the contracts `<table>_touch` trigger keeps `updated_at` — the app never sets it (no SQLAlchemy `onupdate`). On SQLite, `uuid()` is registered by `create_database_engine`. Append-only tables (ledger events) stop at `id` (UUIDv4) and `created_at` — rows are never updated, so there is no `updated_at`. `id is None` = not saved yet; `add` returns the saved record. The app never mints IDs or timestamps. IDs are plain `UUID`, named `<entity>_id` (fields and parameters, matching the DDL's `<table>_id`); never a `str` ID, never a path as an ID, never a path in a signature. Don't re-declare `id` or add your own created/updated timestamps on an `Entity` or `AppendOnly`. Order comes from `created_at` (stamped per statement on both engines); ties break by `id`, which carries no order.
+- **Same shapes everywhere.** Repositories: `get`, `find_*`, `list_*`, `add`, `update` (`add`/`update` return the saved record; append-only tables have no `update`). Skill doors: `prepare(...) -> <X>Input` and `record(doc, provenance)` (`@overload` per document). Rollups: `summary(filter, group_by)`; series: `timeline(filter, bucket)`. Deterministic document checks: `Lint`.
 - **Services return domain results.** `JobResult` exists only at the engine/CLI adapter boundary.
 - **One code path.** Solo (`traust …`, in-process), small group (`traust serve --with-engine`, one process) and scale (REST API, engine and runners as separate processes) run the same services; only the transport (in-process or REST) and the storage (SQLite + dir, or Postgres + object storage) differ. Never add a mode-specific branch of domain logic; the REST API holds none. CI runs service tests in-process and through the REST adapter, plus one end-to-end skill on SQLite with no server
 - **Before adding a symbol, find the existing owner.** Extend it; if it truly can't own the new thing, say why in the PR. `grep -rn "NEED(" src` first.
@@ -45,7 +45,7 @@ Python 3.11+ library. The platform API lives in `traust_core.v1`. Overview: [REA
 
 - `errors` → `contracts` → `models` → `providers` → `repositories` → `rendering` → `services` → `context`; each imports only earlier ones (`tests/v1/test_layering.py`)
 - One question per folder: `models` what it is and the rules about its own state, `providers` how we reach outside (tools, sources, feeds, process, clock), `repositories` how it is stored, `services` what we do with it (use cases + pure policies), `context` how it is wired
-- Organised by role, then subject: `models/<subject>.py`, `repositories/<subject>.py`, `services/<subject>.py`; a subject becomes a package only when it needs several modules (`services/findings/`)
+- Organised by role, then subject: `models/<section>.py` (table models + that section's service types), `repositories/traust/<section>/<table>.py`, `repositories/traust_ops/<table>.py`, `repositories/traust_ledger/<table>.py` (mirroring the contracts namespaces), `repositories/objects/` (bytes), `services/<subject>.py`; a service becomes a package only when it needs several modules (`services/findings/`)
 - `models/base.py`: `Model` (strict, internal), `Dto` (tolerant, boundary data), `Entity` (`id`, `created_at`, `updated_at` — `None` until the database saves it), `AppendOnly` (`id`, `created_at` — rows are never updated)
 - Tests mirror `src`: `tests/v1/<role>/test_<subject>.py`
 - Pack policy (routing rules, cadences, lanes, skills, CLIs) stays in traust
@@ -65,8 +65,8 @@ Python 3.11+ library. The platform API lives in `traust_core.v1`. Overview: [REA
 - Nothing from traust-contracts is copied or generated into this repo
 - Artifacts are named types in `models/artifacts.py` (`TriageArtifact`, ...); their fields come from the installed schema, never a hand-written copy
 - Write with `ArtifactPublisher` (exact schema gate); read with `ArtifactPublisher.get(ArtifactId)`. `AnalysisResultsRepository.read(XArtifact, subject)` is legacy (path-keyed) until the Artifacts slice. No `json.load`/`.get()`/`json.dump` on artifacts
-- Names: `models/artifacts.py` = what an artifact is; `repositories/object_store.py` = raw bytes; `services/artifact_publishing.py` = the publish use case. Modules name the subject; the package names the role (no `_service` suffix)
-- Every repository `sa.Table` over a contracts table has a `schema_drift` test
+- Names: `models/artifacts.py` = what an artifact document is (`ArtifactDocument` and its named types); `models/assessment.py::Artifact` = the `artifact` table row; `repositories/objects/store.py` = raw bytes; `services/artifact_publishing.py` = the publish use case. Modules name the subject; the package names the role (no `_service` suffix)
+- Core never tests the DDL: traust-contracts owns and tests its schema. Core tests repository behaviour against it
 - Artifacts produced by the harness or an LLM go through `ArtifactPublisher`; nothing writes `analysis-results` files directly
 - Protocols implemented outside core (`Materializer`, `Router` in `services/operations.py`) exist only for a decided pattern or backlog item; each gets an implementation in the repo that owns it
 - Work is a service. Services return domain results; a `Materializer` (for traust-engine) and a CLI command (for LLMs) are thin adapters that validate input, call one service and turn its result into a `JobResult`; no logic in either
@@ -96,9 +96,10 @@ Python 3.11+ library. The platform API lives in `traust_core.v1`. Overview: [REA
 - Unknown or unmeasured is `None`, never `0`
 
 **Structure**
-- Repository pattern: one repository per aggregate, written on SQLAlchemy Core; no ORM mapped classes, no generic CRUD base
+- Repository pattern: one repository per contract table: a `TableSpec` (columns once, built per dialect by `TableFactory`) and a `SqlTableRepository` subclass, on SQLAlchemy Core; no ORM mapped classes. The shared base gives only `add`/`get`/`update`; every other method is a named query on its table's module
 - Domain objects never know about the database; rows map to models only inside repositories
 - Services receive units of work, other services and providers through their constructor; they never open connections, build paths, read env vars or start processes
+- A service declares the narrow section protocol it needs (`InventoryUnitOfWork`, `FindingsUnitOfWork`, …, or `TraustUnitOfWork` when it spans sections); `SqlTraustUnitOfWork` satisfies them all on one connection, so services that must write together share one instance and one transaction
 - `Context` only at entry points; pass the specific dependency, not `Context`
 - No `subprocess` outside `ProcessRunner`, no clock reads outside `Clock`
 - Interfaces are `Protocol`s defined where they are consumed
@@ -109,7 +110,8 @@ Python 3.11+ library. The platform API lives in `traust_core.v1`. Overview: [REA
 - Wrap driver and library errors with `raise ... from e`; never swallow them
 
 **Tests**
-- Every repository, store and provider has contract tests that run against each implementation (memory, sqlite, Postgres when `TRAUST_TEST_DATABASE_URL` is set)
+- Table repositories: behaviour tests in `tests/v1/repositories/tables/` — save, get, update and each query — on SQLite and Postgres through the `uow`/`seed` fixtures (`make test-repositories`). No tests that restate the DDL
+- Stores and providers: contract tests that run against each implementation
 - Pure functions get table tests
 - Test names say the behaviour they protect
 - A ported legacy scenario that belongs to a later slice is still written now, in its owner's test module, marked `@pytest.mark.todo("<slice>", legacy="<repo>/tests/<file>::<test>")` (skipped until that slice removes the marker). Never drop a legacy scenario silently