feat(trail): Postgres-backed snapshot store and source for trail

.gitignore

@@ -216,4 +216,5 @@ __marimo__/
 
 # Trail MCP deployment artifacts (k3s manifests + trail Dockerfile) belong with the
 # trail project, not this repo. Kept on disk locally for the live deployment, ignored here.
-/deploy/
+/deploy/trail-snapshot-store/.venv/
+trail-snapshot-source/.venv/

deploy/Dockerfile.trail-mcp

@@ -0,0 +1,67 @@
+# trail-mcp: the Trail MCP server WITH live data-source adapters (FMP / GMD / scores / snapshot).
+#
+# Build from the aiofmp workspace root (the COPYs below are relative to it):
+#   docker build -f deploy/Dockerfile.trail-mcp -t <reg>/trail-mcp-full:<tag> .
+#
+# RECONSTRUCTED 2026-08-12, and committed this time. The working version of this file previously
+# existed only as an uncommitted working-tree edit, and was lost when local main was reset to
+# origin/main (upstream untracked deploy/ in 1eb3b06). Reconstruction verified against the running
+# cache-fix-1 image: trail-lang 0.18.0, trail-fmp 0.3.0, trail-gmd 0.3.0, trail-scores 0.1.0,
+# numpy 2.x, and trail-edgar ABSENT.
+#
+# The base is the upstream release image — language + MCP server, published by trail-lang's own
+# release workflow. This file used to vendor trail-py and reinstall the language on every adapter
+# change; pinning a RELEASED version instead is deliberate. An unpinned working tree is what
+# silently moved the language from 0.15 to 0.17 underneath a tracked view and invalidated a
+# 3.5-hour build. The consequence to remember: local trail-py edits no longer reach the cluster
+# until a version is tagged and this pin is bumped.
+FROM ghcr.io/trail-language/trail-lang:0.18.0
+
+# The base finishes as the unprivileged `trail` user; installs need root, and we drop back after.
+USER root
+WORKDIR /app
+
+# aiofmp from the LOCAL tree, not PyPI. It carries fixes ahead of the published version and pip
+# would consider a same-versioned wheel sufficient — silently installing a client whose bool query
+# params never reach the API, and whose period-based cache never actually avoids a request.
+COPY pyproject.toml README.md aiofmp-src/
+COPY aiofmp aiofmp-src/aiofmp
+RUN pip install --no-cache-dir ./aiofmp-src
+
+# Provider adapters. trail-edgar is deliberately ABSENT: EDGAR was removed from trail.yaml after it
+# aborted a full 5,368-name build by raising CompanyNotFoundError on a ticker it did not know, while
+# only ever covering 39 mega-caps that fmp already served. Two days of runs wrote nothing.
+COPY trail-fmp/pyproject.toml trail-fmp/README.md trail-fmp/
+COPY trail-fmp/trail_fmp trail-fmp/trail_fmp
+COPY trail-gmd/pyproject.toml trail-gmd/README.md trail-gmd/
+COPY trail-gmd/trail_gmd trail-gmd/trail_gmd
+COPY trail-scores/pyproject.toml trail-scores/README.md trail-scores/
+COPY trail-scores/trail_scores trail-scores/trail_scores
+RUN pip install --no-cache-dir ./trail-fmp ./trail-gmd ./trail-scores
+
+# Snapshot integration. These register under DIFFERENT entry-point groups and are not
+# interchangeable: the store under `trail.providers` (writeable view stores), the source under
+# `trail.sources` (read side). A store declared in the wrong group resolves to nothing and trail
+# silently falls back to a local-disk store — a deployment that looks healthy while writing views
+# somewhere else entirely.
+COPY trail-snapshot-store/pyproject.toml trail-snapshot-store/
+COPY trail-snapshot-store/trail_snapshot_store trail-snapshot-store/trail_snapshot_store
+COPY trail-snapshot-source/pyproject.toml trail-snapshot-source/
+COPY trail-snapshot-source/trail_snapshot_source trail-snapshot-source/trail_snapshot_source
+RUN pip install --no-cache-dir ./trail-snapshot-store ./trail-snapshot-source
+
+# numpy is NOT optional despite nothing declaring it. Polars' Rust extension reaches for the NumPy
+# C-API capsule on to_numpy paths and, when numpy is missing, raises a PyO3 PanicException that
+# ABORTS the process — the server dies with SIGSEGV seconds into a fetch, not with an ImportError.
+# It used to arrive transitively (edgartools -> pandas -> numpy); removing EDGAR removed it, and
+# nothing failed until runtime. Pinned here explicitly so the coupling is visible.
+RUN pip install --no-cache-dir "numpy>=2,<3"
+
+# Mount points the ConfigMap/PVCs land on. /scores and /views are written at runtime by the server
+# (to_file and the view store), so they must be writable by the runtime user.
+RUN mkdir -p /cache /gmd-cache /config /scores /views \
+    && chown -R trail:trail /cache /gmd-cache /scores /views
+
+USER trail
+
+# EXPOSE 3000, ENTRYPOINT ["trail","mcp"] and the streamable-http CMD are inherited from the base.

