feat(portfolio): move repository observation and state assembly into Core

.agents/skills/commit-prep/SKILL.md

@@ -0,0 +1,58 @@
+---
+name: commit-prep
+description: Take traust-core changes from working tree to merged PR in one run — gates, conventional signed commit, branch, PR body, auto-merge. Use when asked to prep a commit, open a PR, or land changes in traust-core.
+---
+
+# Commit Prep (traust-core)
+
+Run the changes through this repo's gates, then land them as a PR. The org ruleset
+requires a PR for main but **zero approvals** — self-merge is legal — and commits
+must be signed (`commit.gpgsign` is on in this repo, so signing is automatic).
+
+## Workflow
+
+1. **Scope the change.** `git status`, `git diff --stat`, `git log --oneline -5` for
+   style. If the tree mixes unrelated changes, split commits by concern before
+   proceeding — one PR carries one story.
+2. **Run the gates, in order, all must pass:**
+   ```bash
+   make lint-fix && make test && make coverage && make db-up && make test-integration
+   ```
+   - `make coverage` fails below 80% — new code needs tests; stub bodies are excluded.
+   - **Known flake:** integration fixtures failing with `UndefinedColumn` (e.g.
+     `column "severity" does not exist`) means the shared `traust_test` container
+     holds the other DDL generation (contracts branch vs the pinned dep both
+     recreate it). Clear and re-run:
+     ```bash
+     podman exec traust-postgres psql -U traust -d traust_test -c \
+       'DROP SCHEMA IF EXISTS traust_storage CASCADE; DROP SCHEMA IF EXISTS traust_ledger CASCADE;'
+     ```
+     If it persists after clearing, it is a real schema mismatch — stop and report.
+3. **Commit.** Conventional subject, enforced by the commit-msg hook:
+   `type(scope): subject` — types feat/fix/refactor/docs/test/chore/ci/perf, scope is
+   the narrowest affected area, imperative lowercase subject. Hooks run ruff on
+   staged files; do not bypass with `--no-verify`.
+4. **Branch and push.** Kebab-case `type/short-description`, off current main:
+   `git push -u origin <branch>`. Never force-push `main` (org ruleset blocks it).
+5. **Open the PR** with `gh pr create --base main`, body shaped like:
+   - **What** — the change in 1–3 bullets, tied to the model/slice it serves
+   - **The decisions** — any rule that governed it (one owner, DB-owned identity, pattern set)
+   - **Tests** — which gates ran, any known environmental caveat
+   - **Tracker** — the XWING story key
+6. **Merge.** `gh pr merge <n> --merge --delete-branch --auto` — merges when the five
+   CI checks pass (lint, test 3.11/3.12, api-check, postgres). Post-merge: pull main
+   locally, log the milestone to the story.
+
+## Rules
+
+- Gates first: no PR opens with a red suite. If a gate cannot pass, report why —
+  do not weaken the gate.
+- One PR, one story. Split mixed trees into separate commits/PRs.
+- Never merge another author's PR or one with a human's open review request.
+- Never push directly to `main` — the ruleset rejects it anyway.
+- If contracts or DDL shapes are involved, say so in the PR body and flag the pin.
+
+## Done when
+
+The PR is merged (or open with green CI and auto-merge armed), local main is synced,
+and the milestone is logged to the story.

README.md

