Preserve temporal review history

Author: Coding-Dev-Tools

AGENTS.md

@@ -20,7 +20,7 @@ most common mistake here.
 | Status | Primary scoped, bi-temporal, interface-driven implementation. | Compatibility/reference implementation with flat namespaces. |
 | Model | Scoped + bi-temporal + typed; interface-driven. | Single flat `namespace` string per memory. |
 | Code | `engraphis/core/`, `engraphis/backends/`, `eval/`, `tests/`, `scripts/migrate_to_v2.py` | `engraphis/app.py`, `config.py`, `models.py`, `routes/`, `stores/`, `engines/`, `llm/`, `static/` |
-| Data | new v2 schema (`SCHEMA_VERSION = 5`) | `engraphis_v1.db` |
+| Data | new v2 schema (`SCHEMA_VERSION = 6`) | `engraphis_v1.db` |
 | Entry | `MemoryEngine.create()` → `core/engine.py` | `python -m scripts.start_server` → FastAPI on :8700 |
 
 **Rule:** build new capability on **v2** (`core/` + `backends/`) behind the interfaces.
@@ -177,7 +177,7 @@ These are pure, unit-tested functions — change them only with a corresponding
 
 ---
 
-## 5. Data model cheat-sheet (`core/interfaces.py`, `core/schema.py` — `SCHEMA_VERSION = 5`)
+## 5. Data model cheat-sheet (`core/interfaces.py`, `core/schema.py` — `SCHEMA_VERSION = 6`)
 
 - **Scope hierarchy:** `workspace → repo → session → memory`. Scopes: `session|repo|workspace|user`.
 - **Bi-temporal validity on every record:** world-time `valid_from/valid_to` +

engraphis/core/engine.py

