feat(repositories): base record columns for mutable and append-only tables (XWING-2363)

AGENTS.md

@@ -13,7 +13,7 @@ Python 3.11+ library. The platform API lives in `traust_core.v1`. Overview: [REA
   - repository: persistence, reads and writes, its own domain's tables only
   - 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` (UUID), `created_at`, `updated_at`, generated by the database on insert (`RETURNING`). `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`.
+- **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`.
 - **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
@@ -46,7 +46,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/`)
-- `models/base.py`: `Model` (strict, internal), `Dto` (tolerant, boundary data), `Entity` (`id`, `created_at`, `updated_at` — `None` until the database saves it)
+- `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
 - traust-core is the hub: every other traust repo depends on it, it depends on none of them. One exception: the traust-ledger client, imported only inside the ledger provider adapter (behind `LedgerClient`). traust-contracts is data, read through `traust_core.v1.contracts`

README.md

@@ -8,12 +8,12 @@ All modules live under `traust_core.v1` (see [Versioning](#versioning)).
 |---|---|---|---|
 | **Contracts** | `contracts` | the pinned traust-contracts dependency, read as data: `validate(name, doc)` (exact JSON Schema gate), `schema()`, `enum_values()`, `normalize_enum(name, value)` (registry read view), `ddl()`, `storage_profile()` | skills remembering to run `reporting validate` |
 | **Errors** | `errors` | the exception hierarchy every layer raises (`ValidationError`, `NotFoundError`, …) | ad hoc exceptions |
-| **Models** | `models` | what things are and the rules about their own state: `base` (`Model` strict, `Dto` tolerant, `Entity` with a UUID id), `values` (`HttpsRepoUrl`, `GitSha`, `CveId`), `integrity`, `storage`, `operations` (`JobResult`, `AssetRequest`, `Order`), and one module per subject (`findings`, `portfolio`, …) | dicts passed between scripts |
+| **Models** | `models` | what things are and the rules about their own state: `base` (`Model` strict, `Dto` tolerant, `Entity` with a database-generated UUIDv4 id, `AppendOnly` for tables that are never updated), `values` (`HttpsRepoUrl`, `GitSha`, `CveId`), `integrity`, `storage`, `operations` (`JobResult`, `AssetRequest`, `Order`), and one module per subject (`findings`, `portfolio`, …) | dicts passed between scripts |
 | **Artifacts** | `models.artifacts` | the `Artifact` base + `SchemaView` (fields from the installed schema, `doc.findings[0].verdict`, unknown names raise) and the named documents (`TriageArtifact`, `SecurityAuditArtifact`, …, `Verdict`, `Severity`) | `json.load` + `.get("key")` |
 | **Artifact publishing** | `services.artifact_publishing` | `ctx.artifact_publisher(specs).publish(name, subject, raw, run_id)` (LLM) or `.publish_artifact(artifact, subject, run_id)` (code): exact gate → files (always) → index (when `database.url` is set) | writing `analysis-results` files by hand |
 | **Analysis results** | `repositories.analysis_results` | `ctx.analysis_results().read(TriageArtifact, subject)` / `.read_all(TriageArtifact, tree)`; raw `put/get/find(subject, kind)`; today's naming convention in one place | `json.load` + `"-security-audit.json"` path building |
 | **Storage records** | `repositories.storage`, `services.storage` | `record_artifact()` / `get_binding()` over contracts storage/v1 tables | contracts `Store.ingest/get_binding` |
-| **Records (pattern)** | `repositories.sql` | `SqlUnitOfWork` + one `SqlRepository` per aggregate on SQLAlchemy Core; sqlite or Postgres by URL | `sqlite3.connect`, inline SQL |
+| **Records (pattern)** | `repositories.sql` | `SqlUnitOfWork` + one `SqlRepository` per aggregate on SQLAlchemy Core; sqlite or Postgres by URL. `record_columns()` (mutable: UUIDv4 id + `created_at` + `updated_at`) / `append_only_columns()` (UUIDv4 id + `created_at`, no `updated_at`) + `_insert_record`/`_update_record`, database-generated via `RETURNING`; `updated_at` kept by the contracts touch trigger, SQLite `uuid()` registered by `create_database_engine` | `sqlite3.connect`, inline SQL, client-minted ids/timestamps |
 | **Object store** | `repositories.object_store` | `ObjectStore.put/get/list` raw bytes by key, digest-verified (local dir; S3 later) | `write_text`, `json.dump` |
 | **Rendering** | `rendering` | `Report` → `MarkdownRenderer` → `publish()`; `markdown_table()` | 18 copies of `render_md` |
 | **Jobs** | `models.operations`, `services.operations` | `Materializer.materialize(AssetRequest) -> JobResult` (succeeded · degraded · refused · failed): the seam traust-engine calls; `Router` for continuous-ops `route()` | each CLI inventing its own exit codes |

src/traust_core/v1/models/base.py

@@ -15,7 +15,14 @@ class Dto(pydantic.BaseModel):
 
 
 class Entity(Model):
-    # None until saved: the database generates id and both timestamps on insert.
+    # None until saved: the database generates id and both timestamps; on update
+    # the contracts touch trigger keeps updated_at.
     id: UUID | None = None
     created_at: datetime | None = None
     updated_at: datetime | None = None
+
+
+class AppendOnly(Model):
+    # None until saved; append-only rows are never updated, so there is no updated_at.
+    id: UUID | None = None
+    created_at: datetime | None = None

src/traust_core/v1/repositories/__init__.py

@@ -9,8 +9,10 @@
 from traust_core.v1.repositories.sql import (
     SqlRepository,
     SqlUnitOfWork,
+    append_only_columns,
     apply_schema,
     create_database_engine,
+    record_columns,
     schema_drift,
 )
 from traust_core.v1.repositories.unit_of_work import UnitOfWork
@@ -25,7 +27,9 @@
     "SqlRepository",
     "SqlUnitOfWork",
     "UnitOfWork",
+    "append_only_columns",
     "apply_schema",
     "create_database_engine",
+    "record_columns",
     "schema_drift",
 ]

src/traust_core/v1/repositories/sql.py

@@ -2,15 +2,17 @@
 
 import json
 from collections.abc import Iterable, Mapping, Sequence
-from datetime import datetime
+from datetime import UTC, datetime
 from types import TracebackType
 from typing import Any, Literal, Self
+from uuid import UUID, uuid4
 
 import sqlalchemy as sa
 from sqlalchemy.engine import Connection, CursorResult, Engine
 from sqlalchemy.exc import IntegrityError, SQLAlchemyError
+from sqlalchemy.ext.compiler import compiles
 
-from traust_core.v1.errors import ConflictError, RepositoryError
+from traust_core.v1.errors import ConflictError, NotFoundError, RepositoryError
 
 STORAGE_SCHEMA = "traust_storage"
 LEDGER_SCHEMA = "traust_ledger"
@@ -26,8 +28,10 @@ def create_database_engine(url: str | sa.URL) -> Engine:
         return engine
 
     @sa.event.listens_for(engine, "connect")
-    def _fk_on(dbapi_conn: Any, _: Any) -> None:
+    def _sqlite_functions(dbapi_conn: Any, _: Any) -> None:
         dbapi_conn.execute("PRAGMA foreign_keys = ON")
+        # Contracts' uuid(): ext/misc/uuid.c in other drivers, registered here.
+        dbapi_conn.create_function("uuid", 0, lambda: str(uuid4()))
 
     return engine.execution_options(
         schema_translate_map={STORAGE_SCHEMA: None, LEDGER_SCHEMA: None}
@@ -46,10 +50,84 @@ def load_dialect_impl(self, dialect: sa.Dialect) -> sa.TypeEngine[Any]:
         return dialect.type_descriptor(sa.Text() if dialect.name == "sqlite" else self.impl)
 
     def process_bind_param(self, value: datetime | None, dialect: sa.Dialect) -> Any:
-        return value.isoformat() if value is not None and dialect.name == "sqlite" else value
+        if value is None or dialect.name != "sqlite":
+            return value
+        # Contracts' TEXT format (millisecond precision, Z) so ordering survives mixed writers.
+        if value.tzinfo is None:
+            value = value.replace(tzinfo=UTC)
+        stamp = value.astimezone(UTC)
+        return stamp.strftime("%Y-%m-%dT%H:%M:%S") + f".{stamp.microsecond // 1000:03d}Z"
 
     def process_result_value(self, value: Any, dialect: sa.Dialect) -> datetime | None:
-        return datetime.fromisoformat(value) if isinstance(value, str) else value
+        if isinstance(value, str):
+            value = datetime.fromisoformat(value)
+        if value is not None and value.tzinfo is None:
+            value = value.replace(tzinfo=UTC)  # SQLite CURRENT_TIMESTAMP is UTC without offset
+        return value
+
+
+class RecordUuid(sa.TypeDecorator[UUID]):
+    # Native UUID on Postgres; on SQLite the hyphenated text uuid() writes,
+    # so lookups match the stored form.
+    impl = sa.Uuid
+    cache_ok = True
+
+    def load_dialect_impl(self, dialect: sa.Dialect) -> sa.TypeEngine[Any]:
+        return dialect.type_descriptor(sa.Text() if dialect.name == "sqlite" else self.impl)
+
+    def process_bind_param(self, value: UUID | None, dialect: sa.Dialect) -> Any:
+        return str(value) if value is not None and dialect.name == "sqlite" else value
+
+    def process_result_value(self, value: Any, dialect: sa.Dialect) -> UUID | None:
+        return UUID(value) if isinstance(value, str) else value
+
+
+class _GeneratedUuid(sa.FunctionElement[UUID]):
+    name = "generated_uuid"
+    inherit_cache = True
+
+
+@compiles(_GeneratedUuid, "postgresql")
+def _postgres_uuid(element: Any, compiler: Any, **kw: Any) -> str:
+    return "gen_random_uuid()"
+
+
+@compiles(_GeneratedUuid, "sqlite")
+def _sqlite_uuid(element: Any, compiler: Any, **kw: Any) -> str:
+    return "uuid()"
+
+
+class _DbNow(sa.FunctionElement[datetime]):
+    name = "db_now"
+    inherit_cache = True
+
+
+@compiles(_DbNow, "postgresql")
+def _postgres_now(element: Any, compiler: Any, **kw: Any) -> str:
+    return "now()"
+
+
+@compiles(_DbNow, "sqlite")
+def _sqlite_now(element: Any, compiler: Any, **kw: Any) -> str:
+    # Contracts' TEXT timestamp format, millisecond precision.
+    return "strftime('%Y-%m-%dT%H:%M:%fZ','now')"
+
+
+def record_columns() -> tuple[sa.Column[UUID], sa.Column[datetime], sa.Column[datetime]]:
+    """Mutable base columns: UUIDv4 id; the touch trigger keeps updated_at."""
+    return (
+        sa.Column("id", RecordUuid, primary_key=True, server_default=_GeneratedUuid()),
+        sa.Column("created_at", UtcTimestamp, nullable=False, server_default=_DbNow()),
+        sa.Column("updated_at", UtcTimestamp, nullable=False, server_default=_DbNow()),
+    )
+
+
+def append_only_columns() -> tuple[sa.Column[UUID], sa.Column[datetime]]:
+    """Append-only base columns: UUIDv4 id and created_at, no updated_at."""
+    return (
+        sa.Column("id", RecordUuid, primary_key=True, server_default=_GeneratedUuid()),
+        sa.Column("created_at", UtcTimestamp, nullable=False, server_default=_DbNow()),
+    )
 
 
 def apply_schema(engine: Engine, statements: Iterable[str]) -> None:
@@ -111,6 +189,42 @@ class SqlRepository:
     def __init__(self, connection: Connection) -> None:
         self._connection = connection
 
+    def _insert_record(
+        self, operation: str, table: sa.Table, values: Mapping[str, Any]
+    ) -> dict[str, UUID | datetime]:
+        """INSERT domain columns; returns the database-generated base columns the table has."""
+        returning = [table.c.id, table.c.created_at]
+        if "updated_at" in table.c:
+            returning.append(table.c.updated_at)
+        stmt = sa.insert(table).values(**values).returning(*returning)
+        return dict(self._execute(operation, stmt).mappings().one())
+
+    def _update_record(
+        self, operation: str, table: sa.Table, record_id: UUID, values: Mapping[str, Any]
+    ) -> dict[str, UUID | datetime]:
+        """UPDATE by id; returns the saved base columns (the touch trigger keeps updated_at)."""
+        stmt = (
+            sa.update(table)
+            .where(table.c.id == record_id)
+            .values(**values)
+            .returning(table.c.id, table.c.created_at, table.c.updated_at)
+        )
+        row = self._execute(operation, stmt).mappings().first()
+        if row is None:
+            raise NotFoundError(f"{type(self).__name__}: {operation}: not found")
+        saved = dict(row)
+        if self._connection.dialect.name == "sqlite":
+            # SQLite's AFTER UPDATE trigger fires after RETURNING is computed.
+            saved["updated_at"] = (
+                self._execute(
+                    f"reread {record_id}",
+                    sa.select(table.c.updated_at).where(table.c.id == record_id),
+                )
+                .scalars()
+                .one()
+            )
+        return saved
+
     def _execute(
         self,
         operation: str,

tests/v1/conftest.py

@@ -7,9 +7,10 @@
 from pathlib import Path
 
 import pytest
+import sqlalchemy as sa
 from sqlalchemy.engine import Engine
 
-from tests.v1.example.repository import metadata
+from tests.v1.example import repository as example_repository
 from traust_core.v1 import contracts
 from traust_core.v1.providers.clock import FixedClock
 from traust_core.v1.repositories import apply_schema, create_database_engine
@@ -31,7 +32,9 @@ def _engine(kind: str, tmp_path: Path) -> Engine:
 
 
 @contextmanager
-def database(kind: str, tmp_path: Path) -> Generator[Engine]:
+def database(
+    kind: str, tmp_path: Path, metadata: sa.MetaData = example_repository.metadata
+) -> Generator[Engine]:
     eng = _engine(kind, tmp_path)
     metadata.create_all(eng)
     try: