EtanHey · EtanHey · Aug 1, 2026 · Aug 1, 2026 · Aug 1, 2026 · Aug 1, 2026
@@ -16,7 +16,7 @@ dependencies = [
     "sentence-transformers>=2.2.0",
 
     # MCP server
-    "mcp>=1.0.0",
+    "mcp>=1.0.0,<2",
 
     # CLI
     "typer>=0.9.0",

@@ -0,0 +1,239 @@
+#!/usr/bin/env python3
+"""Re-tag existing Codex chunks whose session is explicitly linked by T3."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import sqlite3
+import sys
+import tempfile
+from pathlib import Path
+from typing import Any
+
+sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src"))
+
+from brainlayer.provenance import PROVENANCE_RANK
+from brainlayer.t3_provenance import T3_APP_SESSION, codex_session_id_from_source, t3_app_codex_session_ids
+
+
+def _candidates(connection: sqlite3.Connection, linked_session_ids: set[str]) -> list[tuple[str, str | None]]:
+    rows = connection.execute(
+        """
+        SELECT id, source_file, provenance_class
+        FROM chunks
+        WHERE source_file LIKE '%/.codex/sessions/%'
+        ORDER BY id
+        """
+    )
+    return [
+        (chunk_id, provenance_class)
+        for chunk_id, source_file, provenance_class in rows
+        if codex_session_id_from_source(source_file) in linked_session_ids
+        and provenance_class == "codex-session"
+        and provenance_class not in PROVENANCE_RANK
+    ]
+
+
+def _atomic_write_jsonl(path: Path, rows: list[tuple[str, str | None]]) -> None:
+    """Durably replace a JSONL artifact without exposing a partial file."""
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temp_fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", suffix=".tmp", dir=path.parent)
+    temp_path = Path(temp_name)
+    try:
+        with os.fdopen(temp_fd, "w", encoding="utf-8") as artifact:
+            for chunk_id, provenance_class in rows:
+                artifact.write(json.dumps({"id": chunk_id, "provenance_class": provenance_class}) + "\n")
+            artifact.flush()
+            os.fsync(artifact.fileno())
+        os.replace(temp_path, path)
+        directory_fd = os.open(path.parent, os.O_RDONLY)
+        try:
+            os.fsync(directory_fd)
+        finally:
+            os.close(directory_fd)
+    finally:
+        if temp_path.exists():
+            temp_path.unlink()
+
+
+def _write_rollback_artifact(path: Path, candidates: list[tuple[str, str | None]]) -> None:
+    existing = (
+        {
+            row["id"]: row["provenance_class"]
+            for line in path.read_text(encoding="utf-8").splitlines()
+            if line
+            for row in [json.loads(line)]
+        }
+        if path.exists()
+        else {}
+    )
+    existing.update(dict(candidates))
+    _atomic_write_jsonl(path, sorted(existing.items()))
+
+
+def retag_t3_app_chunks(
+    *,
+    db_path: str | Path,
+    state_db: str | Path,
+    apply: bool = False,
+    rollback_artifact: str | Path | None = None,
+    batch_size: int = 5_000,
+) -> dict[str, int]:
+    """Report or apply the deterministic T3 provenance re-tagging operation."""
+    if batch_size <= 0:
+        raise ValueError("batch_size must be positive")
+    if apply and rollback_artifact is None:
+        raise ValueError("--rollback-artifact is required with --apply")
+
+    linked_session_ids = t3_app_codex_session_ids(state_db)
+    path = Path(db_path).expanduser()
+    connection = (
+        sqlite3.connect(path, timeout=1.0)
+        if apply
+        else sqlite3.connect(f"{path.absolute().as_uri()}?mode=ro&immutable=0", uri=True, timeout=1.0)
+    )
+    try:
+        if apply:
+            connection.execute("PRAGMA busy_timeout = 30000")
+        candidates = _candidates(connection, linked_session_ids)
+        report = {
+            "linked_sessions": len(linked_session_ids),
+            "candidate_chunks": len(candidates),
+            "matched_chunks": 0,
+            "retagged_chunks": 0,
+        }
+        if not apply:
+            return report
+
+        artifact_path = Path(rollback_artifact).expanduser()
+        _write_rollback_artifact(artifact_path, candidates)
+        connection.execute("PRAGMA wal_checkpoint(FULL)")
+        for batch_start in range(0, len(candidates), batch_size):
+            batch = candidates[batch_start : batch_start + batch_size]
+            cursor = connection.executemany(
+                "UPDATE chunks SET provenance_class = ? WHERE id = ? AND provenance_class = ?",
+                [(T3_APP_SESSION, chunk_id, "codex-session") for chunk_id, _ in batch],
+            )
+            report["matched_chunks"] += cursor.rowcount
+            connection.commit()
+            if ((batch_start // batch_size) + 1) % 3 == 0:
+                connection.execute("PRAGMA wal_checkpoint(FULL)")
+        connection.execute("PRAGMA wal_checkpoint(FULL)")
+        report["retagged_chunks"] = report["matched_chunks"]
+        return report
+    finally:
+        connection.close()
+
+
+def rollback_t3_app_chunks(
+    *,
+    db_path: str | Path,
+    rollback_artifact: str | Path,
+    only_null_prior_values: bool = False,
+    pre_restore_artifact: str | Path | None = None,
+    batch_size: int = 5_000,
+) -> dict[str, int]:
+    """Restore provenance values recorded by a prior T3 re-tag artifact."""
+    if batch_size <= 0:
+        raise ValueError("batch_size must be positive")
+
+    artifact_path = Path(rollback_artifact).expanduser()
+    rows = [json.loads(line) for line in artifact_path.read_text(encoding="utf-8").splitlines() if line]
+    candidates = [(row["id"], row["provenance_class"]) for row in rows]
+    if only_null_prior_values:
+        candidates = [
+            (chunk_id, provenance_class) for chunk_id, provenance_class in candidates if provenance_class is None
+        ]
+    if pre_restore_artifact is not None:
+        pre_restore_path = Path(pre_restore_artifact).expanduser()
+        if pre_restore_path.exists():
+            raise ValueError(f"pre-restore artifact already exists: {pre_restore_path}")
+    else:
+        pre_restore_path = None
+    connection = sqlite3.connect(Path(db_path).expanduser(), timeout=1.0)
+    try:
+        connection.execute("PRAGMA busy_timeout = 30000")
+        current_values = (
+            {
+                chunk_id: provenance_class
+                for chunk_id, provenance_class in connection.execute(
+                    "SELECT id, provenance_class FROM chunks WHERE id IN ({})".format(
+                        ", ".join("?" for _ in candidates)
+                    ),
+                    [chunk_id for chunk_id, _ in candidates],
+                )
+            }
+            if candidates
+            else {}
+        )
+        missing_chunks = sum(chunk_id not in current_values for chunk_id, _ in candidates)
+        matched_candidates = [
+            (chunk_id, provenance_class)
+            for chunk_id, provenance_class in candidates
+            if current_values.get(chunk_id) == T3_APP_SESSION
+        ]
+        skipped_non_t3_chunks = len(candidates) - missing_chunks - len(matched_candidates)
+        if pre_restore_path is not None and matched_candidates:
+            _atomic_write_jsonl(
+                pre_restore_path,
+                [(chunk_id, current_values[chunk_id]) for chunk_id, _ in matched_candidates],
+            )
+        restored_chunks = 0
+        for batch_start in range(0, len(matched_candidates), batch_size):
+            batch = matched_candidates[batch_start : batch_start + batch_size]
+            cursor = connection.executemany(
+                "UPDATE chunks SET provenance_class = ? WHERE id = ? AND provenance_class = ?",
+                [(provenance_class, chunk_id, T3_APP_SESSION) for chunk_id, provenance_class in batch],
+            )
+            restored_chunks += cursor.rowcount
+            connection.commit()
+            if ((batch_start // batch_size) + 1) % 3 == 0:
+                connection.execute("PRAGMA wal_checkpoint(FULL)")
+        connection.execute("PRAGMA wal_checkpoint(FULL)")
+        return {
+            "requested_chunks": len(candidates),
+            "matched_chunks": len(matched_candidates),
+            "restored_chunks": restored_chunks,
+            "missing_chunks": missing_chunks,
+            "skipped_non_t3_chunks": skipped_non_t3_chunks,
+        }
+    finally:
+        connection.close()
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--db-path", type=Path, default=Path.home() / ".local/share/brainlayer/brainlayer.db")
+    parser.add_argument("--state-db", type=Path, default=Path.home() / ".t3/userdata/state.sqlite")
+    parser.add_argument("--apply", action="store_true", help="Perform writes; default is read-only dry run")
+    parser.add_argument("--rollback", action="store_true", help="Restore values from --rollback-artifact")
+    parser.add_argument("--only-null-prior-values", action="store_true")
+    parser.add_argument("--pre-restore-artifact", type=Path)
+    parser.add_argument("--rollback-artifact", type=Path, help="JSONL (id, provenance_class) captured before writes")
+    parser.add_argument("--batch-size", type=int, default=5_000)
+    args = parser.parse_args()
+    if args.rollback:
+        if args.rollback_artifact is None:
+            parser.error("--rollback-artifact is required with --rollback")
+        report: dict[str, Any] = rollback_t3_app_chunks(
+            db_path=args.db_path,
+            rollback_artifact=args.rollback_artifact,
+            only_null_prior_values=args.only_null_prior_values,
+            pre_restore_artifact=args.pre_restore_artifact,
+            batch_size=args.batch_size,
+        )
+    else:
+        report = retag_t3_app_chunks(
+            db_path=args.db_path,
+            state_db=args.state_db,
+            apply=args.apply,
+            rollback_artifact=args.rollback_artifact,
+            batch_size=args.batch_size,
+        )
+    print(json.dumps(report, sort_keys=True))
+
+
+if __name__ == "__main__":
+    main()
@@ -16,6 +16,7 @@
 
 from .content_class import normalize_content_class
 from .ingest_denylist import is_denylisted
+from .t3_provenance import T3_APP_SESSION, is_t3_app_initiated_codex_session
 
 SearchPolicy = Literal["KEEP", "ISOLATE", "OUT"]
 EffectiveVisibility = Literal["default", "operational", "cold"]
@@ -138,11 +139,23 @@ def classify_provenance(
     content_class: str | None = None,
     *,
     content: str | None = None,
+    t3_state_db: str | Path | None = None,
+    t3_linked_session_ids: set[str] | None = None,
 ) -> ProvenanceDecision:
     """Classify a source path into an auditable provenance search policy."""
     del content_class
     path = _abspath(source_file)
 
+    is_t3_app_session = _under_provider_sessions(path, ".codex") and (
+        is_t3_app_initiated_codex_session(
+            source_file,
+            state_db=t3_state_db,
+            linked_session_ids=t3_linked_session_ids,
+        )
+    )
+    if is_t3_app_session:
+        return ProvenanceDecision(T3_APP_SESSION, "KEEP", "T3 runtime cursor links Codex session")
+
     if has_recon_agent_signature(content) or _has_recon_path_signature(path):
         return ProvenanceDecision("recon-agent", "OUT", "recon Agent-tool signature wins precedence")
 

@@ -0,0 +1,97 @@
+"""Read-only discriminator for Codex sessions initiated by the T3 Code app."""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import sqlite3
+from pathlib import Path
+
+from .alarm import raise_alarm
+
+T3_APP_SESSION = "t3-app-session"
+DEFAULT_T3_STATE_DB = Path.home() / ".t3" / "userdata" / "state.sqlite"
+_CODEX_SESSION_ID_RE = re.compile(r"[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}", re.IGNORECASE)
+_REQUIRED_RUNTIME_COLUMNS = frozenset({"thread_id", "provider_name", "resume_cursor_json"})
+
+
+def codex_session_id_from_source(source_file: str | Path) -> str | None:
+    """Return the final UUID in a Codex JSONL filename, if present."""
+    matches = _CODEX_SESSION_ID_RE.findall(Path(source_file).stem)
+    return matches[-1].lower() if matches else None
+
+
+def is_t3_app_initiated_codex_session(
+    source_file: str | Path,
+    *,
+    state_db: str | Path | None = None,
+    linked_session_ids: set[str] | None = None,
+) -> bool:
+    """Return whether a Codex transcript is explicitly linked by T3 runtime state.
+
+    A missing T3 database means there is no local T3 app installation to link
+    against. An existing database with a changed or unreadable schema is fatal:
+    silently treating those sessions as ordinary Codex would reintroduce the
+    provenance collision this module prevents.
+    """
+    session_id = codex_session_id_from_source(source_file)
+    if session_id is None:
+        return False
+
+    if linked_session_ids is not None:
+        return session_id in linked_session_ids
+
+    path = Path(state_db or os.environ.get("BRAINLAYER_T3_STATE_DB", DEFAULT_T3_STATE_DB)).expanduser()
+    if not path.exists():
+        return False
+    return session_id in t3_app_codex_session_ids(path)
+
+
+def t3_app_codex_session_ids(state_db: str | Path = DEFAULT_T3_STATE_DB) -> set[str]:
+    """Return Codex session IDs explicitly linked by T3 runtime cursors."""
+    path = Path(state_db).expanduser()
+
+    try:
+        connection = sqlite3.connect(f"{path.absolute().as_uri()}?mode=ro&immutable=0", uri=True, timeout=1.0)
+    except sqlite3.Error as exc:
+        raise_alarm(
+            "t3_runtime_unavailable",
+            "could not open T3 runtime state read-only",
+            {"path": str(path), "error": str(exc)},
+        )
+
+    try:
+        columns = {row[1] for row in connection.execute("PRAGMA table_info(provider_session_runtime)")}
+        missing = sorted(_REQUIRED_RUNTIME_COLUMNS - columns)
+        if missing:
+            raise_alarm(
+                "t3_runtime_schema_drift",
+                "provider_session_runtime no longer exposes the required T3 linkage columns",
+                {"path": str(path), "missing_columns": missing},
+            )
+
+        session_ids: set[str] = set()
+        for provider_name, resume_cursor_json in connection.execute(
+            "SELECT provider_name, resume_cursor_json FROM provider_session_runtime WHERE provider_name = ?", ("codex",)
+        ):
+            try:
+                resume_cursor = json.loads(resume_cursor_json)
+            except (TypeError, json.JSONDecodeError) as exc:
+                raise_alarm(
+                    "t3_runtime_linkage_invalid",
+                    "provider_session_runtime.resume_cursor_json is not valid JSON",
+                    {"path": str(path), "provider_name": provider_name, "error": str(exc)},
+                )
+            thread_id = resume_cursor.get("threadId") if isinstance(resume_cursor, dict) else None
+            if isinstance(thread_id, str) and _CODEX_SESSION_ID_RE.fullmatch(thread_id):
+                session_ids.add(thread_id.lower())
+        return session_ids
+    except sqlite3.Error as exc:
+        raise_alarm(
+            "t3_runtime_schema_drift",
+            "could not query provider_session_runtime for T3 provenance",
+            {"path": str(path), "error": str(exc)},
+        )
+    finally:
+        connection.close()