trail-snapshot-source/pyproject.toml

@@ -0,0 +1,21 @@
+[project]
+name = "trail-snapshot-source"
+version = "0.1.0"
+description = "Expose snapshot facts to trail as panel columns, with a published_at changefeed"
+requires-python = ">=3.11"
+dependencies = ["trail-lang>=0.18.0", "polars>=1.0", "psycopg[binary]>=3.1"]
+
+# trail.sources is the READ-side group - the one trail-fmp uses. View stores go
+# under trail.providers instead; the two are not interchangeable.
+[project.entry-points."trail.sources"]
+snapshot = "trail_snapshot_source:SnapshotSource"
+
+[dependency-groups]
+dev = ["pytest>=8.0"]
+
+[build-system]
+requires = ["hatchling"]
+build-backend = "hatchling.build"
+
+[tool.hatch.build.targets.wheel]
+packages = ["trail_snapshot_source"]

trail-snapshot-source/tests/conftest.py

@@ -0,0 +1,33 @@
+import os
+import pathlib
+
+import psycopg
+import pytest
+from psycopg.rows import dict_row
+
+# Reuse the harness migrations: this source reads the same schema.
+MIGRATIONS = pathlib.Path(
+    "/home/usmanshahid/Documents/claude/workspaces/harness/snapshot/migrations"
+)
+
+
+@pytest.fixture
+def conn():
+    c = psycopg.connect(
+        os.environ["SNAPSHOT_TEST_DSN"], autocommit=True, row_factory=dict_row
+    )
+    c.execute("DROP SCHEMA IF EXISTS snapshot CASCADE")
+    c.execute("LOAD 'age'")
+    c.execute('SET search_path = ag_catalog, "$user", public')
+    if c.execute(
+        "SELECT 1 FROM ag_catalog.ag_graph WHERE name='snapshot_graph'"
+    ).fetchone():
+        c.execute("SELECT ag_catalog.drop_graph('snapshot_graph', true)")
+    for f in sorted(MIGRATIONS.glob("*.sql")):
+        c.execute(f.read_text())
+    c.execute(
+        "INSERT INTO snapshot.sources (source_id, name) VALUES (1,'test') "
+        "ON CONFLICT DO NOTHING"
+    )
+    yield c
+    c.close()

trail-snapshot-source/tests/test_source.py

@@ -0,0 +1,66 @@
+"""Facts as trail panel columns, plus the changefeed that drives incremental recompute."""
+
+import os
+
+import pytest
+from trail_snapshot_source import SnapshotSource
+
+
+@pytest.fixture
+def src(conn):
+    return SnapshotSource({"dsn": os.environ["SNAPSHOT_TEST_DSN"]})
+
+
+def _fact(conn, subj, pred, val, valid_from, published_at):
+    conn.execute(
+        """INSERT INTO snapshot.facts
+             (subject, predicate, value, kind, valid_from, published_at, fidelity, source_id)
+           VALUES (%s, %s, %s, 'observation', %s, %s, 'vintage', 1)""",
+        (subj, pred, val, valid_from, published_at),
+    )
+
+
+def test_freshness_token_is_none_when_empty(conn, src):
+    assert src.freshness_token() is None
+
+
+def test_freshness_token_moves_when_a_fact_arrives(conn, src):
+    before = src.freshness_token()
+    _fact(conn, "AAPL", "revenue", 1.0, "2025-12-31", "2026-02-01Z")
+    assert src.freshness_token() != before
+
+
+def test_changed_since_returns_only_newer_cells(conn, src):
+    _fact(conn, "AAPL", "revenue", 1.0, "2024-12-31", "2025-02-01Z")
+    _fact(conn, "MSFT", "revenue", 2.0, "2025-12-31", "2026-02-01Z")
+    changed = src.changed_since("2025-06-01T00:00:00+00:00")
+    assert {e for e, _ in changed} == {"MSFT"}
+
+
+def test_changed_since_none_cursor_returns_everything(conn, src):
+    _fact(conn, "AAPL", "revenue", 1.0, "2024-12-31", "2025-02-01Z")
+    assert len(src.changed_since(None)) == 1
+
+
+def test_restatement_dirties_the_cell_again(conn, src):
+    """A restatement arrives as a NEW row with a later published_at, so the same
+    (entity, period) cell must reappear in the changefeed - otherwise the view
+    would never pick the correction up."""
+    _fact(conn, "AAPL", "revenue", 1.0, "2024-12-31", "2025-02-01Z")
+    _fact(conn, "AAPL", "revenue", 0.9, "2024-12-31", "2026-03-01Z")
+    changed = src.changed_since("2025-06-01T00:00:00+00:00")
+    assert ("AAPL", __import__("datetime").date(2024, 12, 31)) in changed
+
+
+def test_load_pivots_predicates_into_columns(conn, src):
+    _fact(conn, "AAPL", "revenue", 100.0, "2025-12-31", "2026-02-01Z")
+    _fact(conn, "AAPL", "net_income", 20.0, "2025-12-31", "2026-02-01Z")
+    df = src.load()
+    assert set(df.columns) >= {"entity", "time", "revenue", "net_income"}
+    assert df.height == 1
+
+
+def test_load_takes_the_latest_published_value(conn, src):
+    _fact(conn, "AAPL", "revenue", 100.0, "2025-12-31", "2026-02-01Z")
+    _fact(conn, "AAPL", "revenue", 90.0, "2025-12-31", "2026-05-01Z")
+    assert src.load()["revenue"].to_list() == [90.0]