@@ -13,6 +13,7 @@ All modules live under `traust_core.v1` (see [Versioning](#versioning)).
 | **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) → an `artifact` row (when `database.url` is set) | writing `analysis-results` files by hand |
 | **Analysis results** | `repositories.objects.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 |
 | **Table repositories** | `repositories.traust.*`, `repositories.traust_ops`, `repositories.traust_ledger`, `repositories.contract` | one repository per contract table (`uow.findings.find_by_fingerprint(...)`, `uow.run_orders.claim(...)`); `SqlTraustUnitOfWork(engine)` opens all 31 on one connection — one transaction across sections | `sqlite3.connect`, inline SQL, per-script tables |
+| **Repository observation** | `providers.forge`, `services.observation`, `services.portfolio`, `repositories.observation` | gh/glab measurements through `ProcessRunner` (`CommandSourceProvider`, strict reply validation); targets and open counts read the contract tables (`repository`, `assessment`, the `finding_status` view); `RepoStateQuery.observed` derives exposure / risk tier / freshness. Observations persist in memory until contracts lands `repository_observation` | per-script forge scraping, findings.db direct reads |
 | **Records (pattern)** | `repositories.sql`, `repositories.tables` | `TableSpec` (columns once) + `TableFactory` (Postgres schemas or SQLite `ops_`/`ledger_` prefixes, by engine) + `SqlTableRepository` (`add`/`get`/`update`) 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.objects.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` |

src/traust_core/v1/context/config.py

@@ -8,6 +8,7 @@
 import yaml
 
 from traust_core.v1.errors import ConfigError
+from traust_core.v1.models.inventory import DEFAULT_FORGE_HOSTS
 
 
 class _Model(pydantic.BaseModel):
@@ -60,6 +61,7 @@ class Config(_Model):
     artifacts: ArtifactsConfig = ArtifactsConfig()
     process: ProcessConfig = ProcessConfig()
     allowed_repo_hosts: tuple[str, ...] = ()
+    forge_hosts: dict[str, str] = pydantic.Field(default_factory=lambda: dict(DEFAULT_FORGE_HOSTS))
 
     @classmethod
     def from_file(cls, path: Path) -> Config:

src/traust_core/v1/context/context.py

@@ -1,13 +1,16 @@
 from __future__ import annotations
 
-from collections.abc import Callable
+from collections.abc import Callable, Iterator
+from contextlib import contextmanager
+from dataclasses import dataclass
 from functools import cached_property
 from pathlib import Path
 from typing import Any, TypeVar
 
 from sqlalchemy.engine import Engine
 
 from traust_core.v1.context.config import Config
+from traust_core.v1.context.observation import command_source
 from traust_core.v1.models.artifacts import ARTIFACT_FOR_KIND
 from traust_core.v1.providers.base import ProviderRegistry
 from traust_core.v1.providers.clock import Clock, SystemClock
@@ -22,15 +25,23 @@
     create_database_engine,
 )
 from traust_core.v1.repositories.contract import SqlTraustUnitOfWork
+from traust_core.v1.repositories.designations import FileDesignationRepository
 from traust_core.v1.repositories.objects.analysis_results import AnalysisResultsRepository
+from traust_core.v1.repositories.observation import InMemoryObservationStore
 from traust_core.v1.repositories.sql import Dialect
 from traust_core.v1.repositories.tables import TableFactory
 from traust_core.v1.services.artifact_publishing import (
     ArtifactPublisher,
     PublishSpecs,
     SqlArtifactIndex,
 )
+from traust_core.v1.services.assessment import AssessmentService
 from traust_core.v1.services.operations import Materializer
+from traust_core.v1.services.portfolio import (
+    PortfolioPolicy,
+    PortfolioService,
+    RepoStateQuery,
+)
 
 U = TypeVar("U", bound=UnitOfWork)
 
@@ -40,6 +51,15 @@
 class Services: ...
 
 
