chore(guard-rails): architecture tests, CODEOWNERS, PR template, patterns in the core skill

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

@@ -20,6 +20,38 @@ Rules live in `AGENTS.md` → **Design rules**: one owner per capability, one se
 | A config field | [service.md](service.md#config) | `src/traust_core/v1/context/` |
 | A contracts bump or a named artifact | [contracts.md](contracts.md) | `src/traust_core/v1/models/artifacts.py`, `tests/v1/models/test_artifacts.py` |
 
+## Who owns what
+
+| Layer | Owns | Never |
+|---|---|---|
+| Model (`models/<section>.py`) | a contract table row (`Entity`/`AppendOnly`, fields = columns); other types are plain `Model`; pure rules on one value | I/O, database types, behaviour on row models |
+| Repository (`repositories/traust*/<section>/<table>.py`) | SQL for its one table: inherited `add`/`get`/`update` + named queries | rules, joins to another section, files |
+| Unit of work (`SqlTraustUnitOfWork`) | one connection, one transaction across every repository | logic |
+| Policy | pure rules needing config or spanning domains | I/O |
+| Service (`services/<domain>.py`) | use cases: read via repositories, apply policies, write, commit | SQL, another domain's repositories, files, env, clocks |
+| Provider (`providers/`) | anything outside the process | domain rules |
+| Route / CLI / materializer | validate → call one service → return its result | logic |
+
+```text
+needs SQL?                      → named query on the table's repository (one table)
+several tables of one section?  → service method over that section's repositories
+two domains?                    → call the other domain's service; compose by <entity>_id
+rule with no I/O?               → policy (or the model, if it's about one value)
+outside the process?            → provider
+SQL rule several readers share? → a view in traust-contracts
+```
+
+## No duplicated responsibility
+
+- One query, one place: the owning table's repository. Filtering rows in Python that SQL could filter → a named query.
+- A repository never decides (thresholds, status rules) → policy.
+- One model per concept: extend it; never a parallel `ObservedX` / flattened copy of a DTO.
+- One id scheme: `<entity>_id: UUID`.
+- No `Protocol` without a second implementation.
+- When a change touches a module that duplicates a responsibility, move it to its owner in the same PR and delete the copy.
+
+`tests/v1/test_architecture.py` enforces the structural rules (row models are tables, services use their own section, no file/sqlite access in services or table repositories, no side repositories, no in-memory stand-ins in `src/`, one engine in `Context`). If it fails, the code is in the wrong place; don't widen the allowlists without review.
+
 Python style (typing, models, errors, idioms): [../python-patterns/SKILL.md](../python-patterns/SKILL.md).
 
 Finish every change with `make lint-fix && make test`.

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

@@ -5,18 +5,22 @@ Rules: `AGENTS.md` → Design rules. Shape to copy: any domain service, e.g. `se
 A domain service:
 
 1. One class per domain in `services/<subject>.py` (a package only when the domain holds several responsibilities, e.g. `services/findings/`). Before adding a method, check no other service already owns the capability.
-2. Constructor takes its unit of work, the other domains' **services** it needs (never their repositories), providers and `Clock`.
+2. Constructor takes its own section's unit-of-work protocol (`InventoryUnitOfWork`, …; `TraustUnitOfWork` only for use cases that write several sections, listed in `SPANS_SECTIONS` in `tests/v1/test_architecture.py`), the other domains' **services** it needs (never their repositories), providers and `Clock`.
 3. Methods take typed IDs, commands or contracts artifacts and return domain models or `Page[...]` — never `dict`, never `JobResult`.
 4. Skill doors: `prepare(...) -> <X>Input` (extends `SkillInput`) and `record(doc: <X>Artifact, provenance: Provenance)`; `@overload` when the domain records more than one document. Run the document's `Lint` rules inside `record`.
 5. Reads use the shared verbs: `get`, `find_*`, `list(filter, page)`, `summary(filter, group_by)`, `timeline(filter, bucket)`.
 6. Behaviour about one entity's own state goes on the model; rules needing config or several domains go in a policy class next to the service.
-7. Contract tests run the service against every unit-of-work implementation.
+7. Tests run the service over `SqlTraustUnitOfWork` with rows seeded through the `seed` fixture; assert behaviour, never the DDL.
 
 Interfaces implemented outside core (`Materializer`, `Router`):
 
 1. Request/response models in `models/<subject>.py`; the `Protocol` in the service module that consumes it.
 2. Implemented in the owning repo; export from `services/__init__.py`; README row.
 
+## Composition root
+
+`Context.services()` builds every service from **one** shared `SqlTraustUnitOfWork`, one line per service, collaborators passed explicitly. No wiring anywhere else, no second unit of work, no extra databases.
+
 ## Config
 
 1. Add a frozen `_Model` section in `context/config.py`; unknown keys stay rejected.

.github/CODEOWNERS

@@ -0,0 +1,9 @@
+# Structure of the platform API: models are the contract tables, repositories own their table,
+# context is the composition root. Either owner can approve.
+/src/traust_core/v1/models/        @smith-xyz @chubirka
+/src/traust_core/v1/repositories/  @smith-xyz @chubirka
+/src/traust_core/v1/context/       @smith-xyz @chubirka
+/tests/v1/test_architecture.py     @smith-xyz @chubirka
+/AGENTS.md                         @smith-xyz @chubirka
+/.agents/                          @smith-xyz @chubirka
+/pyproject.toml                    @smith-xyz @chubirka

.github/pull_request_template.md

@@ -0,0 +1,15 @@
+## What
+
+<!-- One domain, one slice. Which service methods / queries does this implement? -->
+
+## Checklist
+
+- [ ] Targets `main` (traust-core is pre-release; other repos target `development`)
+- [ ] Only this domain's service and its section's repositories changed; other domains reached through their services
+- [ ] No new tables, columns or row models (model gaps are raised, not worked around)
+- [ ] New SQL is a named query on the owning table's repository
+- [ ] No reads of files or databases outside the contract tables and `ObjectStore`; no in-memory stand-ins in `src/`
+- [ ] Duplicated responsibility touched by this change moved to its owner, copy deleted
+- [ ] Tests: behaviour on SQLite (+ Postgres for new queries) via `uow` / `seed`
+- [ ] Legacy code deleted only with a parity diff attached
+- [ ] `make lint-fix && make test` green

AGENTS.md

@@ -30,7 +30,7 @@ Python 3.11+ library. The platform API lives in `traust_core.v1`. Overview: [REA
 
 - First time in a clone: `make setup` (uv sync + git hooks via `core.hooksPath=.githooks`). Hooks: ruff on staged files (pre-commit), conventional subject (commit-msg), `make lint` + `make coverage` (pre-push). Don't bypass with `--no-verify`
 
-- `make lint-fix` then `make test`; both must pass. CI runs `make coverage`, which fails below 80% line+branch coverage (`fail_under` in `pyproject.toml`); stub bodies (`raise NotImplementedError`) are excluded, so new code must arrive with tests. Database changes also need `make db-up && make test-integration`
+- `make lint-fix` then `make test`; both must pass. `tests/v1/test_architecture.py` holds the placement rules; a failure there means the code is in the wrong layer, not that the allowlist needs widening. CI runs `make coverage`, which fails below 80% line+branch coverage (`fail_under` in `pyproject.toml`); stub bodies (`raise NotImplementedError`) are excluded, so new code must arrive with tests. Database changes also need `make db-up && make test-integration`
 - Tests that need Postgres are marked `@pytest.mark.integration`; `make test` excludes them
 - Commits: conventional `type(scope): subject`
 - Public repo: no deployment-specific hostnames or figures in code, tests or docs

tests/v1/test_architecture.py

@@ -0,0 +1,188 @@
+"""Where code may live. Each test names the rule it holds (AGENTS.md "Design rules")."""
+
+from __future__ import annotations
+
+import ast
+import importlib
+import inspect
+import pkgutil
+from pathlib import Path
+
+import pytest
+
+import traust_core.v1.models as models
+from traust_core.v1.models.base import AppendOnly, Entity
+from traust_core.v1.repositories.tables import SqlTableRepository
+
+SRC = Path(__file__).resolve().parents[2] / "src" / "traust_core" / "v1"
+REPOSITORIES = SRC / "repositories"
+SERVICES = SRC / "services"
+
+# The section each service writes. A service imports only its own section's unit of work.
+SERVICE_SECTION = {
+    "portfolio": "traust.inventory",
+    "onboarding": "traust.inventory",
+    "assessment": "traust.assessments",
+    "artifact_publishing": "traust.assessments",
+    "code_facts": "traust.assessments",
+    "pqc": "traust.assessments",
+    "compliance": "traust.assessments",
+    "detection_quality": "traust.assessments",
+    "attestation": "traust.findings",
+    "operations": "traust_ops",
+    "provider_catalog": "traust_ops",
+    "ledger": "traust_ledger",
+}
+# Use cases that write several sections in one transaction declare TraustUnitOfWork.
+SPANS_SECTIONS = {"findings", "exposure", "validation", "remediation", "insight"}
+
+REPOSITORY_PACKAGES = {"traust", "traust_ops", "traust_ledger", "objects"}
+REPOSITORY_MODULES = {"__init__", "contract", "memory", "sql", "tables", "unit_of_work"}
+IN_MEMORY_ALLOWED = {"InMemoryUnitOfWork", "InMemoryObjectStore"}
+FILE_ACCESS = {"open", "read_text", "read_bytes", "write_text", "write_bytes"}
+
+
+def _tree(path: Path) -> ast.Module:
+    return ast.parse(path.read_text())
+
+
+def _imported_modules(path: Path) -> set[str]:
+    found: set[str] = set()
+    for node in ast.walk(_tree(path)):
+        if isinstance(node, ast.ImportFrom) and node.module:
+            found |= {f"{node.module}.{alias.name}" for alias in node.names} | {node.module}
+        elif isinstance(node, ast.Import):
+            found |= {alias.name for alias in node.names}
+    return found
+
+
+def _service_key(path: Path) -> str:
+    rel = path.relative_to(SERVICES)
+    return rel.parts[0] if len(rel.parts) > 1 else rel.stem
+
+
+def _sql_table_repositories() -> set[type]:
+    import traust_core.v1.repositories as repositories
+
+    found: set[type] = set()
+    for info in pkgutil.walk_packages(repositories.__path__, "traust_core.v1.repositories."):
+        module = importlib.import_module(info.name)
+        for _, cls in inspect.getmembers(module, inspect.isclass):
+            if issubclass(cls, SqlTableRepository) and getattr(cls, "spec", None) is not None:
+                found.add(cls)
+    return found
+
+
+def _row_models() -> set[type]:
+    found: set[type] = set()
+    for info in pkgutil.walk_packages(models.__path__, "traust_core.v1.models."):
+        module = importlib.import_module(info.name)
+        for _, cls in inspect.getmembers(module, inspect.isclass):
+            own = cls.__module__ == module.__name__ and cls not in (Entity, AppendOnly)
+            if own and issubclass(cls, Entity | AppendOnly):
+                found.add(cls)
+    return found
+
+
+def test_every_row_model_is_a_contract_table() -> None:
+    """Entity/AppendOnly mean "a row of a contract table"; anything else is a plain Model."""
+    tables = {cls.record for cls in _sql_table_repositories()}
+    orphans = sorted(cls.__qualname__ for cls in _row_models() - tables)
+    assert not orphans, (
+        f"row models with no contract table (use Model, or raise a model gap): {orphans}"
+    )
+
+
+def test_repositories_hold_only_contract_tables_and_object_storage() -> None:
+    """No side repositories: tables live under traust*/, bytes under objects/."""
+    extra = sorted(
+        p.name
+        for p in REPOSITORIES.iterdir()
+        if p.name != "__pycache__"
+        and (
+            (p.is_dir() and p.name not in REPOSITORY_PACKAGES)
+            or (p.suffix == ".py" and p.stem not in REPOSITORY_MODULES)
+        )
+    )
+    assert not extra, f"unexpected modules under repositories/: {extra}"
+
+
+@pytest.mark.parametrize(
+    "path", sorted(SERVICES.rglob("*.py")), ids=lambda p: str(p.relative_to(SRC))
+)
+def test_service_uses_its_own_section(path: Path) -> None:
+    """A service reaches another domain through that domain's service, never its repositories."""
+    key = _service_key(path)
+    allowed = {"traust_core.v1.repositories.objects", "traust_core.v1.repositories.unit_of_work"}
+    if key in SERVICE_SECTION:
+        allowed.add(f"traust_core.v1.repositories.{SERVICE_SECTION[key]}")
+    if key in SPANS_SECTIONS:
+        allowed.add("traust_core.v1.repositories.contract.TraustUnitOfWork")
+    bad = sorted(
+        name
+        for name in _imported_modules(path)
+        if name.startswith("traust_core.v1.repositories")
+        and name != "traust_core.v1.repositories.contract"
+        and not any(name == a or name.startswith(a + ".") for a in allowed)
+    )
+    assert not bad, f"{key} may not import {bad}"
+
+
+@pytest.mark.parametrize(
+    "path",
+    sorted(
+        [
+            *SERVICES.rglob("*.py"),
+            *(REPOSITORIES / "traust").rglob("*.py"),
+            *(REPOSITORIES / "traust_ops").rglob("*.py"),
+            *(REPOSITORIES / "traust_ledger").rglob("*.py"),
+        ]
+    ),
+    ids=lambda p: str(p.relative_to(SRC)),
+)
+def test_no_file_or_sqlite_access_outside_object_storage(path: Path) -> None:
+    """Services and table repositories read contract tables only; bytes go via ObjectStore."""
+    hits = []
+    for node in ast.walk(_tree(path)):
+        if isinstance(node, ast.Import | ast.ImportFrom):
+            names = {a.name for a in node.names} | {getattr(node, "module", None) or ""}
+            if "sqlite3" in names:
+                hits.append("import sqlite3")
+        elif isinstance(node, ast.Call):
+            fn = node.func
+            name = fn.attr if isinstance(fn, ast.Attribute) else getattr(fn, "id", "")
+            if name in FILE_ACCESS and not (isinstance(fn, ast.Attribute) and name == "open"):
+                hits.append(f"{name}()")
+            if isinstance(fn, ast.Name) and fn.id == "open":
+                hits.append("open()")
+    assert not hits, f"{path.relative_to(SRC)}: {sorted(set(hits))}"
+
+
+def test_in_memory_doubles_live_in_tests() -> None:
+    """In-memory stand-ins hide missing persistence; only the shared doubles stay in src."""
+    found = sorted(
+        f"{p.relative_to(SRC)}:{node.name}"
+        for p in SRC.rglob("*.py")
+        for node in ast.walk(_tree(p))
+        if isinstance(node, ast.ClassDef)
+        and node.name.startswith("InMemory")
+        and node.name not in IN_MEMORY_ALLOWED
+    )
+    assert not found, f"move in-memory doubles to tests/: {found}"
+
+
+def test_context_opens_only_the_configured_database() -> None:
+    """Context builds one engine from config; it never opens another database."""
+    engines, connects = [], []
+    for path in (SRC / "context").rglob("*.py"):
+        for node in ast.walk(_tree(path)):
+            if isinstance(node, ast.Call):
+                fn = node.func
+                name = fn.attr if isinstance(fn, ast.Attribute) else getattr(fn, "id", "")
+                where = f"{path.relative_to(SRC)}:{node.lineno}"
+                if name in {"create_database_engine", "create_engine"}:
+                    engines.append(where)
+                if name == "connect":
+                    connects.append(where)
+    assert len(engines) <= 1, f"Context creates more than one engine: {engines}"
+    assert not connects, f"Context opens connections itself (use a unit of work): {connects}"