trail-snapshot-source/trail_snapshot_source/__init__.py

@@ -0,0 +1,3 @@
+from .source import SnapshotSource
+
+__all__ = ["SnapshotSource"]

trail-snapshot-source/trail_snapshot_source/source.py

@@ -0,0 +1,98 @@
+"""Expose snapshot facts to trail as panel columns, with a changefeed.
+
+This is the load-bearing half of the trail integration. trail already has the
+incremental update engine the snapshot design needs — per ``trail/views.py``,
+staleness is a declaration hash plus a panel-config fingerprint plus per-source
+freshness, and a source exposing ``changed_since`` gets footprint-scoped
+recompute instead of a whole-view rebuild. We are not building an update engine;
+we are feeding one that works.
+
+``changed_since`` keys on ``published_at`` — the point-in-time clock — so a
+restatement, which arrives as a NEW row with a later ``published_at``, dirties
+exactly the cells it should. ``trail-fmp`` does the same thing keyed on filing
+date (``trail_fmp/source.py:172``), noting that an amendment carries a newer
+acceptedDate; this is the same idea for a different backend.
+"""
+
+from __future__ import annotations
+
+import datetime as dt
+import os
+
+import polars as pl
+import psycopg
+from psycopg.rows import dict_row
+
+# One row per (entity, period), latest publication winning. The DISTINCT ON
+# ordering mirrors snapshot.facts.as_of: published_at DESC picks the newest
+# vintage, and source_id breaks ties deterministically so two sources publishing
+# at the same instant cannot make the panel non-reproducible.
+_LOAD = """
+SELECT DISTINCT ON (subject, predicate, valid_from)
+       subject AS entity, valid_from AS time, predicate, value
+  FROM {schema}.facts
+ WHERE value IS NOT NULL
+ ORDER BY subject, predicate, valid_from, published_at DESC, source_id
+"""
+
+
+class SnapshotSource:
+    """A trail source over ``snapshot.facts``."""
+
+    name = "snapshot"
+
+    def __init__(self, options: dict | None = None) -> None:
+        options = options or {}
+        # Option first, environment second - the precedence trail-fmp uses for
+        # its API key. trail.yaml is a ConfigMap, so a password-bearing DSN has
+        # to arrive through the environment from a secret.
+        self._dsn = options.get("dsn") or os.environ.get("SNAPSHOT_DSN")
+        if not self._dsn:
+            raise ValueError(
+                "E-SNAPSHOT-DSN snapshot source requires a connection string; "
+                "set options.dsn or the SNAPSHOT_DSN environment variable"
+            )
+        self._schema = options.get("schema", "snapshot")
+
+    def _connect(self):
+        return psycopg.connect(self._dsn, autocommit=True, row_factory=dict_row)
+
+    def load(self, *args, **kwargs) -> pl.DataFrame:
+        """The panel: entity x time, one column per predicate."""
+        with self._connect() as c:
+            rows = c.execute(_LOAD.format(schema=self._schema)).fetchall()
+        if not rows:
+            return pl.DataFrame({"entity": [], "time": []})
+        return (
+            pl.DataFrame(rows)
+            .pivot(values="value", index=["entity", "time"], on="predicate")
+            .sort(["entity", "time"])
+        )
+
+    def freshness_token(self) -> str | None:
+        """Max published_at — cheap, and moves whenever anything new lands.
+
+        None means "no freshness signal", which tells trail to serve the stored
+        view until the program changes. That is the correct answer for an empty
+        store: there is nothing to be stale about yet.
+        """
+        with self._connect() as c:
+            row = c.execute(
+                f"SELECT max(published_at) AS m FROM {self._schema}.facts"
+            ).fetchone()
+        return row["m"].isoformat() if row and row["m"] else None
+
+    def changed_since(self, cursor: str | None):
+        """The ``(entity, period-end)`` cells published after ``cursor``.
+
+        ``cursor`` is a token previously returned by :meth:`freshness_token`.
+        None means "everything", which is what a first build wants.
+        """
+        sql = f"SELECT DISTINCT subject, valid_from FROM {self._schema}.facts"
+        params: tuple = ()
+        if cursor is not None:
+            sql += " WHERE published_at > %s"
+            params = (dt.datetime.fromisoformat(cursor),)
+        with self._connect() as c:
+            rows = c.execute(sql, params).fetchall()
+        return {(r["subject"], r["valid_from"]) for r in rows}