@@ -458,7 +458,7 @@ def _resolve_and_store(self, content: str, *, text: str, vec: np.ndarray,
                 text, vec, workspace_id=workspace_id, repo_id=repo_id,
                 session_id=session_id, scope=scope, mtype=mtype,
                 candidate_k=candidate_k, subject_key=subject_key,
-                claim_kind=claim_kind,
+                claim_kind=claim_kind, valid_at=valid_from,
             )
         if (decision is not None
                 and decision.op == ResolutionOp.INVALIDATE
@@ -743,20 +743,37 @@ def _evolve(self, new_id: str, neighbors: list, *, exclude: Optional[set] = None
     def _resolve_against_neighbors(self, text: str, vec: np.ndarray, *, workspace_id: str,
                                    repo_id: Optional[str], session_id: Optional[str],
                                    scope: Scope, mtype: MemoryType, candidate_k: int,
-                                   subject_key: str = "", claim_kind: str = ""):
+                                   subject_key: str = "", claim_kind: str = "",
+                                   valid_at: Optional[float] = None):
         """Fetch same-scope neighbors via the vector index and run the deterministic
         resolver (``core.resolve``). Returns ``(decision, neighbors)`` so the caller can
         also evolve the neighborhood. Never raises — a broken/missing index degrades to
         "no neighbors found" (ADD), not a write failure."""
         flt = SearchFilter(
             workspace_id=workspace_id, repo_id=repo_id,
             session_id=session_id if scope == Scope.SESSION else None,
-            scopes=[scope], mtypes=[mtype],
+            scopes=[scope], mtypes=[mtype], valid_at=valid_at,
         )
         try:
             hits = self.index.search(vec, candidate_k, filter=flt)
         except Exception:
             return None, []
+        if not hits and valid_at is not None:
+            # A candidate may be backdated before an already-recorded claim. That claim
+            # is intentionally outside the candidate's valid-time view, but it still
+            # has to be found so the caller can reject an impossible supersession rather
+            # than silently creating overlapping history. This fallback is only a guard
+            # for an otherwise-empty temporal neighborhood; normal scheduled resolution
+            # remains anchored at the candidate's validity time above.
+            current_filter = SearchFilter(
+                workspace_id=workspace_id, repo_id=repo_id,
+                session_id=session_id if scope == Scope.SESSION else None,
+                scopes=[scope], mtypes=[mtype],
+            )
+            try:
+                hits = self.index.search(vec, candidate_k, filter=current_filter)
+            except Exception:
+                pass
         now = now_ts()
         neighbors = []
         for nid, sim in hits:
@@ -2030,7 +2047,7 @@ def export_code_graph(self, *, repo_id: str,
         analysis.pop("_node_community", None)
         # Fetch one sentinel row beyond the payload cap so truncation stays observable
         # without materializing every indexed file in a large repository.
-        files = self.store.list_code_files(repo_id, limit=limit + 1)
+        files = self.store.list_code_files(repo_id, flt=flt, limit=limit + 1)
         truncated_files = len(files) > limit
         files = files[:limit]
         nodes = self.store.list_symbols(repo_id, limit=limit, flt=flt)

engraphis/core/schema.py

@@ -8,7 +8,7 @@
 """
 from __future__ import annotations
 
-SCHEMA_VERSION = 5
+SCHEMA_VERSION = 6
 
 SCHEMA_SQL = """
 CREATE TABLE IF NOT EXISTS schema_migrations (
@@ -361,6 +361,31 @@
 );
 CREATE INDEX IF NOT EXISTS idx_code_files_lang ON code_files(repo_id, lang);
 
+-- ``code_files`` is the current indexing manifest. Historical code exports use this
+-- append-only companion so a deleted or replaced file remains visible at the correct
+-- world/system-time anchors alongside its retired symbols and code edges.
+CREATE TABLE IF NOT EXISTS code_file_history (
+    version                INTEGER PRIMARY KEY,
+    repo_id                TEXT NOT NULL,
+    file                   TEXT NOT NULL,
+    lang                   TEXT,
+    content_hash           TEXT NOT NULL,
+    size_bytes             INTEGER DEFAULT 0,
+    mtime_ns               INTEGER DEFAULT 0,
+    backend                TEXT DEFAULT '',
+    indexed_at             REAL,
+    valid_from             REAL,
+    valid_to               REAL,
+    valid_to_recorded_at   REAL,
+    ingested_at            REAL,
+    expired_at             REAL
+);
+CREATE INDEX IF NOT EXISTS idx_code_file_history_temporal
+    ON code_file_history(repo_id, file, valid_to, expired_at);
+CREATE UNIQUE INDEX IF NOT EXISTS idx_code_file_history_live
+    ON code_file_history(repo_id, file)
+    WHERE valid_to IS NULL AND expired_at IS NULL;
+
 CREATE TABLE IF NOT EXISTS code_memory_links (
     id          TEXT PRIMARY KEY,
     repo_id     TEXT NOT NULL,

engraphis/core/store.py

@@ -1075,6 +1075,8 @@ def _apply_schema(self, previous_version: int) -> None:
             self._backfill_memory_entities_v5()
         if previous_version < 5 or mem_links_need_temporal_backfill:
             self._migrate_mem_link_history_v5()
+        if previous_version < 6:
+            self._migrate_code_file_history_v6()
         # Classify pre-v3 edges. Existing rows defaulted to semantic during ALTER TABLE;
         # infer their more specific logical layer from the relationship label.
         if previous_version < 3:
@@ -1247,6 +1249,29 @@ def _migrate_mem_link_history_v5(self) -> None:
             (stamp, stamp),
         )
 
+    def _migrate_code_file_history_v6(self) -> None:
+        """Seed temporal file manifests from the v5 current-file snapshot."""
+        stamp = now_ts()
+        rows = self.conn.execute("SELECT * FROM code_files").fetchall()
+        for row in rows:
+            existing = self.conn.execute(
+                "SELECT 1 FROM code_file_history WHERE repo_id=? AND file=? "
+                "AND valid_to IS NULL AND expired_at IS NULL",
+                (row["repo_id"], row["file"]),
+            ).fetchone()
+            if existing is None:
+                started = row["indexed_at"] if row["indexed_at"] is not None else stamp
+                self.conn.execute(
+                    "INSERT INTO code_file_history("
+                    "repo_id, file, lang, content_hash, size_bytes, mtime_ns, backend, "
+                    "indexed_at, valid_from, ingested_at) VALUES (?,?,?,?,?,?,?,?,?,?)",
+                    (
+                        row["repo_id"], row["file"], row["lang"], row["content_hash"],
+                        row["size_bytes"], row["mtime_ns"], row["backend"],
+                        row["indexed_at"], started, started,
+                    ),
+                )
+
     def _backfill_claim_identity_v5(self) -> None:
         """Lift already-present metadata hints into indexed, optional claim columns."""
         rows = self.conn.execute(
@@ -2709,10 +2734,29 @@ def add_link(self, a: str, b: str, relation: str = "related",
             normalize_graph_layer(layer, relation).value
             if layer is not None else None
         )
+        graph_layer = requested_layer or normalize_graph_layer(None, relation).value
         started_transaction = not self.conn.in_transaction
         if started_transaction:
             self.conn.execute("BEGIN IMMEDIATE")
         try:
+            # A sync bundle may carry a closed link interval.  It has no live row to
+            # match below, so recognize an exact historical version before inserting
+            # it again on every replay. ``IS`` deliberately gives NULL-safe equality.
+            exact = self.conn.execute(
+                "SELECT 1 FROM mem_links "
+                "WHERE ((a=? AND b=?) OR (a=? AND b=?)) AND relation=? "
+                "AND layer=? AND reason=? AND valid_from IS ? AND valid_to IS ? "
+                "AND valid_to_recorded_at IS ? AND ingested_at IS ? AND expired_at IS ? "
+                "LIMIT 1",
+                (
+                    a, b, b, a, relation, graph_layer, reason,
+                    valid_from, valid_to, valid_to_recorded_at, ingested_at, expired_at,
+                ),
+            ).fetchone()
+            if exact is not None:
+                if started_transaction:
+                    self.conn.commit()
+                return
             existing = self.conn.execute(
                 "SELECT rowid, a, b, relation, layer, reason, created_at, "
                 "valid_from, valid_to, valid_to_recorded_at, ingested_at, expired_at "
@@ -2768,7 +2812,6 @@ def add_link(self, a: str, b: str, relation: str = "related",
                     # for ``commit=False``; the old no-op path never opened a transaction.
                     self.conn.commit()
                 return
-            graph_layer = requested_layer or normalize_graph_layer(None, relation).value
             stamp = now_ts()
             world_start = stamp if valid_from is None else valid_from
             system_start = stamp if ingested_at is None else ingested_at
@@ -2818,7 +2861,14 @@ def get_links(self, memory_id: str, *,
     def edges_in_scope(self, flt: Optional[SearchFilter] = None,
                        *, at: Optional[float] = None,
                        limit: Optional[int] = None) -> list[Edge]:
-        """Edges visible at ``at``/``filter.valid_at`` and ``filter.known_at``."""
+        """Edges visible at ``at``/``filter.valid_at`` and ``filter.known_at``.
+
+        Normalized supports are authoritative for edges that have them.  The edge row
+        aggregates its support starts for current-read efficiency, but independently
+        minimizing world and system time can fabricate a pair no source established.
+        A historical read must therefore see at least one individually visible support.
+        Legacy direct edges with no normalized support retain the edge-row fallback.
+        """
         valid_at, known_at = _temporal_anchors(flt, valid_at=at)
         sql = ("SELECT * FROM edges WHERE (valid_from IS NULL OR valid_from<=?) "
                "AND (valid_to IS NULL OR ?<valid_to "
@@ -2829,6 +2879,16 @@ def edges_in_scope(self, flt: Optional[SearchFilter] = None,
         params: list[Any] = [
             valid_at, valid_at, known_at, known_at, known_at,
         ]
+        support_visibility, support_params = _temporal_visibility_sql(
+            "s", flt, valid_at=valid_at
+        )
+        sql += (
+            " AND (NOT EXISTS (SELECT 1 FROM edge_supports any_support "
+            "WHERE any_support.edge_id=edges.id) OR EXISTS (SELECT 1 FROM "
+            "edge_supports s WHERE s.edge_id=edges.id AND "
+            + support_visibility + "))"
+        )
+        params.extend(support_params)
         if flt and flt.workspace_id:
             sql += " AND workspace_id=?"
             params.append(flt.workspace_id)
@@ -2851,9 +2911,13 @@ def edges_in_scope(self, flt: Optional[SearchFilter] = None,
     def links_among(self, ids: list[str], *,
                     layers: Optional[list[GraphLayer]] = None,
                     flt: Optional[SearchFilter] = None,
+                    include_invalid: bool = False,
                     limit: Optional[int] = None) -> list[dict]:
         """Return memory links visible under both temporal anchors.
 
+        ``include_invalid`` is for full-state replication only: a closed interval is
+        state that must synchronize even though normal graph reads do not expose it.
+
         Chunk only the indexed ``a`` side and filter ``b`` against an in-memory set.
         This keeps every statement below SQLite's portable variable limit while
         preserving exact pair semantics for graphs containing thousands of memories.
@@ -2879,15 +2943,17 @@ def links_among(self, ids: list[str], *,
             sql = (
                 "SELECT a, b, relation, layer, reason, created_at, valid_from, valid_to, "
                 "valid_to_recorded_at, ingested_at, expired_at FROM mem_links "
-                f"WHERE a IN ({marks}) "
-                f"AND {visibility_sql}"
+                f"WHERE a IN ({marks})"
             )
-            params: list[Any] = [*chunk, *visibility_params]
+            params: list[Any] = [*chunk]
+            if not include_invalid:
+                sql += f" AND {visibility_sql}"
+                params.extend(visibility_params)
             if layers is not None:
                 layer_marks = ",".join("?" for _ in layers)
                 sql += f" AND layer IN ({layer_marks})"
                 params.extend(_enum(layer) for layer in layers)
-            sql += " ORDER BY a, b, relation"
+            sql += " ORDER BY a, b, relation, valid_from, ingested_at"
             found = self.conn.execute(sql, params).fetchall()
             for row in found:
                 if row["b"] not in wanted:
@@ -2917,6 +2983,16 @@ def neighbors(self, node_ids: list[str], *, at: Optional[float] = None,
             *node_ids, *node_ids,
             valid_at, valid_at, known_at, known_at, known_at,
         ]
+        support_visibility, support_params = _temporal_visibility_sql(
+            "s", flt, valid_at=valid_at
+        )
+        sql += (
+            " AND (NOT EXISTS (SELECT 1 FROM edge_supports any_support "
+            "WHERE any_support.edge_id=edges.id) OR EXISTS (SELECT 1 FROM "
+            "edge_supports s WHERE s.edge_id=edges.id AND "
+            + support_visibility + "))"
+        )
+        params.extend(support_params)
         if layers is not None:
             if not layers:
                 return []
@@ -3009,14 +3085,22 @@ def get_code_file(self, repo_id: str, file: str) -> Optional[dict]:
 
     def list_code_files(self, repo_id: str, *,
                         languages: Optional[set] = None,
+                        flt: Optional[SearchFilter] = None,
                         limit: Optional[int] = None) -> list[dict]:
-        sql = "SELECT * FROM code_files WHERE repo_id=?"
+        """Return the current manifest, or its bi-temporal history when anchored."""
+        historical = bool(flt and flt.historical)
+        table = "code_file_history" if historical else "code_files"
+        sql = f"SELECT * FROM {table} WHERE repo_id=?"
         params: list[Any] = [repo_id]
+        if historical:
+            temporal, temporal_params = _temporal_visibility_sql("", flt)
+            sql += " AND " + temporal
+            params.extend(temporal_params)
         if languages:
             marks = ",".join("?" for _ in languages)
             sql += f" AND lang IN ({marks})"
             params.extend(sorted(languages))
-        sql += " ORDER BY file"
+        sql += " ORDER BY file" + (", version" if historical else "")
         if limit is not None:
             sql += " LIMIT ?"
             params.append(max(0, int(limit)))  # never -1 == SQLite "unlimited"
@@ -3025,6 +3109,34 @@ def list_code_files(self, repo_id: str, *,
     def upsert_code_file(self, *, repo_id: str, file: str, lang: str,
                          content_hash: str, size_bytes: int, mtime_ns: int,
                          backend: str, commit: bool = True) -> None:
+        stamp = now_ts()
+        current_history = self.conn.execute(
+            "SELECT version, lang, content_hash, size_bytes, mtime_ns, backend "
+            "FROM code_file_history WHERE repo_id=? AND file=? "
+            "AND valid_to IS NULL AND expired_at IS NULL",
+            (repo_id, file),
+        ).fetchone()
+        unchanged = current_history is not None and (
+            current_history["lang"], current_history["content_hash"],
+            int(current_history["size_bytes"] or 0), int(current_history["mtime_ns"] or 0),
+            current_history["backend"] or "",
+        ) == (lang, content_hash, int(size_bytes), int(mtime_ns), backend)
+        if not unchanged:
+            if current_history is not None:
+                self.conn.execute(
+                    "UPDATE code_file_history SET valid_to=?, valid_to_recorded_at=? "
+                    "WHERE version=?",
+                    (stamp, stamp, current_history["version"]),
+                )
+            self.conn.execute(
+                "INSERT INTO code_file_history("
+                "repo_id, file, lang, content_hash, size_bytes, mtime_ns, backend, "
+                "indexed_at, valid_from, ingested_at) VALUES (?,?,?,?,?,?,?,?,?,?)",
+                (
+                    repo_id, file, lang, content_hash, int(size_bytes), int(mtime_ns),
+                    backend, stamp, stamp, stamp,
+                ),
+            )
         self.conn.execute(
             "INSERT INTO code_files(repo_id, file, lang, content_hash, size_bytes, "
             "mtime_ns, backend, indexed_at) VALUES (?,?,?,?,?,?,?,?) "
@@ -3033,13 +3145,19 @@ def upsert_code_file(self, *, repo_id: str, file: str, lang: str,
             "size_bytes=excluded.size_bytes, mtime_ns=excluded.mtime_ns, "
             "backend=excluded.backend, indexed_at=excluded.indexed_at",
             (repo_id, file, lang, content_hash, int(size_bytes), int(mtime_ns),
-             backend, now_ts()),
+             backend, stamp),
         )
         if commit:
             self.conn.commit()
 
     def remove_code_file(self, repo_id: str, file: str, *, commit: bool = True) -> None:
         self.clear_symbols_for_file(repo_id, file, commit=False)
+        stamp = now_ts()
+        self.conn.execute(
+            "UPDATE code_file_history SET valid_to=?, valid_to_recorded_at=? "
+            "WHERE repo_id=? AND file=? AND valid_to IS NULL AND expired_at IS NULL",
+            (stamp, stamp, repo_id, file),
+        )
         self.conn.execute("DELETE FROM code_files WHERE repo_id=? AND file=?", (repo_id, file))
         if commit:
             self.conn.commit()