+@dataclass(frozen=True)
+class _Observation:
+    """What the observation door yields: the write path, the composite read, the store."""
+
+    service: PortfolioService
+    query: RepoStateQuery
+    store: InMemoryObservationStore
+
+
 class Context:
     def __init__(
         self,
@@ -106,3 +126,43 @@ def services(self) -> Services:
 
     def analysis_results(self) -> AnalysisResultsRepository:
         return AnalysisResultsRepository(self.object_store, ARTIFACT_FOR_KIND)
+
+    @contextmanager
+    def repository_observation(
+        self,
+        *,
+        designations: Path | None = None,
+    ) -> Iterator[_Observation]:
+        """Observe repositories: forge facts in, derived routing state out.
+
+        Reads inventory, assessments and open counts from the contract tables
+        over one session unit of work; the observation store is in memory until
+        contracts lands repository_observation (NEED marker in
+        repositories/observation.py).
+        """
+        store = InMemoryObservationStore()
+        policy = PortfolioPolicy()
+        with SqlTraustUnitOfWork(self.database) as uow:
+            assessment = AssessmentService(uow, clock=self.clock, hosts=self.config.forge_hosts)
+            service = PortfolioService(
+                uow,
+                policy,
+                store=store,
+                assessment=assessment,
+                sources=command_source(self.config.forge_hosts, clock=self.clock),
+                clock=self.clock,
+                hosts=self.config.forge_hosts,
+            )
+            query = RepoStateQuery(
+                service,
+                assessment,
+                None,  # findings service: open counts come from the finding_status view
+                uow.open_counts,
+                policy,
+                designations=(
+                    FileDesignationRepository(designations) if designations is not None else None
+                ),
+                clock=self.clock,
+                hosts=self.config.forge_hosts,
+            )
+            yield _Observation(service=service, query=query, store=store)

src/traust_core/v1/context/observation.py

@@ -0,0 +1,34 @@
+from collections.abc import Mapping
+
+from traust_core.v1.providers.clock import Clock, SystemClock
+from traust_core.v1.providers.forge import CommandSourceProvider
+from traust_core.v1.providers.process import ProcessRunner
+
+# Preserve forge CLI credentials during the existing batch command's migration.
+_FORGE_ENV = (
+    "PATH",
+    "HOME",
+    "LANG",
+    "LC_ALL",
+    "TMPDIR",
+    "GH_TOKEN",
+    "GITHUB_TOKEN",
+    "GH_ENTERPRISE_TOKEN",
+    "GITHUB_ENTERPRISE_TOKEN",
+    "GH_HOST",
+    "GH_CONFIG_DIR",
+    "GITLAB_TOKEN",
+    "GITLAB_HOST",
+    "GLAB_CONFIG_DIR",
+    "XDG_CONFIG_HOME",
+)
+
+
+def command_source(
+    hosts: Mapping[str, str], *, timeout_seconds: float = 120, clock: Clock | None = None
+) -> CommandSourceProvider:
+    return CommandSourceProvider(
+        ProcessRunner(("gh", "glab"), keep_env=_FORGE_ENV, timeout_seconds=timeout_seconds),
+        clock or SystemClock(),
+        hosts,
+    )

src/traust_core/v1/models/assessment.py

@@ -74,6 +74,10 @@ def ratio(self) -> float:
         raise NotImplementedError("stub: replaces build_coverage_gap_rollup.build_rollup")
 
 
+# NEED(contracts): the assessment-status enum, once it lands, owns these values.
+ASSESSMENT_COMPLETED = "completed"
+
+
 class Assessment(AppendOnly):
     """3a · what was examined: a ref (and image), by an assessor, with what outcome."""
 
@@ -129,12 +133,14 @@ class IsolationResult(Model):
 
 
 class Freshness(Model):
-    repository_ref_id: UUID
+    repository_ref_id: UUID | None = None
     last_assessed_at: datetime | None = None
     churn_ratio: float | None = None
 
     def age_days(self, now: datetime) -> int | None:
-        raise NotImplementedError("stub: replaces build_rescan_worklist audit_age_days")
+        if self.last_assessed_at is None:
+            return None
+        return (now.date() - self.last_assessed_at.date()).days
 
 
 class Population(Model):

src/traust_core/v1/models/inventory.py

@@ -78,6 +78,9 @@ class RiskTier(StrEnum):
     P3 = "P3"
 
 
+DEFAULT_FORGE_HOSTS: dict[str, str] = {"github.com": "github", "gitlab.com": "gitlab"}
+
+
 class ForgeHost(StrEnum):
     GITHUB = "github"
     GITLAB = "gitlab"

src/traust_core/v1/models/observation.py

@@ -0,0 +1,118 @@
+"""Repository observation: forge measurements and the state derived from them.
+
+Keyed by repository_id, matching the inventory tables.
+"""
+
+from __future__ import annotations
+
+from enum import StrEnum
+from uuid import UUID
+
+from pydantic import AwareDatetime, field_serializer
+
+from traust_core.v1.models.base import Model
+from traust_core.v1.models.inventory import ExposureClass, RiskTier
+from traust_core.v1.models.values import HttpsRepoUrl
+
+
+class Designation(StrEnum):
+    """Adopter-configured exposure hint; customer-reach derives from component -> product."""
+
+    EXTERNAL = "external"
+    INTERNAL_TOOLING = "internal-tooling"
+
+
+class ComparisonStatus(StrEnum):
+    OK = "ok"
+    QUOTA_DEFERRED = "quota-deferred"
+    NO_PINNED_SHA = "no-pinned-sha"
+    UNSUPPORTED_HOST = "unsupported-host"
+    NO_REPO_URL = "no-repo-url"
+    NO_CREDENTIALS = "no-credentials"
+    UNREACHABLE = "unreachable"
+    ERROR = "error"
+    BAN_SUSPECTED = "ban-suspected"
+    NETWORK_SKIPPED = "network-skipped"
+    NEVER_AUDITED = "never-audited"
+
+
+class ObservationTarget(Model):
+    """What the observation measures: a repository, its latest audit, its pinned commit."""
+
+    repository_id: UUID
+    repo_url: str | None
+    has_audit: bool = True
+    audit_date: str | None = None
+    pinned_sha: str | None = None
+    # NEED(artifacts): from the assessment's artifact document once that lane lands.
+    lines_reviewed: int | None = None
+    designation: Designation | None = None
+    repository_ref_id: UUID | None = None
+
+
+class ForgeFacts(Model):
+    repository_id: UUID
+    repo: HttpsRepoUrl | None
+    repo_url: str | None
+    audited_sha: str | None
+    audit_date: str | None
+    has_audit: bool
+    status: ComparisonStatus
+    observed_at: AwareDatetime
+    pushed_at: str | None = None
+    archived: bool | None = None
+    default_branch: str | None = None
+    visibility: str | None = None
+    fork: bool | None = None
+    parent: str | None = None
+    ahead_by: int | None = None
+    changed_lines: int | None = None
+    sensitive_changed: bool | None = None
+    sensitive_lines: int | None = None
+    deps_only: bool | None = None
+    compare_truncated: bool | None = None
+    error: str | None = None
+
+    @field_serializer("repo", when_used="json")
+    def serialize_repo(self, repo: HttpsRepoUrl | None) -> str | None:
+        return str(repo) if repo else None
+
+
+class ChangeMetrics(Model):
+    changed_lines: int
+    sensitive_changed: bool
+    sensitive_lines: int
+    deps_only: bool
+    first_party_files: int
+    iac_files: int
+    threat_model_files: int
+
+
+class DesignationRecord(Model):
+    match: str
+    designation: Designation
+
+
+class ObservedRepoState(Model):
+    """The routing state derived from one repository's latest observation."""
+
+    repository_id: UUID
+    repo: HttpsRepoUrl | None
+    status: ComparisonStatus
+    risk_tier: RiskTier
+    exposure: ExposureClass
+    audit_age_days: int | None
+    changed_lines: int | None
+    churn_ratio: float | None
+    sensitive_changed: bool | None
+    sensitive_lines: int | None
+    deps_only: bool | None
+    ahead_by: int | None
+    compare_truncated: bool | None
+    push_changed: bool | None
+    observed_at: AwareDatetime
+    error: str | None = None
+
+    @field_serializer("repo", when_used="json")
+    def serialize_repo(self, repo: HttpsRepoUrl | None) -> str | None:
+        return str(repo) if repo else None