diff --git a/docs.local/tasks/REPORT.md b/docs.local/tasks/REPORT.md new file mode 100644 index 00000000..a9fc4d02 --- /dev/null +++ b/docs.local/tasks/REPORT.md @@ -0,0 +1,177 @@ +# D4-T3 ingestion iterate report + +Status: iteration implemented and project-UUID regression repaired on +`feat/d4-t3-thread-ingest`; production entrypoint executed; no push or PR. + +## Review blockers resolved + +- `brainlayer ingest-t3` is now a first-class CLI entrypoint. The dedicated + `scripts/launchd/com.brainlayer.t3-ingest.plist` runs it at load and daily at + 03:45; `scripts/launchd/install.sh all` installs it. +- Provider runtime linkage now reaches every emitted chunk as + `t3_provider_name`, `t3_provider_session_id`, and `t3_mirrored` metadata. +- The fatal schema contract is limited to columns actually consumed: + `projection_threads(thread_id, project_id, title, created_at)`, + `projection_thread_messages(message_id, thread_id, role, text, created_at)`, + `projection_projects(project_id, title)`, + and `provider_session_runtime(thread_id, provider_name, resume_cursor_json)`. + `projection_thread_sessions` and all `updated_at` columns are no longer + required. +- The dead `read_t3_threads` export was removed. +- `health-check` accepts `--t3-health-path`, includes the JSON snapshot in its + result, and raises a critical `t3_ingest_unhealthy` issue when `alerting` is + true. +- SQLite/DB sources no longer pass a 704 MB binary file through the JSONL + timestamp sniffer; T3 chunk metadata supplies its timestamps. + +Relevant implementation locations are `src/brainlayer/cli/__init__.py:3681-3715`, +`scripts/launchd/com.brainlayer.t3-ingest.plist`, +`src/brainlayer/ingest/t3.py:28-33,156-235,295-317`, and +`src/brainlayer/health_check.py:779-795,961-964` (iteration changes; final +branch commit is recorded below). The original reader’s read-only and alarm +primitives remain at `src/brainlayer/ingest/t3.py:156-180`. + +## Re-verified live source facts + +The source is opened with SQLite URI `mode=ro&immutable=0`, autocommit, +`PRAGMA query_only=ON`, and a 1-second busy timeout. The production source +remained read-only during the live ingest. + +| Fact | Observed | +|---|---:| +| `~/.t3/userdata/state.sqlite` size | 704,430,080 bytes | +| `projection_threads` | 45 | +| `projection_thread_messages` | 2,349 | +| `projection_thread_sessions` | 44 (not required by the reader) | +| `projection_thread_activities` | 38,381 | +| `provider_session_runtime` | 34 | + +## Schema map and provider linkage + +| Table/column | Meaning and relationship | +|---|---| +| `projection_threads.thread_id` | Thread identity and primary join key. | +| `projection_threads.project_id`, `title`, `created_at` | Thread identity/title/timestamp fields consumed by the reader. | +| `projection_projects.project_id`, `title` | Human project-name lookup keyed by `projection_threads.project_id`. | +| `projection_thread_messages.message_id` | Message identity. | +| `projection_thread_messages.thread_id` | Message-to-thread link to `projection_threads.thread_id`. | +| `projection_thread_messages.role`, `text`, `created_at` | Message payload and ordering fields consumed. | +| `provider_session_runtime.thread_id` | Runtime-mirror-to-thread link; a matching row marks the thread mirrored. | +| `provider_session_runtime.provider_name` | Provider name persisted to chunk metadata. | +| `provider_session_runtime.resume_cursor_json.threadId` | Provider session ID persisted to `t3_provider_session_id` when present. | + +The reader validates this consumed contract before snapshot queries. Missing +tables/columns raise the existing `t3_schema_drift` alarm and write an alerting +health snapshot before raising. The 34 runtime rows are accepted as duplicate +content by design; no mirror exclusion or source-specific dedup heuristic was +added. + +## Project-name derivation and live repair + +The initial live ingest exposed a regression: the prior reader copied the raw +T3 UUID into `chunks.project` (`src/brainlayer/ingest/t3.py:301-308 @ 74baae28`). +The fix joins `projection_projects` on `project_id` and chooses `title` because +it is T3’s curated human-facing project label; it avoids inferring a name from +filesystem path components. A missing mapping yields `None`, never the UUID. +The raw source database remains read-only. + +Before the repair, the canonical DB contained 2,505 `source='t3'` rows with +UUID projects: 2,338 still had `provenance_class='t3-thread'`, and 167 had +subsequently been reclassified by enrichment (`AGENT-INFERENCE` 147, +`AGENT-PARAPHRASE` 4, `RAW-ETAN-DIRECT` 16). The repair intentionally covered +all 2,505 T3 rows so the project invariant applies regardless of later +provenance classification. + +| Repair metric | Count | +|---|---:| +| T3 rows examined | 2,505 | +| Rows re-derived | 2,505 | +| Rows left alone | 0 | +| Missing project mappings | 0 | +| Batches | 1 × 5,000-row batch | +| UUID projects remaining after repair | 0 | + +The rollback artifact was captured before the write at +`/Users/etanheyman/.local/share/brainlayer/t3-project-uuid-rollback-74baae28.jsonl`: +2,505 `(id, project)` records, SHA-256 +`c7c5eee753cc9c8ab5174c4770ec5a5fc16c89e9b3d3999e0e673cc0e99ab928`. +Enrichment, watcher, drain, index, and hotlane writers were paused; WAL +checkpoint before the repair returned `0|2863|2863`, and the post-repair +checkpoint returned `0|2273|2273`. All paused services were restored afterward. + +## The 11 unmirrored threads + +“Unmirrored” means no matching `provider_session_runtime` row. It is not an +exclusion rule. + +| Thread ID | Created | Messages | +|---|---|---:| +| `a7b35b2a-50f2-4e8c-83a5-747e7a29757c` | 2026-03-06T17:45:31.449Z | 1 | +| `2b5cad3c-eefb-4766-949b-01ffcdfcfbf5` | 2026-03-07T18:21:10.185Z | 1 | +| `e9cf5dd4-a039-4dec-bfe8-d717dd2e9c23` | 2026-03-07T18:23:16.780Z | 1 | +| `44dd2387-fbf5-4b92-b778-b1f5487a15f0` | 2026-03-07T18:25:01.591Z | 1 | +| `263650ba-2190-43f7-a32a-2a87975a7d5e` | 2026-03-07T18:28:00.738Z | 1 | +| `7eb8d353-7eb2-4420-b7ad-e31926128aef` | 2026-03-07T18:28:53.755Z | 0 | +| `de99ab14-0595-42da-b3f1-cf863e8d5835` | 2026-03-07T18:29:26.869Z | 1 | +| `da804581-3b59-49e3-bef1-96e9f851a1f1` | 2026-03-07T19:08:49.225Z | 1 | +| `1babbcbf-0e37-40a4-bb14-14b1b336d542` | 2026-03-07T19:13:08.735Z | 1 | +| `d762ec6e-cb56-41a2-aea6-46dd547f2e75` | 2026-03-07T19:16:05.780Z | 1 | +| `95eee65c-59b3-49c7-9f3f-5051ead264ca` | 2026-03-07T19:17:18.220Z | 1 | + +## Real production invocation + +The installed console entrypoint was run with the checked-out source: + +```text +PYTHONPATH=src brainlayer ingest-t3 \ + --state-db /Users/etanheyman/.t3/userdata/state.sqlite \ + --db /Users/etanheyman/.local/share/brainlayer/brainlayer.db \ + --health-path /Users/etanheyman/.local/share/brainlayer/t3-health.json +``` + +| Metric | Count | +|---|---:| +| Threads seen / ingested | 45 / 45 | +| Messages seen / ingested | 2,349 / 2,349 | +| Chunks planned | 2,506 | +| Chunks indexed | 2,505 | +| Mirrored threads accepted as duplicates | 34 | +| T3 messages skipped by role/empty-text policy | 0 | +| Chunks filtered by shared system-prompt guard | 1 | + +The one filtered chunk is message +`c65c95ed-2d60-4f6d-92a6-8a2ff6ad0b40` in thread +`f2a72bc0-1abc-450d-ae50-ab6b3e9812dd`; its user text begins with agent +instruction scaffolding and matched the existing global guard. This is why the +indexed count is one below the planned count; it was not a T3 reader failure. + +Read-only post-ingest verification of the canonical BrainLayer DB found: + +- `2,505` T3 source rows; enrichment later reclassified 167, leaving `2,338` + with `provenance_class = 't3-thread'`; +- `44` nonempty T3 conversations; +- `2,495` mirrored rows and `2,487` rows carrying a non-null provider session ID; +- `t3-health.json` with `alerting=false`, `threads_seen=45`, + `messages_seen=2349`, `chunks_planned=2506`, and `chunks_indexed=2505`. +- After the project repair: 2,505 T3 source rows, 15 distinct human project + titles, 0 UUID projects, and 0 NULL projects. + +## Verification + +Passed: + +- Focused and cross-cutting suite: **82 passed**. +- Isolated baseline phase-2 queue file: **11 passed**. +- Live production invocation and read-only destination/source verification. + +The full `pytest -q` run reached the repository’s existing phase-2 queue +failure at 68%, then cascaded into file-descriptor errors (`OSError: [Errno +24] Too many open files`) during temporary-directory cleanup. The isolated +phase-2 file passes, and no phase-2 queue file changed in this iteration. + +## Commit + +The implementation and report are committed on the branch; no push or PR was +performed. + +TASK_DONE diff --git a/scripts/launchd/com.brainlayer.t3-ingest.plist b/scripts/launchd/com.brainlayer.t3-ingest.plist new file mode 100644 index 00000000..291508b6 --- /dev/null +++ b/scripts/launchd/com.brainlayer.t3-ingest.plist @@ -0,0 +1,53 @@ + + + + + Label + com.brainlayer.t3-ingest + ProgramArguments + + __BRAINLAYER_ENV_RUN__ + __BRAINLAYER_BIN__ + ingest-t3 + + StartCalendarInterval + + Hour + 3 + Minute + 45 + + RunAtLoad + + StandardOutPath + __HOME__/Library/Logs/brainlayer/t3-ingest.out.log + StandardErrorPath + __HOME__/Library/Logs/brainlayer/t3-ingest.err.log + EnvironmentVariables + + PATH + /usr/local/bin:/usr/bin:/bin:/usr/sbin:/sbin:__HOME__/.local/bin + PYTHONUNBUFFERED + 1 + BRAINLAYER_REPO_ROOT + __BRAINLAYER_DIR__ + BRAINLAYER_ENV_FILE + __BRAINLAYER_ENV_FILE__ + BRAINLAYER_LAUNCHD_SERVICE + t3-ingest + + Nice + 10 + ProcessType + Background + ExitTimeOut + 1800 + LowPriorityIO + + SoftResourceLimits + + NumberOfFiles + 4096 + + + diff --git a/scripts/launchd/install.sh b/scripts/launchd/install.sh index ac97c6c2..1abbeac1 100755 --- a/scripts/launchd/install.sh +++ b/scripts/launchd/install.sh @@ -20,6 +20,7 @@ # ./scripts/launchd/install.sh throughput-watchdog # Install watcher throughput watchdog only # ./scripts/launchd/install.sh hotlane # Install BrainBar hotlane embed/enrich daemon only # ./scripts/launchd/install.sh p0-counter # Install daily P0 longitudinal counter only +# ./scripts/launchd/install.sh t3-ingest # Install T3 thread ingestion only # ./scripts/launchd/install.sh remove # Unload and remove all set -euo pipefail @@ -661,13 +662,16 @@ case "${1:-all}" in p0-counter) install_plist p0-counter ;; + t3-ingest) + install_plist t3-ingest + ;; all) install_env_runner verify_config_file verify_gemini_env_file failures=0 enrichment_ok=0 - if ! install_many index drain watch hotlane-brainbar; then + if ! install_many index t3-ingest drain watch hotlane-brainbar; then failures=1 fi if install_plist enrichment; then @@ -707,6 +711,7 @@ case "${1:-all}" in ;; remove) remove_plist index + remove_plist t3-ingest 2>/dev/null || true remove_plist enrich 2>/dev/null || true remove_plist enrichment 2>/dev/null || true remove_plist watch 2>/dev/null || true @@ -730,7 +735,7 @@ case "${1:-all}" in rm -f "$HOTLANE_BRAINBAR_DST" ;; *) - echo "Usage: $0 [index|watch|enrich|enrichment|decay|drain|hotlane|repair-fts|load [name]|unload [name]|checkpoint|backup|jsonl-backup|maintenance|maintenance-nightly|maintenance-weekly|health-check|tier0-watchdog|throughput-watchdog|p0-counter|all|remove]" + echo "Usage: $0 [index|t3-ingest|watch|enrich|enrichment|decay|drain|hotlane|repair-fts|load [name]|unload [name]|checkpoint|backup|jsonl-backup|maintenance|maintenance-nightly|maintenance-weekly|health-check|tier0-watchdog|throughput-watchdog|p0-counter|all|remove]" exit 1 ;; esac diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py index c9ff3ead..99591db8 100644 --- a/src/brainlayer/cli/__init__.py +++ b/src/brainlayer/cli/__init__.py @@ -177,6 +177,7 @@ def _default_pause_labels() -> list[str]: return [ "com.brainlayer.watch", "com.brainlayer.drain", + "com.brainlayer.t3-ingest", "com.brainlayer.hotlane-brainbar", "com.brainlayer.enrichment", "com.brainlayer.health-check", @@ -845,6 +846,11 @@ def health_check_command( "--drain-health-path", help="Drain health JSON path.", ), + t3_health_path: Path = typer.Option( + Path("~/.local/share/brainlayer/t3-health.json"), + "--t3-health-path", + help="T3 ingestion health JSON path.", + ), queue_dir: Path = typer.Option(Path("~/.brainlayer/queue"), "--queue-dir", help="Durable queue directory."), offsets_path: Path = typer.Option( Path("~/.local/share/brainlayer/offsets.json"), @@ -880,6 +886,7 @@ def health_check_command( else HealthCheckConfig().source_jsonl_globs, pause_sentinel_path=pause_sentinel_path.expanduser(), drain_health_path=drain_health_path.expanduser(), + t3_health_path=t3_health_path.expanduser(), queue_dir=queue_dir.expanduser(), offsets_path=offsets_path.expanduser(), watcher_health_path=watcher_health_path.expanduser(), @@ -3671,6 +3678,43 @@ def progress_callback(embedded_count, total_embed): raise typer.Exit(1) +@app.command("ingest-t3") +def ingest_t3_command( + state_db: Path = typer.Option( + Path("~/.t3/userdata/state.sqlite"), + "--state-db", + help="Read-only T3 state SQLite database.", + ), + db: Optional[Path] = typer.Option(None, "--db", help="BrainLayer destination database."), + health_path: Path = typer.Option( + Path("~/.local/share/brainlayer/t3-health.json"), + "--health-path", + help="T3 ingestion health JSON path.", + ), + dry_run: bool = typer.Option(False, "--dry-run", help="Read and plan chunks without writing BrainLayer."), +) -> None: + """Ingest the live T3 thread projection into BrainLayer.""" + from ..ingest.t3 import ingest_t3 + from ..paths import get_db_path + + result = ingest_t3( + state_db.expanduser(), + db_path=(db.expanduser() if db is not None else get_db_path()), + health_path=health_path.expanduser(), + dry_run=dry_run, + ) + mode = "dry-run " if dry_run else "" + rprint( + f"{mode}threads_seen={result.threads_seen} " + f"threads_ingested={result.threads_ingested} " + f"messages_seen={result.messages_seen} " + f"messages_ingested={result.messages_ingested} " + f"chunks_planned={result.chunks_planned} " + f"chunks_indexed={result.chunks_indexed} " + f"duplicates_accepted={result.duplicates_accepted}" + ) + + @app.command("ingest-codex") def ingest_codex( path: Optional[Path] = typer.Argument( diff --git a/src/brainlayer/health_check.py b/src/brainlayer/health_check.py index e1bb402f..2714a60a 100644 --- a/src/brainlayer/health_check.py +++ b/src/brainlayer/health_check.py @@ -135,6 +135,7 @@ class HealthCheckConfig: drain_health_path: Path = field( default_factory=lambda: Path("~/.local/share/brainlayer/drain-health.json").expanduser() ) + t3_health_path: Path = field(default_factory=lambda: Path("~/.local/share/brainlayer/t3-health.json").expanduser()) queue_dir: Path = field(default_factory=lambda: Path("~/.brainlayer/queue").expanduser()) pending_stores_path: Path = field( default_factory=lambda: Path("~/.local/share/brainlayer/pending-stores.jsonl").expanduser() @@ -196,6 +197,7 @@ class HealthCheckResult: duration_seconds: float = 0.0 slow_check: bool = False slow_check_stage: str | None = None + t3_health: dict[str, Any] | None = None def to_dict(self) -> dict[str, Any]: return asdict(self) @@ -772,6 +774,25 @@ def _load_json(path: Path) -> dict[str, Any]: return payload if isinstance(payload, dict) else {} +def _t3_health_issue(payload: dict[str, Any]) -> HealthIssue | None: + """Turn the T3 adapter's durable health snapshot into a check issue.""" + if not payload.get("alerting"): + return None + reasons = payload.get("alert_reasons") + reason_text = ", ".join(str(reason) for reason in reasons if reason) if isinstance(reasons, list) else "unknown" + failures = payload.get("failures") + failure_text = "" + if isinstance(failures, list) and failures: + latest = failures[-1] + if isinstance(latest, dict): + failure_text = f" ({latest.get('message') or latest.get('code') or 'reader failure'})" + return HealthIssue( + "t3_ingest_unhealthy", + "critical", + f"T3 ingestion health alert: {reason_text}{failure_text}", + ) + + def _parse_iso_datetime(value: Any) -> datetime | None: if not isinstance(value, str) or not value: return None @@ -935,6 +956,11 @@ def deadline_reached(stage: str) -> HealthCheckResult | None: return None return finish_slow(stage, f"health-check exceeded {config.max_duration_seconds:.0f}s during {stage}") + t3_health = _load_json(config.t3_health_path) + result.t3_health = t3_health or None + if t3_issue := _t3_health_issue(t3_health): + add_issue(t3_issue.code, t3_issue.severity, t3_issue.message) + ps_output = ps_output_fn() lock_holder = None if ps_output is None: diff --git a/src/brainlayer/index_new.py b/src/brainlayer/index_new.py index 93243838..b6084d8f 100644 --- a/src/brainlayer/index_new.py +++ b/src/brainlayer/index_new.py @@ -58,22 +58,25 @@ def embedding_progress(completed: int, total: int) -> None: if not embedded_chunks: return 0 - # Try to get timestamp from source file (first JSONL message) + # Try to get timestamp from a text source (first JSONL message). SQLite + # adapters already put timestamps in chunk metadata; opening a live DB as + # JSONL would wastefully read binary data and can block on a large file. created_at = None - try: - import json as _json - - with open(source_file) as _f: - for _line in _f: - _line = _line.strip() - if not _line: - continue - _data = _json.loads(_line) - if "timestamp" in _data: - created_at = _data["timestamp"] - break - except Exception as e: - logger.debug("Could not extract timestamp from %s: %s", source_file, e) + if Path(source_file).suffix.lower() not in {".sqlite", ".db"}: + try: + import json as _json + + with open(source_file) as _f: + for _line in _f: + _line = _line.strip() + if not _line: + continue + _data = _json.loads(_line) + if "timestamp" in _data: + created_at = _data["timestamp"] + break + except Exception as e: + logger.debug("Could not extract timestamp from %s: %s", source_file, e) if not created_at: from datetime import datetime, timezone @@ -90,10 +93,10 @@ def embedding_progress(completed: int, total: int) -> None: for i, ec in enumerate(embedded_chunks): chunk = ec.chunk - - chunk_id = f"{source_file}:{i}" - conversation_id = chunk.metadata.get("session_id") or file_stem metadata = dict(chunk.metadata) + + chunk_id = metadata.get("chunk_id") or f"{source_file}:{i}" + conversation_id = metadata.get("conversation_id") or metadata.get("session_id") or file_stem if claude_conversation_id: metadata["claude_conversation_id"] = claude_conversation_id @@ -103,15 +106,18 @@ def embedding_progress(completed: int, total: int) -> None: "content": chunk.content, "metadata": metadata, "source_file": source_file, - "project": project, + "project": metadata.get("project") or project, "content_type": chunk.content_type.value, "value_type": chunk.value.value, "char_count": chunk.char_count, - "created_at": created_at, + "created_at": metadata.get("created_at") or created_at, "conversation_id": conversation_id, "position": i, "sender": metadata.get("sender"), "source": metadata.get("source", "claude_code"), + "provenance_class": metadata.get("provenance_class"), + "source_uri": metadata.get("source_uri"), + "allow_duplicate": metadata.get("allow_duplicate", False), } ) diff --git a/src/brainlayer/ingest/__init__.py b/src/brainlayer/ingest/__init__.py index 071be4db..c6846ad7 100644 --- a/src/brainlayer/ingest/__init__.py +++ b/src/brainlayer/ingest/__init__.py @@ -1 +1,27 @@ -"""Ingest adapters for non-Claude AI session transcripts.""" +"""Source-specific ingestion adapters.""" + +from .t3 import ( + DEFAULT_T3_HEALTH_PATH, + DEFAULT_T3_STATE_DB, + T3_PROVENANCE_CLASS, + T3_SOURCE, + T3IngestionResult, + T3Message, + T3Reader, + T3SchemaError, + T3Thread, + ingest_t3, +) + +__all__ = [ + "DEFAULT_T3_HEALTH_PATH", + "DEFAULT_T3_STATE_DB", + "T3_PROVENANCE_CLASS", + "T3_SOURCE", + "T3IngestionResult", + "T3Message", + "T3Reader", + "T3SchemaError", + "T3Thread", + "ingest_t3", +] diff --git a/src/brainlayer/ingest/t3.py b/src/brainlayer/ingest/t3.py new file mode 100644 index 00000000..b5395196 --- /dev/null +++ b/src/brainlayer/ingest/t3.py @@ -0,0 +1,404 @@ +"""Read-only ingestion for T3 Code's SQLite thread projection. + +T3 is a live application database, not a JSONL transcript root. This adapter +opens it with SQLite's read-only URI mode, validates the projection schema +before reading, and emits ordinary BrainLayer chunks with a distinct +``t3-thread`` provenance class. +""" + +from __future__ import annotations + +import json +import logging +import sqlite3 +from dataclasses import dataclass, field +from datetime import datetime, timezone +from pathlib import Path +from typing import Any + +from ..alarm import BrainLayerAlarm, raise_alarm + +logger = logging.getLogger(__name__) + +T3_SOURCE = "t3" +T3_PROVENANCE_CLASS = "t3-thread" +DEFAULT_T3_STATE_DB = Path("~/.t3/userdata/state.sqlite").expanduser() +DEFAULT_T3_HEALTH_PATH = Path("~/.local/share/brainlayer/t3-health.json").expanduser() + +_REQUIRED_COLUMNS: dict[str, frozenset[str]] = { + "projection_threads": frozenset({"thread_id", "project_id", "title", "created_at"}), + "projection_thread_messages": frozenset({"message_id", "thread_id", "role", "text", "created_at"}), + "projection_projects": frozenset({"project_id", "title"}), + "provider_session_runtime": frozenset({"thread_id", "provider_name", "resume_cursor_json"}), +} + + +@dataclass(frozen=True) +class T3Message: + message_id: str + thread_id: str + role: str + text: str + created_at: str + + +@dataclass(frozen=True) +class T3Thread: + thread_id: str + project_id: str + title: str + created_at: str + project_name: str | None = None + messages: tuple[T3Message, ...] = () + provider_name: str | None = None + provider_session_id: str | None = None + mirrored: bool = False + + +@dataclass +class T3IngestionResult: + threads_seen: int = 0 + threads_ingested: int = 0 + messages_seen: int = 0 + messages_ingested: int = 0 + chunks_planned: int = 0 + chunks_indexed: int = 0 + duplicates_accepted: int = 0 + messages_skipped: dict[str, int] = field(default_factory=dict) + + +class T3SchemaError(RuntimeError): + """The live T3 projection no longer satisfies the reader contract.""" + + def __init__(self, missing_tables: list[str], missing_columns: dict[str, list[str]]) -> None: + self.missing_tables = missing_tables + self.missing_columns = missing_columns + super().__init__( + "T3 schema drift: " + + json.dumps( + {"missing_tables": missing_tables, "missing_columns": missing_columns}, + sort_keys=True, + ) + ) + + +class T3Reader: + """Read T3 projection rows without ever opening the source for writing.""" + + def __init__( + self, + state_db_path: Path | str = DEFAULT_T3_STATE_DB, + *, + health_path: Path | str | None = DEFAULT_T3_HEALTH_PATH, + ): + self.state_db_path = Path(state_db_path).expanduser() + self.health_path = Path(health_path).expanduser() if health_path is not None else None + self._last_counts: dict[str, int] = {} + self._failures: list[dict[str, Any]] = [] + + def read_threads(self) -> list[T3Thread]: + """Read a complete short snapshot of the T3 projections. + + Each query runs in SQLite autocommit mode and is fully materialized + before the next query, so the adapter does not hold an application + transaction open while indexing. + """ + conn: sqlite3.Connection | None = None + try: + conn = self._connect() + self._validate_schema(conn) + threads = self._read_snapshot(conn) + except BrainLayerAlarm: + raise + except T3SchemaError as error: + self._fail( + "t3_schema_drift", + "T3 schema no longer matches the ingestion reader contract", + { + "missing_tables": error.missing_tables, + "missing_columns": error.missing_columns, + }, + error, + ) + except sqlite3.OperationalError as error: + if "no such table" in str(error).lower() or "no such column" in str(error).lower(): + self._fail( + "t3_schema_drift", + "T3 schema changed while the reader was taking its snapshot", + {"sqlite_error": str(error)}, + error, + ) + self._fail( + "t3_reader_failed", + "T3 state database could not be read", + {"state_db_path": str(self.state_db_path), "error_type": type(error).__name__}, + error, + ) + except (OSError, sqlite3.Error) as error: + self._fail( + "t3_reader_failed", + "T3 state database could not be read", + {"state_db_path": str(self.state_db_path), "error_type": type(error).__name__}, + error, + ) + else: + self._last_counts = { + "threads_seen": len(threads), + "messages_seen": sum(len(thread.messages) for thread in threads), + "mirrored_threads": sum(thread.mirrored for thread in threads), + } + self._write_health(alerting=False, alert_reasons=[]) + return threads + finally: + if conn is not None: + conn.close() + + raise AssertionError("T3 reader failure path must raise") + + def _connect(self) -> sqlite3.Connection: + if not self.state_db_path.is_file(): + raise FileNotFoundError(f"T3 state database not found: {self.state_db_path}") + uri = f"file:{self.state_db_path}?mode=ro&immutable=0" + conn = sqlite3.connect(uri, uri=True, timeout=1.0, isolation_level=None) + conn.execute("PRAGMA query_only=ON") + conn.execute("PRAGMA busy_timeout=1000") + return conn + + @staticmethod + def _validate_schema(conn: sqlite3.Connection) -> None: + actual_tables = { + row[0] for row in conn.execute("SELECT name FROM sqlite_master WHERE type = 'table'").fetchall() + } + missing_tables = sorted(set(_REQUIRED_COLUMNS) - actual_tables) + missing_columns: dict[str, list[str]] = {} + for table, required in _REQUIRED_COLUMNS.items(): + if table not in actual_tables: + continue + actual_columns = {row[1] for row in conn.execute(f'PRAGMA table_info("{table}")').fetchall()} + missing = sorted(required - actual_columns) + if missing: + missing_columns[table] = missing + if missing_tables or missing_columns: + raise T3SchemaError(missing_tables, missing_columns) + + @staticmethod + def _read_snapshot(conn: sqlite3.Connection) -> list[T3Thread]: + thread_rows = conn.execute( + """ + SELECT t.thread_id, t.project_id, p.title, t.title, t.created_at + FROM projection_threads AS t + LEFT JOIN projection_projects AS p ON p.project_id = t.project_id + ORDER BY t.created_at, t.thread_id + """ + ).fetchall() + message_rows = conn.execute( + """ + SELECT message_id, thread_id, role, text, created_at + FROM projection_thread_messages + ORDER BY thread_id, created_at, message_id + """ + ).fetchall() + runtime_rows = conn.execute( + """ + SELECT thread_id, provider_name, resume_cursor_json + FROM provider_session_runtime + """ + ).fetchall() + + messages_by_thread: dict[str, list[T3Message]] = {} + for message_id, thread_id, role, text, created_at in message_rows: + messages_by_thread.setdefault(thread_id, []).append( + T3Message( + message_id=message_id, + thread_id=thread_id, + role=role, + text=text or "", + created_at=created_at, + ) + ) + + runtimes = {row[0]: row[1:] for row in runtime_rows} + threads = [] + for thread_id, project_id, project_name, title, created_at in thread_rows: + runtime = runtimes.get(thread_id) + runtime_provider = runtime[0] if runtime else None + runtime_session_id = _provider_session_id(runtime[1]) if runtime else None + threads.append( + T3Thread( + thread_id=thread_id, + project_id=project_id, + title=title, + created_at=created_at, + project_name=project_name, + messages=tuple(messages_by_thread.get(thread_id, ())), + provider_name=runtime_provider, + provider_session_id=runtime_session_id, + mirrored=runtime is not None, + ) + ) + return threads + + def _fail(self, code: str, message: str, context: dict[str, Any], error: BaseException) -> None: + failure = { + "code": code, + "message": message, + "error_type": type(error).__name__, + "error": str(error), + **context, + } + self._failures.append(failure) + self._write_health( + alerting=True, alert_reasons=["schema_drift" if code == "t3_schema_drift" else "reader_failure"] + ) + try: + raise_alarm(code, message, {"state_db_path": str(self.state_db_path), **context}) + except BrainLayerAlarm: + raise + raise error + + def _write_health(self, *, alerting: bool, alert_reasons: list[str]) -> None: + if self.health_path is None: + return + payload = { + "updated_at": datetime.now(timezone.utc).isoformat(), + "source": T3_SOURCE, + "state_db_path": str(self.state_db_path), + "alerting": alerting, + "alert_reasons": alert_reasons, + "failures": list(self._failures), + **self._last_counts, + } + try: + self.health_path.parent.mkdir(parents=True, exist_ok=True) + temp_path = self.health_path.with_suffix(".tmp") + temp_path.write_text(json.dumps(payload, sort_keys=True), encoding="utf-8") + temp_path.replace(self.health_path) + except OSError: + logger.exception("Failed to write T3 health snapshot to %s", self.health_path) + + +def _provider_session_id(resume_cursor_json: str | None) -> str | None: + if not resume_cursor_json: + return None + try: + value = json.loads(resume_cursor_json) + except (TypeError, json.JSONDecodeError): + return None + if not isinstance(value, dict): + return None + provider_session_id = value.get("threadId") + return str(provider_session_id) if provider_session_id else None + + +def _index_chunks(chunks, *, source_file: str, project: str | None, db_path: Path) -> int: + from ..index_new import index_chunks_to_sqlite + + return index_chunks_to_sqlite(chunks, source_file=source_file, project=project, db_path=db_path) + + +def _message_chunks(thread: T3Thread, message: T3Message) -> list: + from ..pipeline.chunk import Chunk, chunk_content + from ..pipeline.classify import ClassifiedContent, ContentType, ContentValue + + content_type = ContentType.USER_MESSAGE if message.role == "user" else ContentType.ASSISTANT_TEXT + value = ContentValue.HIGH if message.role == "user" else ContentValue.MEDIUM + base_metadata = { + "chunk_id": f"t3:{thread.thread_id}:{message.message_id}:0", + "conversation_id": thread.thread_id, + "created_at": message.created_at, + "allow_duplicate": True, + "message_id": message.message_id, + "project": thread.project_name, + "provenance_class": T3_PROVENANCE_CLASS, + "sender": message.role, + "session_id": thread.thread_id, + "source": T3_SOURCE, + "source_uri": f"t3://thread/{thread.thread_id}/message/{message.message_id}", + "t3_thread_id": thread.thread_id, + "t3_title": thread.title, + "t3_mirrored": thread.mirrored, + "t3_provider_name": thread.provider_name, + "t3_provider_session_id": thread.provider_session_id, + } + classified = ClassifiedContent( + content=message.text, + content_type=content_type, + value=value, + metadata=base_metadata, + ) + chunks = chunk_content(classified) + if not chunks: + if not message.text.strip(): + return [] + chunks = [ + Chunk( + content=message.text, + content_type=content_type, + value=value, + metadata=base_metadata, + char_count=len(message.text), + ) + ] + for index, chunk in enumerate(chunks): + chunk.metadata = { + **base_metadata, + **chunk.metadata, + "chunk_id": f"t3:{thread.thread_id}:{message.message_id}:{index}", + } + return chunks + + +def ingest_t3( + state_db_path: Path | str = DEFAULT_T3_STATE_DB, + *, + db_path: Path | str | None = None, + health_path: Path | str | None = DEFAULT_T3_HEALTH_PATH, + dry_run: bool = False, +) -> T3IngestionResult: + """Ingest all T3 threads, deliberately retaining mirrored content.""" + destination = Path(db_path).expanduser() if db_path is not None else None + reader = T3Reader(state_db_path, health_path=health_path) + threads = reader.read_threads() + result = T3IngestionResult( + threads_seen=len(threads), + threads_ingested=len(threads), + messages_seen=sum(len(thread.messages) for thread in threads), + duplicates_accepted=sum(thread.mirrored for thread in threads), + ) + chunks = [] + for thread in threads: + for message in thread.messages: + if message.role not in {"user", "assistant"}: + result.messages_skipped["unsupported_role"] = result.messages_skipped.get("unsupported_role", 0) + 1 + continue + message_chunks = _message_chunks(thread, message) + if not message_chunks: + result.messages_skipped["empty_text"] = result.messages_skipped.get("empty_text", 0) + 1 + continue + result.messages_ingested += 1 + chunks.extend(message_chunks) + result.chunks_planned = len(chunks) + if chunks and not dry_run: + if destination is None: + from ..paths import DEFAULT_DB_PATH + + destination = DEFAULT_DB_PATH + result.chunks_indexed = _index_chunks( + chunks, + source_file=str(Path(state_db_path).expanduser()), + project=None, + db_path=destination, + ) + if reader.health_path is not None: + reader._last_counts.update( + { + "threads_ingested": result.threads_ingested, + "messages_ingested": result.messages_ingested, + "chunks_planned": result.chunks_planned, + "chunks_indexed": result.chunks_indexed, + "duplicates_accepted": result.duplicates_accepted, + "messages_skipped": sum(result.messages_skipped.values()), + } + ) + reader._write_health(alerting=False, alert_reasons=[]) + return result diff --git a/src/brainlayer/vector_store.py b/src/brainlayer/vector_store.py index 1421debc..7e1848fb 100644 --- a/src/brainlayer/vector_store.py +++ b/src/brainlayer/vector_store.py @@ -2517,14 +2517,21 @@ def stop_expired_statement() -> bool: chunk = {**chunk, "created_at": created_at} tags_value = chunk.get("tags") tags_json = json.dumps(tags_value) if isinstance(tags_value, (list, dict)) else tags_value - duplicate, dedupe_fields = find_duplicate( - self.conn, - chunk_id=chunk_id, - content=chunk["content"], - created_at=created_at, - project=chunk.get("project"), - content_type=chunk.get("content_type"), - ) + # T3 is intentionally a first-class mirror source. Its + # explicit opt-out preserves the accepted duplication + # instead of inventing a source-specific dedup heuristic. + if chunk.get("allow_duplicate"): + duplicate = None + dedupe_fields = compute_dedupe_fields(chunk["content"], created_at) + else: + duplicate, dedupe_fields = find_duplicate( + self.conn, + chunk_id=chunk_id, + content=chunk["content"], + created_at=created_at, + project=chunk.get("project"), + content_type=chunk.get("content_type"), + ) if duplicate is not None: duplicate_row_exists = cursor.execute( "SELECT 1 FROM chunks WHERE id = ?", (chunk_id,) @@ -2565,16 +2572,24 @@ def stop_expired_statement() -> bool: self._upsert_chunk_vector(cursor, chunk_id, embedding) continue + has_provenance_class = getattr(self, "_has_provenance_class", False) + provenance_column = ", provenance_class" if has_provenance_class else "" + provenance_value = ", ?" if has_provenance_class else "" + provenance_update = ( + ", provenance_class = COALESCE(excluded.provenance_class, chunks.provenance_class)" + if has_provenance_class + else "" + ) cursor.execute( - """ + f""" INSERT INTO chunks (id, content, metadata, source_file, project, content_type, value_type, char_count, source, created_at, conversation_id, position, sender, chunk_origin, tags, importance, half_life_days, seen_count, last_seen_at, dedupe_hash, simhash, simhash_band_0, simhash_band_1, simhash_band_2, simhash_band_3, - brick_id, source_uri, status, ingested_at, topic_cluster) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + brick_id, source_uri, status, ingested_at, topic_cluster{provenance_column}) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?{provenance_value}) ON CONFLICT(id) DO UPDATE SET content = excluded.content, metadata = excluded.metadata, @@ -2616,7 +2631,7 @@ def stop_expired_statement() -> bool: WHEN chunks.chunk_origin IS NULL THEN COALESCE(excluded.chunk_origin, 'unknown') ELSE chunks.chunk_origin - END + END{provenance_update} """, ( chunk_id, @@ -2649,6 +2664,7 @@ def stop_expired_statement() -> bool: chunk.get("status", "active"), chunk.get("ingested_at") or int(time.time()), chunk.get("topic_cluster"), + *([chunk.get("provenance_class")] if has_provenance_class else []), ), ) self._upsert_chunk_vector(cursor, chunk_id, embedding) diff --git a/tests/test_cli_launchd_mode_a.py b/tests/test_cli_launchd_mode_a.py index 2f820d59..958a7bed 100644 --- a/tests/test_cli_launchd_mode_a.py +++ b/tests/test_cli_launchd_mode_a.py @@ -50,6 +50,8 @@ def fake_run_health_check(config): str(tmp_path / "offsets.json"), "--watcher-health-path", str(tmp_path / "watcher-health.json"), + "--t3-health-path", + str(tmp_path / "t3-health.json"), "--json", ], ) @@ -68,6 +70,7 @@ def fake_run_health_check(config): assert config.queue_dir == tmp_path / "queue" assert config.offsets_path == tmp_path / "offsets.json" assert config.watcher_health_path == tmp_path / "watcher-health.json" + assert config.t3_health_path == tmp_path / "t3-health.json" def test_pause_and_resume_record_labels_and_call_launchctl(monkeypatch, tmp_path): diff --git a/tests/test_ingest_t3.py b/tests/test_ingest_t3.py new file mode 100644 index 00000000..5890dec6 --- /dev/null +++ b/tests/test_ingest_t3.py @@ -0,0 +1,346 @@ +"""Behavioral tests for the read-only T3 SQLite ingestion adapter.""" + +import json +import sqlite3 +from pathlib import Path +from uuid import UUID + +import pytest +from typer.testing import CliRunner + +from brainlayer.alarm import BrainLayerAlarm +from brainlayer.embeddings import EmbeddedChunk +from brainlayer.pipeline.chunk import Chunk +from brainlayer.pipeline.classify import ContentType, ContentValue + + +def _is_uuid(value: str | None) -> bool: + if value is None: + return False + try: + UUID(value) + except (AttributeError, TypeError, ValueError): + return False + return True + + +def _create_t3_fixture(path: Path, *, drift: bool = False) -> Path: + conn = sqlite3.connect(path) + try: + conn.executescript( + """ + CREATE TABLE projection_threads ( + thread_id TEXT PRIMARY KEY, + project_id TEXT NOT NULL, + title TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE TABLE projection_thread_messages ( + message_id TEXT PRIMARY KEY, + thread_id TEXT NOT NULL, + role TEXT NOT NULL, + text TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ); + CREATE TABLE projection_thread_sessions ( + thread_id TEXT PRIMARY KEY, + status TEXT NOT NULL, + provider_name TEXT, + provider_session_id TEXT, + provider_thread_id TEXT, + updated_at TEXT NOT NULL + ); + CREATE TABLE provider_session_runtime ( + thread_id TEXT PRIMARY KEY, + provider_name TEXT NOT NULL, + adapter_key TEXT NOT NULL, + status TEXT NOT NULL, + last_seen_at TEXT NOT NULL, + resume_cursor_json TEXT, + runtime_payload_json TEXT + ); + CREATE TABLE projection_projects ( + project_id TEXT PRIMARY KEY, + title TEXT NOT NULL + ); + """ + ) + if drift: + conn.execute("ALTER TABLE projection_thread_messages RENAME COLUMN text TO body") + + conn.executemany( + "INSERT INTO projection_threads VALUES (?, ?, ?, ?, ?)", + [ + ("thread-1", "brainlayer", "Mirrored thread", "2026-07-01T00:00:00Z", "2026-07-01T00:02:00Z"), + ("thread-2", "golems", "Unmirrored thread", "2026-07-02T00:00:00Z", "2026-07-02T00:01:00Z"), + ], + ) + conn.executemany( + "INSERT INTO projection_projects VALUES (?, ?)", + [("brainlayer", "BrainLayer"), ("golems", "Golems")], + ) + if not drift: + conn.executemany( + "INSERT INTO projection_thread_messages VALUES (?, ?, ?, ?, ?, ?)", + [ + ("message-1", "thread-1", "user", "u", "2026-07-01T00:00:01Z", "2026-07-01T00:00:01Z"), + ( + "message-2", + "thread-1", + "assistant", + "assistant reply", + "2026-07-01T00:00:02Z", + "2026-07-01T00:00:02Z", + ), + ( + "message-3", + "thread-2", + "user", + "a useful unmirrored prompt", + "2026-07-02T00:00:01Z", + "2026-07-02T00:00:01Z", + ), + ], + ) + conn.execute( + "INSERT INTO projection_thread_sessions VALUES (?, ?, ?, ?, ?, ?)", + ("thread-1", "stopped", "codex", None, None, "2026-07-01T00:02:00Z"), + ) + conn.execute( + "INSERT INTO provider_session_runtime VALUES (?, ?, ?, ?, ?, ?, ?)", + ( + "thread-1", + "codex", + "codex", + "stopped", + "2026-07-01T00:02:00Z", + json.dumps({"threadId": "provider-session-1"}), + json.dumps({"cwd": "/Users/test/Gits/brainlayer"}), + ), + ) + conn.commit() + finally: + conn.close() + return path + + +def test_t3_reader_maps_messages_and_thread_provider_linkage(tmp_path): + from brainlayer.ingest.t3 import T3Reader + + state_db = _create_t3_fixture(tmp_path / "state.sqlite") + + threads = T3Reader(state_db, health_path=tmp_path / "t3-health.json").read_threads() + + assert [thread.thread_id for thread in threads] == ["thread-1", "thread-2"] + assert [message.message_id for message in threads[0].messages] == ["message-1", "message-2"] + assert threads[0].provider_session_id == "provider-session-1" + assert threads[0].project_name == "BrainLayer" + assert threads[0].mirrored is True + assert threads[1].provider_session_id is None + assert threads[1].project_name == "Golems" + assert threads[1].mirrored is False + + +def test_t3_project_mapping_missing_row_falls_back_to_none(tmp_path): + from brainlayer.ingest.t3 import T3Reader + + state_db = _create_t3_fixture(tmp_path / "state.sqlite") + with sqlite3.connect(state_db) as conn: + conn.execute("DELETE FROM projection_projects WHERE project_id = ?", ("golems",)) + + threads = T3Reader(state_db, health_path=tmp_path / "t3-health.json").read_threads() + + assert threads[0].project_name == "BrainLayer" + assert threads[1].project_name is None + + +def test_t3_reader_does_not_require_unused_session_projection(tmp_path): + from brainlayer.ingest.t3 import T3Reader + + state_db = _create_t3_fixture(tmp_path / "state.sqlite") + with sqlite3.connect(state_db) as conn: + conn.execute("DROP TABLE projection_thread_sessions") + + threads = T3Reader(state_db, health_path=tmp_path / "t3-health.json").read_threads() + + assert len(threads) == 2 + assert threads[0].provider_session_id == "provider-session-1" + + +def test_t3_reader_opens_source_with_readonly_wal_safe_uri(tmp_path, monkeypatch): + from brainlayer.ingest.t3 import T3Reader + + state_db = _create_t3_fixture(tmp_path / "state.sqlite") + connect_calls = [] + real_connect = sqlite3.connect + + def capture_connect(database, *args, **kwargs): + connect_calls.append((database, kwargs)) + return real_connect(database, *args, **kwargs) + + monkeypatch.setattr("brainlayer.ingest.t3.sqlite3.connect", capture_connect) + + T3Reader(state_db, health_path=tmp_path / "t3-health.json").read_threads() + + assert connect_calls[0][0] == f"file:{state_db}?mode=ro&immutable=0" + assert connect_calls[0][1]["uri"] is True + assert connect_calls[0][1]["isolation_level"] is None + + +def test_t3_schema_drift_raises_alarm_and_writes_health(tmp_path, monkeypatch): + from brainlayer.ingest.t3 import T3Reader + + state_db = _create_t3_fixture(tmp_path / "state.sqlite", drift=True) + health_path = tmp_path / "t3-health.json" + alarms = [] + + def capture_alarm(code, message, context): + alarms.append((code, message, context)) + raise BrainLayerAlarm(code, message, context) + + monkeypatch.setattr("brainlayer.ingest.t3.raise_alarm", capture_alarm) + + with pytest.raises(BrainLayerAlarm) as raised: + T3Reader(state_db, health_path=health_path).read_threads() + + assert raised.value.code == "t3_schema_drift" + assert alarms[0][2]["missing_columns"]["projection_thread_messages"] == ["text"] + health = json.loads(health_path.read_text()) + assert health["alerting"] is True + assert "schema_drift" in health["alert_reasons"] + assert health["failures"][0]["code"] == "t3_schema_drift" + + +def test_t3_ingestion_keeps_short_messages_and_sets_first_class_provenance(tmp_path, monkeypatch): + from brainlayer.ingest.t3 import ingest_t3 + + state_db = _create_t3_fixture(tmp_path / "state.sqlite") + indexed: list[Chunk] = [] + + def capture_index(chunks, *, source_file, project, db_path): + indexed.extend(chunks) + assert source_file == str(state_db) + assert project is None + assert db_path == tmp_path / "brainlayer.db" + return len(chunks) + + monkeypatch.setattr("brainlayer.ingest.t3._index_chunks", capture_index) + + result = ingest_t3( + state_db, + db_path=tmp_path / "brainlayer.db", + health_path=tmp_path / "t3-health.json", + ) + + assert result.threads_seen == 2 + assert result.threads_ingested == 2 + assert result.messages_seen == 3 + assert result.messages_ingested == 3 + assert result.messages_skipped == {} + assert result.duplicates_accepted == 1 + assert len(indexed) == 3 + assert {chunk.metadata["provenance_class"] for chunk in indexed} == {"t3-thread"} + assert {chunk.metadata["source"] for chunk in indexed} == {"t3"} + assert {chunk.metadata["project"] for chunk in indexed} == {"BrainLayer", "Golems"} + assert all(not _is_uuid(chunk.metadata["project"]) for chunk in indexed) + assert indexed[0].metadata["t3_provider_name"] == "codex" + assert indexed[0].metadata["t3_provider_session_id"] == "provider-session-1" + assert indexed[0].metadata["t3_mirrored"] is True + assert {chunk.metadata["conversation_id"] for chunk in indexed} == {"thread-1", "thread-2"} + assert {chunk.metadata["chunk_id"] for chunk in indexed} == { + "t3:thread-1:message-1:0", + "t3:thread-1:message-2:0", + "t3:thread-2:message-3:0", + } + + +def test_indexer_preserves_stable_identity_timestamp_and_provenance(monkeypatch): + from brainlayer import index_new + + chunk = Chunk( + content="T3 message", + content_type=ContentType.USER_MESSAGE, + value=ContentValue.HIGH, + metadata={ + "chunk_id": "t3:thread-1:message-1:0", + "created_at": "2026-07-01T00:00:01Z", + "provenance_class": "t3-thread", + "source": "t3", + "session_id": "thread-1", + "sender": "user", + }, + char_count=11, + ) + captured = {} + + class FakeStore: + def upsert_chunks(self, chunks, embeddings, *, deadline_monotonic=None): + captured["chunks"] = chunks + captured["embeddings"] = embeddings + return len(chunks) + + monkeypatch.setattr(index_new, "embed_chunks", lambda chunks, on_progress=None: [EmbeddedChunk(chunk, [0.1])]) + + assert index_new.index_chunks_to_sqlite([chunk], source_file="/missing/state.sqlite", store=FakeStore()) == 1 + assert captured["chunks"][0]["id"] == "t3:thread-1:message-1:0" + assert captured["chunks"][0]["created_at"] == "2026-07-01T00:00:01Z" + assert captured["chunks"][0]["provenance_class"] == "t3-thread" + + +def test_ingest_t3_cli_is_a_real_production_entrypoint(tmp_path, monkeypatch): + from brainlayer.cli import app + from brainlayer.ingest.t3 import T3IngestionResult + + captured = {} + + def fake_ingest(state_db_path, *, db_path, health_path, dry_run): + captured.update( + state_db_path=state_db_path, + db_path=db_path, + health_path=health_path, + dry_run=dry_run, + ) + return T3IngestionResult( + threads_seen=45, + threads_ingested=45, + messages_seen=2349, + messages_ingested=2349, + chunks_planned=2506, + chunks_indexed=2506, + duplicates_accepted=34, + ) + + monkeypatch.setattr("brainlayer.ingest.t3.ingest_t3", fake_ingest) + state_db = tmp_path / "state.sqlite" + db_path = tmp_path / "brainlayer.db" + health_path = tmp_path / "t3-health.json" + + result = CliRunner().invoke( + app, + [ + "ingest-t3", + "--state-db", + str(state_db), + "--db", + str(db_path), + "--health-path", + str(health_path), + ], + ) + + assert result.exit_code == 0, result.output + assert captured == { + "state_db_path": state_db, + "db_path": db_path, + "health_path": health_path, + "dry_run": False, + } + assert "chunks_indexed=2506" in result.output + + +def test_read_t3_threads_export_is_removed(): + import brainlayer.ingest as ingest + + assert not hasattr(ingest, "read_t3_threads") diff --git a/tests/test_launchd_hygiene.py b/tests/test_launchd_hygiene.py index 01c7dbfd..ba4b2950 100644 --- a/tests/test_launchd_hygiene.py +++ b/tests/test_launchd_hygiene.py @@ -230,6 +230,16 @@ def test_index_launchagent_runs_nightly_without_keepalive_or_run_at_load(): assert index["StartCalendarInterval"] == {"Hour": 3, "Minute": 15} +def test_t3_ingest_launchagent_invokes_first_class_source(): + t3 = _load("scripts/launchd/com.brainlayer.t3-ingest.plist") + + assert t3["Label"] == "com.brainlayer.t3-ingest" + assert t3["ProgramArguments"][:3] == ["__BRAINLAYER_ENV_RUN__", "__BRAINLAYER_BIN__", "ingest-t3"] + assert t3["StartCalendarInterval"] == {"Hour": 3, "Minute": 45} + assert t3["RunAtLoad"] is True + assert "KeepAlive" not in t3 + + def test_canonical_launchagent_env_has_no_concrete_dev_src_paths(): plist_paths = [ *sorted((REPO_ROOT / "scripts/launchd").glob("com.brainlayer.*.plist")), @@ -527,6 +537,15 @@ def test_launchd_installer_wires_health_check_target(): assert "remove_plist health-check" in install_source +def test_launchd_installer_wires_t3_ingest_target(): + install_source = (REPO_ROOT / "scripts/launchd/install.sh").read_text(encoding="utf-8") + + assert "./scripts/launchd/install.sh t3-ingest" in install_source + assert "t3-ingest)" in install_source + assert "install_plist t3-ingest" in install_source + assert "remove_plist t3-ingest" in install_source + + def test_launchd_installer_wires_throughput_watchdog_target(): install_source = (REPO_ROOT / "scripts/launchd/install.sh").read_text(encoding="utf-8") diff --git a/tests/test_stability_health_check.py b/tests/test_stability_health_check.py index 775e039c..07c52f31 100644 --- a/tests/test_stability_health_check.py +++ b/tests/test_stability_health_check.py @@ -131,6 +131,39 @@ def _ok_canary(_socket_path: Path, _query: str, _timeout_seconds: float) -> dict } +def test_health_check_consumes_alerting_t3_health_snapshot(tmp_path): + health_path = tmp_path / "t3-health.json" + payload = { + "alerting": True, + "alert_reasons": ["schema_drift"], + "failures": [{"code": "t3_schema_drift", "error": "missing text"}], + } + health_path.write_text(json.dumps(payload), encoding="utf-8") + db_path = tmp_path / "brainlayer.db" + _make_db(db_path, total=1, vector_rows=1) + + result = health_check.run_health_check( + health_check.HealthCheckConfig( + db_path=db_path, + state_path=tmp_path / "state.json", + t3_health_path=health_path, + watcher_health_path=tmp_path / "watcher-health.json", + drain_health_path=tmp_path / "drain-health.json", + source_jsonl_globs=[], + queue_dir=tmp_path / "queue", + pending_stores_path=tmp_path / "pending-stores.jsonl", + ), + ps_output_fn=lambda: "", + socket_request_fn=_ok_canary, + command_runner=lambda _args: SimpleNamespace(returncode=0, stdout="", stderr=""), + ) + + assert result.t3_health == payload + issue = next(issue for issue in result.issues if issue.code == "t3_ingest_unhealthy") + assert issue.severity == "critical" + assert "schema_drift" in issue.message + + def test_backlog_batch_zero_alarms_but_waits_until_repeated_failure_to_kickstart_hotlane(tmp_path, capsys): db_path = tmp_path / "brainlayer.db" state_path = tmp_path / "health-state.json" diff --git a/tests/test_vector_store_upsert_transactions.py b/tests/test_vector_store_upsert_transactions.py index ef730a7c..a99287b8 100644 --- a/tests/test_vector_store_upsert_transactions.py +++ b/tests/test_vector_store_upsert_transactions.py @@ -172,6 +172,38 @@ def test_upsert_chunks_preserves_dedupe_and_repeat_upsert_shape(isolated_store, assert vector_rows == [("canonical",), ("distinct",)] +def test_t3_upsert_persists_provenance_and_accepts_mirrored_duplicates(isolated_store): + duplicate_content = "T3 mirrored content must remain a first-class duplicate" + chunks = [ + { + **_chunk("t3-thread-1", duplicate_content), + "project": None, + "source": "t3", + "provenance_class": "t3-thread", + "allow_duplicate": True, + }, + { + **_chunk("t3-thread-2", duplicate_content), + "project": None, + "source": "t3", + "provenance_class": "t3-thread", + "allow_duplicate": True, + }, + ] + + assert isolated_store.upsert_chunks(chunks, [_embedding(1), _embedding(2)]) == 2 + + rows = ( + isolated_store.conn.cursor() + .execute("SELECT id, source, provenance_class FROM chunks WHERE id LIKE 't3-thread-%' ORDER BY id") + .fetchall() + ) + assert rows == [ + ("t3-thread-1", "t3", "t3-thread"), + ("t3-thread-2", "t3", "t3-thread"), + ] + + def test_busy_sub_batch_retries_without_replaying_committed_sub_batches(isolated_store, monkeypatch): monkeypatch.setenv("BRAINLAYER_INDEX_TXN_BATCH", "2") insert_attempts: dict[str, int] = {}