From d4abd2ac6a4152092c8153fd1c60dba6c6f4bf5e Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Sun, 13 Sep 2026 23:55:33 +0300 Subject: [PATCH 1/5] test(observability): pin store emitter red contract Co-Authored-By: brainlayerCodex-8726de2b running unknown --- tests/test_observability_surface.py | 105 ++++++++++++++++++++++++++++ 1 file changed, 105 insertions(+) create mode 100644 tests/test_observability_surface.py diff --git a/tests/test_observability_surface.py b/tests/test_observability_surface.py new file mode 100644 index 00000000..277becaf --- /dev/null +++ b/tests/test_observability_surface.py @@ -0,0 +1,105 @@ +from __future__ import annotations + +import json +import os +import subprocess +import sys +from pathlib import Path + +import pytest + + +REPO = Path(__file__).resolve().parents[1] +FIXTURES = REPO / "tests/fixtures/observability" +OWNED_SECTIONS = ("stores", "emitters", "author_unknown") + + +def _dev_cases() -> list[dict[str, object]]: + manifest = json.loads((FIXTURES / "cases.json").read_text(encoding="utf-8")) + return [case for case in manifest["cases"] if case["split"] == "dev"] + + +def _run_case(case: dict[str, object], tmp_path: Path) -> tuple[dict[str, object], list[str]]: + inputs = case["inputs"] + assert isinstance(inputs, dict) + output = tmp_path / "observability.json" + trace = tmp_path / "trace.json" + env = { + key: os.environ[key] + for key in ("HOME", "PATH", "BRAINLAYER_FORBID_EMBEDDING_MODEL") + if key in os.environ + } + env.update( + { + "PYTHONPATH": str(REPO / "src"), + "BRAINLAYER_DB": str(FIXTURES / str(inputs["db"])), + "BRAINLAYER_OBSERVABILITY_PATH": str(output), + "BRAINLAYER_OBSERVABILITY_TRACE_PATH": str(trace), + "BRAINLAYER_OBSERVABILITY_INPUT_ROOT": str(FIXTURES), + "BRAINLAYER_OBSERVABILITY_JSONL_BACKUP_LOG": str(FIXTURES / str(inputs["jsonl_backup_log"])), + "BRAINLAYER_OBSERVABILITY_BACKUP_DAILY_LOG": str(FIXTURES / str(inputs["backup_daily_log"])), + "BRAINLAYER_OBSERVABILITY_LAUNCHD_OUTPUT": str(FIXTURES / str(inputs["launchd_output"])), + "BRAINLAYER_OBSERVABILITY_DISABLED_DIR": str(FIXTURES / str(inputs["disabled_dir"])), + "BRAINLAYER_OBSERVABILITY_NOW": str(case["generated_at"]), + } + ) + run = subprocess.run( + [sys.executable, "-m", "brainlayer.observability_surface"], + cwd=REPO, + env=env, + capture_output=True, + text=True, + timeout=30, + ) + assert run.returncode == 0, run.stderr + return json.loads(output.read_text(encoding="utf-8")), json.loads(trace.read_text(encoding="utf-8")) + + +@pytest.mark.parametrize("case", _dev_cases(), ids=lambda case: str(case["case_id"])) +def test_dev_goldens_for_owned_sections(case: dict[str, object], tmp_path: Path) -> None: + actual, trace = _run_case(case, tmp_path) + golden_path = FIXTURES / "golden" / f"{case['case_id']}.json" + expected = json.loads(golden_path.read_text(encoding="utf-8")) + + for section in OWNED_SECTIONS: + assert actual[section] == expected[section] + assert actual["db_path"] == expected["db_path"] + assert trace.count(str(case["inputs"]["db"])) == 1 + + +def test_preview_is_secret_scrubbed_and_limited_to_80_chars(tmp_path: Path) -> None: + import shutil + import sqlite3 + + case = next(case for case in _dev_cases() if case["case_id"] == "healthy-dev") + db = tmp_path / "preview.sqlite" + shutil.copy2(FIXTURES / str(case["inputs"]["db"]), db) + secret = "sk-ant-" + "A" * 30 + connection = sqlite3.connect(db) + connection.execute("UPDATE chunks SET content = ? WHERE id = 'synthetic-00'", (secret + "x" * 100,)) + connection.commit() + connection.close() + case = {**case, "inputs": {**case["inputs"], "db": str(db)}} + + actual, _ = _run_case(case, tmp_path) + preview = actual["stores"]["latest"][0]["preview"] + assert secret not in preview + assert preview.startswith("[REDACTED:anthropic]") + assert len(preview) <= 80 + + +@pytest.mark.parametrize( + ("source_file", "expected"), + [ + ("/Users/x/.claude/projects/-Users-x-Gits-brainlayer/session.jsonl", "brainlayer"), + ("/Users/x/.codex/sessions/2026/09/13/rollout.jsonl", "codex"), + ("brainbar-store", "brainbar-store"), + ("realtime-hook", "realtime-hook"), + ("unknown", "unknown"), + ("", "unknown"), + ], +) +def test_source_file_emitter_derivation(source_file: str, expected: str) -> None: + from brainlayer.observability_surface import derive_emitter + + assert derive_emitter(None, None, source_file) == (expected, "source_file") From 6ecaca04e1b04c5aec7df00ed582ab23431a9426 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Mon, 14 Sep 2026 00:17:38 +0300 Subject: [PATCH 2/5] feat(observability): produce store emitter metrics Co-Authored-By: brainlayerCodex-8726de2b running unknown --- scripts/observability_eval.py | 1 + src/brainlayer/cli/__init__.py | 13 ++ src/brainlayer/observability_surface.py | 269 ++++++++++++++++++++++++ tests/test_observability_surface.py | 56 ++--- 4 files changed, 314 insertions(+), 25 deletions(-) create mode 100644 src/brainlayer/observability_surface.py diff --git a/scripts/observability_eval.py b/scripts/observability_eval.py index e90346ef..43a2ce79 100644 --- a/scripts/observability_eval.py +++ b/scripts/observability_eval.py @@ -156,6 +156,7 @@ def _run_case( "BRAINLAYER_OBSERVABILITY_LAUNCHD_OUTPUT": str(input_root / case["inputs"]["launchd_output"]), "BRAINLAYER_OBSERVABILITY_DISABLED_DIR": str(input_root / case["inputs"]["disabled_dir"]), "BRAINLAYER_OBSERVABILITY_NOW": case["generated_at"], "PYTHONPATH": str(producer_root / "src"), + "BRAINLAYER_OBSERVABILITY_PRODUCER_ROOT": str(producer_root), }) try: run = subprocess.run([sys.executable, "-m", "brainlayer.observability_surface"], cwd=producer_root, diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py index f55b29d8..796c9d45 100644 --- a/src/brainlayer/cli/__init__.py +++ b/src/brainlayer/cli/__init__.py @@ -76,6 +76,19 @@ def _index_max_runtime_s() -> float: return value +@app.command("observability") +def observability_command( + write: bool = typer.Option(False, "--write", help="Write observability.json beside the resolved database."), + stdout: bool = typer.Option(False, "--stdout", help="Print the observability document instead of writing it."), +) -> None: + """Produce the versioned BrainLayer observability document.""" + if write and stdout: + raise typer.BadParameter("choose either --write or --stdout") + from ..observability_surface import write_document + + write_document(stdout=stdout) + + @app.command("writer-telemetry") def writer_telemetry_command( action: Annotated[str, typer.Argument(help="Read mode: tail or summary.")] = "summary", diff --git a/src/brainlayer/observability_surface.py b/src/brainlayer/observability_surface.py new file mode 100644 index 00000000..893f6d1a --- /dev/null +++ b/src/brainlayer/observability_surface.py @@ -0,0 +1,269 @@ +"""Produce the versioned BrainLayer observability document.""" + +from __future__ import annotations + +import hashlib +import json +import os +import sqlite3 +from collections import Counter +from collections.abc import Mapping +from datetime import UTC, datetime, timedelta +from pathlib import Path +from typing import Any + +from .paths import get_db_path +from .pipeline.secret_scrub import scrub_secrets + +# fmt: off +WINDOW_HOURS = 24 +DERIVATION_NOTE = "metadata.attributionAgent is absent in the 2026-09-13 census; emitter derives from source, then sender, then source_file, or the section is unmeasurable" +CENSUS_NOTE = "never_classified = 476,679 of 731,153 live (65.2%); both-NULL = 65,017; classified_unknown literal not observed on the copy" +def _iso_utc(value: datetime) -> str: + return value.astimezone(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") + + +def _parse_time(value: str) -> datetime: + parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) + return parsed.replace(tzinfo=UTC) if parsed.tzinfo is None else parsed.astimezone(UTC) +class InputRecorder: + def __init__(self, *, root: Path | None, trace_path: Path | None, now: datetime) -> None: + self.root = root.resolve() if root else None + self.trace_path = trace_path + self.now = now + self.trace: list[str] = [] + + def display_path(self, path: Path) -> str: + resolved = path.expanduser().resolve() + if self.root is not None: + try: + return str(resolved.relative_to(self.root)) + except ValueError: + pass + return str(resolved) + + def __call__(self, path: Path | str, *, status: str = "read", rows_or_bytes: int | None = None, + skipped_lines: int = 0, in_section_inputs: bool = True) -> dict[str, Any]: + del in_section_inputs # The caller decides whether to retain the returned object. + resolved = Path(path).expanduser().resolve() + displayed = self.display_path(resolved) + self.trace.append(displayed) + try: + stat = resolved.stat() + except FileNotFoundError: + return _input(displayed, "missing", None, None, None, skipped_lines) + effective_status = status + if resolved.is_file() and stat.st_size == 0: + effective_status = "empty" + elif status == "read" and datetime.fromtimestamp(stat.st_mtime, UTC) > self.now: + effective_status = "future" + digest = None + if resolved.is_file() and resolved.suffix not in {".sqlite", ".db"}: + with resolved.open("rb") as handle: + digest = hashlib.sha256(handle.read(65_536)).hexdigest() + return _input(displayed, effective_status, _iso_utc(datetime.fromtimestamp(stat.st_mtime, UTC)), + stat.st_size if rows_or_bytes is None else rows_or_bytes, digest, skipped_lines) + + def write_trace(self) -> None: + if self.trace_path is not None: + self.trace_path.parent.mkdir(parents=True, exist_ok=True) + self.trace_path.write_text(json.dumps(self.trace, indent=2) + "\n", encoding="utf-8") + + +def _clean(value: object) -> str | None: + text = str(value).strip() if value is not None else "" + return text or None +def _input(path: str, status: str, mtime: str | None, size: int | None, + digest: str | None, skipped: int) -> dict[str, Any]: + return {"path": path, "status": status, "mtime": mtime, "rows_or_bytes": size, + "sha256_first_64kb": digest, "skipped_lines": skipped} + + +def derive_emitter(source: object, sender: object, source_file: object) -> tuple[str, str]: + """Derive the emitter in source, sender, source_file precedence order.""" + for value, origin in ((source, "source"), (sender, "sender")): + if cleaned := _clean(value): + return cleaned, origin + path = _clean(source_file) or "unknown" + if "/.codex/sessions/" in path: + return "codex", "source_file" + if "/.claude/projects/" in path: + project_dir = path.split("/.claude/projects/", 1)[1].split("/", 1)[0] + if "-Gits-" in project_dir: + return project_dir.rsplit("-Gits-", 1)[1] or "unknown", "source_file" + return project_dir or "unknown", "source_file" + return path, "source_file" + + +def _unmeasurable(reason: str, db_input: dict[str, Any]) -> dict[str, Any]: + return {"state": "unmeasurable", "reason": reason, "inputs": [db_input]} +def _measured(db_input: dict[str, Any], **fields: Any) -> dict[str, Any]: + return {"state": "measured", "reason": "", "inputs": [db_input], **fields} + + +def _missing(columns: set[str], required: tuple[str, ...]) -> str | None: + return next((name for name in required if name not in columns), None) +def _stores(connection: sqlite3.Connection, columns: set[str], db_input: dict[str, Any], now: datetime) -> dict[str, Any]: + required = ("source_class", "id", "content", "source", "sender", "source_file", "created_at", "content_class") + if missing := _missing(columns, required): + return _unmeasurable(f"required column missing: chunks.{missing}", db_input) + cutoff = _iso_utc(now - timedelta(hours=WINDOW_HOURS)) + now_text = _iso_utc(now) + total = connection.execute("SELECT COUNT(*) FROM chunks").fetchone()[0] + rows = connection.execute("SELECT content_class, COUNT(*) FROM chunks GROUP BY content_class ORDER BY content_class IS NULL, content_class") + content_classes = [{"content_class": row[0], "count": row[1]} for row in rows] + rows = connection.execute("SELECT strftime('%Y-%m-%dT%H:00:00Z', created_at), COUNT(*) FROM chunks WHERE created_at >= ? AND created_at <= ? GROUP BY 1 ORDER BY 1", (cutoff, now_text)) + by_hour = [{"hour": row[0], "count": row[1]} for row in rows] + latest = [] + for row in connection.execute( + "SELECT id, created_at, source_class, source, sender, source_file, content " + "FROM chunks ORDER BY created_at DESC LIMIT 5" + ): + emitter, _ = derive_emitter(row[3], row[4], row[5]) + latest.append({"chunk_id": str(row[0]), "stored_at": _iso_utc(_parse_time(row[1])), + "source_class": row[2], "emitter": emitter, + "preview": scrub_secrets(row[6] or "").text[:80]}) + return _measured(db_input, total_chunks=total, + in_window={"count": sum(item["count"] for item in by_hour), "by_hour": by_hour}, + by_content_class=content_classes, latest=latest) + + +def _emitters(connection: sqlite3.Connection, columns: set[str], db_input: dict[str, Any], now: datetime) -> dict[str, Any]: + required = ("source_class", "source", "sender", "source_file", "created_at") + if missing := _missing(columns, required): + return _unmeasurable(f"required column missing: chunks.{missing}", db_input) + cutoff, now_text = _iso_utc(now - timedelta(hours=WINDOW_HOURS)), _iso_utc(now) + rows = connection.execute("SELECT source_class, COUNT(*), SUM(CASE WHEN created_at >= ? AND created_at <= ? THEN 1 ELSE 0 END) FROM chunks GROUP BY source_class ORDER BY source_class IS NULL, source_class", (cutoff, now_text)) + by_source = [{"source_class": row[0], "count": row[1], "in_window": row[2]} for row in rows] + counts: Counter[tuple[str, str]] = Counter() + for row in connection.execute( + "SELECT source, sender, source_file FROM chunks WHERE created_at >= ? AND created_at <= ?", + (cutoff, now_text), + ): + counts[derive_emitter(*row)] += 1 + by_emitter = [{"emitter": emitter, "derived_from": origin, "count_in_window": count} + for (emitter, origin), count in sorted(counts.items())] + hidden = connection.execute("SELECT COUNT(*) FROM chunks WHERE source_class IN ('desktop', 'brain-worker')").fetchone()[0] + return _measured(db_input, by_source_class=by_source, by_emitter=by_emitter, + derivation_note=DERIVATION_NOTE, hidden_from_default_search=hidden) + + +def _author_unknown(connection: sqlite3.Connection, columns: set[str], db_input: dict[str, Any], now: datetime) -> dict[str, Any]: + required = ("source_class", "provenance_class", "archived_at", "created_at", "source_file") + if missing := _missing(columns, required): + return _unmeasurable(f"required column missing: chunks.{missing}", db_input) + live = connection.execute("SELECT COUNT(*) FROM chunks WHERE archived_at IS NULL").fetchone()[0] + never_where = "archived_at IS NULL AND (provenance_class IS NULL OR source_class IS NULL)" + unknown_where = "archived_at IS NULL AND provenance_class = 'unknown'" + never = connection.execute(f"SELECT COUNT(*) FROM chunks WHERE {never_where}").fetchone()[0] + unknown = connection.execute(f"SELECT COUNT(*) FROM chunks WHERE {unknown_where}").fetchone()[0] + start_day = (now - timedelta(days=6)).date().isoformat() + daily: dict[str, list[int]] = {} + for row in connection.execute( + f"SELECT date(created_at), SUM(CASE WHEN {never_where} THEN 1 ELSE 0 END), " + f"SUM(CASE WHEN {unknown_where} THEN 1 ELSE 0 END) FROM chunks " + "WHERE date(created_at) >= ? AND date(created_at) <= ? GROUP BY 1", + (start_day, now.date().isoformat()), + ): + daily[row[0]] = [row[1], row[2]] + trend = [] + for offset in range(7): + day = (now.date() - timedelta(days=6 - offset)).isoformat() + values = daily.get(day, [0, 0]) + trend.append({"day": day, "never_classified": values[0], "classified_unknown": values[1]}) + rows = connection.execute(f"SELECT source_file, COUNT(*) FROM chunks WHERE {never_where} OR ({unknown_where}) GROUP BY source_file ORDER BY COUNT(*) DESC, source_file LIMIT 5") + top_files = [{"source_file": row[0] or "", "count": row[1]} for row in rows] + return _measured( + db_input, + never_classified={ + "count": never, + "share": round(never / live, 6) if live else 0.0, + "definition": "provenance_class IS NULL OR source_class IS NULL, archived_at IS NULL", + }, + classified_unknown={ + "count": unknown, + "share": round(unknown / live, 6) if live else 0.0, + "definition": "provenance_class = 'unknown', archived_at IS NULL", + }, + census_2026_09_13=CENSUS_NOTE, + trend_7d=trend, + top_source_files=top_files, + ) + + +def build_document(*, env: Mapping[str, str] = os.environ) -> tuple[dict[str, Any], InputRecorder]: + expected_root = env.get("BRAINLAYER_OBSERVABILITY_PRODUCER_ROOT") + if expected_root and not Path(__file__).resolve().is_relative_to(Path(expected_root).resolve() / "src"): + raise RuntimeError(f"observability_surface imported outside producer root: {Path(__file__).resolve()}") + now = _parse_time(env.get("BRAINLAYER_OBSERVABILITY_NOW", datetime.now(UTC).isoformat())) + db_path = Path(env.get("BRAINLAYER_DB", str(get_db_path()))).expanduser().resolve() + root = Path(env["BRAINLAYER_OBSERVABILITY_INPUT_ROOT"]).expanduser() if env.get("BRAINLAYER_OBSERVABILITY_INPUT_ROOT") else None + trace_path = ( + Path(env["BRAINLAYER_OBSERVABILITY_TRACE_PATH"]).expanduser() if env.get("BRAINLAYER_OBSERVABILITY_TRACE_PATH") else None + ) + recorder = InputRecorder(root=root, trace_path=trace_path, now=now) + connection: sqlite3.Connection | None = None + db_status, db_rows, failure = "read", None, None + try: + connection = sqlite3.connect(f"{db_path.as_uri()}?mode=ro", uri=True) + connection.execute("PRAGMA query_only=ON") + db_rows = connection.execute("SELECT COUNT(*) FROM chunks").fetchone()[0] + columns = {row[1] for row in connection.execute("PRAGMA table_info(chunks)")} + except (OSError, sqlite3.Error) as exc: + db_status = "missing" if not db_path.exists() else "malformed" + columns, failure = set(), f"database unreadable: {exc}" + db_input = recorder(db_path, status=db_status, rows_or_bytes=db_rows) + if db_input["status"] in {"missing", "malformed", "empty", "future"}: + failure = failure or f"database input status: {db_input['status']}" + if failure or connection is None: + stores = emitters = author_unknown = _unmeasurable(failure or "database unavailable", db_input) + else: + stores = _stores(connection, columns, db_input, now) + emitters = _emitters(connection, columns, db_input, now) + author_unknown = _author_unknown(connection, columns, db_input, now) + if connection is not None: + connection.close() + try: + from .observability_backup import build_backups_section + + backups = build_backups_section(env=env, record_input=recorder, now=now) + except ImportError: + for name in ("JSONL_BACKUP_LOG", "BACKUP_DAILY_LOG", "LAUNCHD_OUTPUT", "DISABLED_DIR"): + if path := env.get(f"BRAINLAYER_OBSERVABILITY_{name}"): + recorder(path, in_section_inputs=False) + backups = {"state": "unmeasurable", "reason": "backups module not installed", "inputs": []} + document = { + "schema_version": 1, + "generated_at": _iso_utc(now), + "db_path": recorder.display_path(db_path), + "window_hours": WINDOW_HOURS, + "stores": stores, + "emitters": emitters, + "author_unknown": author_unknown, + "backups": backups, + } + return document, recorder +def write_document(*, env: Mapping[str, str] = os.environ, stdout: bool = False) -> dict[str, Any]: + document, recorder = build_document(env=env) + try: + payload = json.dumps(document, indent=2, sort_keys=True) + "\n" + if stdout: + print(payload, end="") + else: + db_path = Path(env.get("BRAINLAYER_DB", str(get_db_path()))).expanduser().resolve() + output = Path(env.get("BRAINLAYER_OBSERVABILITY_PATH", str(db_path.parent / "observability.json"))).expanduser() + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(payload, encoding="utf-8") + return document + finally: + recorder.write_trace() + + +def main() -> int: + write_document() + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) +# fmt: on diff --git a/tests/test_observability_surface.py b/tests/test_observability_surface.py index 277becaf..2c832bb9 100644 --- a/tests/test_observability_surface.py +++ b/tests/test_observability_surface.py @@ -2,12 +2,14 @@ import json import os +import shutil import subprocess import sys +from datetime import datetime from pathlib import Path import pytest - +from typer.testing import CliRunner REPO = Path(__file__).resolve().parents[1] FIXTURES = REPO / "tests/fixtures/observability" @@ -22,24 +24,25 @@ def _dev_cases() -> list[dict[str, object]]: def _run_case(case: dict[str, object], tmp_path: Path) -> tuple[dict[str, object], list[str]]: inputs = case["inputs"] assert isinstance(inputs, dict) + source_db = FIXTURES / str(inputs["db"]) + db_relative = Path(str(inputs["db"])) if not Path(str(inputs["db"])).is_absolute() else Path("db") / source_db.name + staged_db = tmp_path / "inputs" / db_relative + staged_db.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source_db, staged_db) + fixed = int(datetime.fromisoformat(str(case["generated_at"]).replace("Z", "+00:00")).timestamp()) + os.utime(staged_db, (fixed, fixed)) output = tmp_path / "observability.json" trace = tmp_path / "trace.json" - env = { - key: os.environ[key] - for key in ("HOME", "PATH", "BRAINLAYER_FORBID_EMBEDDING_MODEL") - if key in os.environ - } + env = {key: os.environ[key] for key in ("HOME", "PATH", "BRAINLAYER_FORBID_EMBEDDING_MODEL") if key in os.environ} + input_env = {"jsonl_backup_log": "BRAINLAYER_OBSERVABILITY_JSONL_BACKUP_LOG", "backup_daily_log": "BRAINLAYER_OBSERVABILITY_BACKUP_DAILY_LOG", "launchd_output": "BRAINLAYER_OBSERVABILITY_LAUNCHD_OUTPUT", "disabled_dir": "BRAINLAYER_OBSERVABILITY_DISABLED_DIR"} # fmt: skip + env.update({target: str(FIXTURES / str(inputs[source])) for source, target in input_env.items()}) env.update( { "PYTHONPATH": str(REPO / "src"), - "BRAINLAYER_DB": str(FIXTURES / str(inputs["db"])), + "BRAINLAYER_DB": str(staged_db), "BRAINLAYER_OBSERVABILITY_PATH": str(output), "BRAINLAYER_OBSERVABILITY_TRACE_PATH": str(trace), - "BRAINLAYER_OBSERVABILITY_INPUT_ROOT": str(FIXTURES), - "BRAINLAYER_OBSERVABILITY_JSONL_BACKUP_LOG": str(FIXTURES / str(inputs["jsonl_backup_log"])), - "BRAINLAYER_OBSERVABILITY_BACKUP_DAILY_LOG": str(FIXTURES / str(inputs["backup_daily_log"])), - "BRAINLAYER_OBSERVABILITY_LAUNCHD_OUTPUT": str(FIXTURES / str(inputs["launchd_output"])), - "BRAINLAYER_OBSERVABILITY_DISABLED_DIR": str(FIXTURES / str(inputs["disabled_dir"])), + "BRAINLAYER_OBSERVABILITY_INPUT_ROOT": str(tmp_path / "inputs"), "BRAINLAYER_OBSERVABILITY_NOW": str(case["generated_at"]), } ) @@ -60,7 +63,6 @@ def test_dev_goldens_for_owned_sections(case: dict[str, object], tmp_path: Path) actual, trace = _run_case(case, tmp_path) golden_path = FIXTURES / "golden" / f"{case['case_id']}.json" expected = json.loads(golden_path.read_text(encoding="utf-8")) - for section in OWNED_SECTIONS: assert actual[section] == expected[section] assert actual["db_path"] == expected["db_path"] @@ -68,7 +70,6 @@ def test_dev_goldens_for_owned_sections(case: dict[str, object], tmp_path: Path) def test_preview_is_secret_scrubbed_and_limited_to_80_chars(tmp_path: Path) -> None: - import shutil import sqlite3 case = next(case for case in _dev_cases() if case["case_id"] == "healthy-dev") @@ -79,6 +80,8 @@ def test_preview_is_secret_scrubbed_and_limited_to_80_chars(tmp_path: Path) -> N connection.execute("UPDATE chunks SET content = ? WHERE id = 'synthetic-00'", (secret + "x" * 100,)) connection.commit() connection.close() + frozen = int(datetime.fromisoformat(str(case["generated_at"]).replace("Z", "+00:00")).timestamp()) + os.utime(db, (frozen, frozen)) case = {**case, "inputs": {**case["inputs"], "db": str(db)}} actual, _ = _run_case(case, tmp_path) @@ -88,18 +91,21 @@ def test_preview_is_secret_scrubbed_and_limited_to_80_chars(tmp_path: Path) -> N assert len(preview) <= 80 -@pytest.mark.parametrize( - ("source_file", "expected"), - [ - ("/Users/x/.claude/projects/-Users-x-Gits-brainlayer/session.jsonl", "brainlayer"), - ("/Users/x/.codex/sessions/2026/09/13/rollout.jsonl", "codex"), - ("brainbar-store", "brainbar-store"), - ("realtime-hook", "realtime-hook"), - ("unknown", "unknown"), - ("", "unknown"), - ], -) +@pytest.mark.parametrize(("source_file", "expected"), [("/Users/x/.claude/projects/-Users-x-Gits-brainlayer/session.jsonl", "brainlayer"), ("/Users/x/.codex/sessions/2026/09/13/rollout.jsonl", "codex"), ("brainbar-store", "brainbar-store"), ("realtime-hook", "realtime-hook"), ("unknown", "unknown"), ("", "unknown")]) # fmt: skip def test_source_file_emitter_derivation(source_file: str, expected: str) -> None: from brainlayer.observability_surface import derive_emitter assert derive_emitter(None, None, source_file) == (expected, "source_file") + + +def test_cli_observability_stdout(monkeypatch: pytest.MonkeyPatch) -> None: + from brainlayer.cli import app + + case = next(case for case in _dev_cases() if case["case_id"] == "empty-db-dev") + monkeypatch.setenv("BRAINLAYER_DB", str(FIXTURES / str(case["inputs"]["db"]))) + monkeypatch.setenv("BRAINLAYER_OBSERVABILITY_NOW", str(case["generated_at"])) + result = CliRunner().invoke(app, ["observability", "--stdout"]) + assert result.exit_code == 0 + assert json.loads(result.stdout)["stores"]["total_chunks"] == 0 + monkeypatch.setenv("BRAINLAYER_OBSERVABILITY_PRODUCER_ROOT", str(REPO.parent)) + assert CliRunner().invoke(app, ["observability", "--stdout"]).exit_code == 1 From e70958d6f6d738f157f0da2e9bbfe281a40932e3 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Mon, 14 Sep 2026 01:00:54 +0300 Subject: [PATCH 3/5] fix(observability): harden previews and timestamp ordering Co-Authored-By: brainlayerCodex-8726de2b running gpt-5.6-sol --- src/brainlayer/observability_surface.py | 21 ++++++-------------- tests/test_observability_surface.py | 26 +++++++++++++++++++------ 2 files changed, 26 insertions(+), 21 deletions(-) diff --git a/src/brainlayer/observability_surface.py b/src/brainlayer/observability_surface.py index 893f6d1a..06cd3b28 100644 --- a/src/brainlayer/observability_surface.py +++ b/src/brainlayer/observability_surface.py @@ -21,8 +21,6 @@ CENSUS_NOTE = "never_classified = 476,679 of 731,153 live (65.2%); both-NULL = 65,017; classified_unknown literal not observed on the copy" def _iso_utc(value: datetime) -> str: return value.astimezone(UTC).replace(microsecond=0).isoformat().replace("+00:00", "Z") - - def _parse_time(value: str) -> datetime: parsed = datetime.fromisoformat(value.replace("Z", "+00:00")) return parsed.replace(tzinfo=UTC) if parsed.tzinfo is None else parsed.astimezone(UTC) @@ -68,8 +66,6 @@ def write_trace(self) -> None: if self.trace_path is not None: self.trace_path.parent.mkdir(parents=True, exist_ok=True) self.trace_path.write_text(json.dumps(self.trace, indent=2) + "\n", encoding="utf-8") - - def _clean(value: object) -> str | None: text = str(value).strip() if value is not None else "" return text or None @@ -77,8 +73,6 @@ def _input(path: str, status: str, mtime: str | None, size: int | None, digest: str | None, skipped: int) -> dict[str, Any]: return {"path": path, "status": status, "mtime": mtime, "rows_or_bytes": size, "sha256_first_64kb": digest, "skipped_lines": skipped} - - def derive_emitter(source: object, sender: object, source_file: object) -> tuple[str, str]: """Derive the emitter in source, sender, source_file precedence order.""" for value, origin in ((source, "source"), (sender, "sender")): @@ -93,8 +87,6 @@ def derive_emitter(source: object, sender: object, source_file: object) -> tuple return project_dir.rsplit("-Gits-", 1)[1] or "unknown", "source_file" return project_dir or "unknown", "source_file" return path, "source_file" - - def _unmeasurable(reason: str, db_input: dict[str, Any]) -> dict[str, Any]: return {"state": "unmeasurable", "reason": reason, "inputs": [db_input]} def _measured(db_input: dict[str, Any], **fields: Any) -> dict[str, Any]: @@ -112,32 +104,31 @@ def _stores(connection: sqlite3.Connection, columns: set[str], db_input: dict[st total = connection.execute("SELECT COUNT(*) FROM chunks").fetchone()[0] rows = connection.execute("SELECT content_class, COUNT(*) FROM chunks GROUP BY content_class ORDER BY content_class IS NULL, content_class") content_classes = [{"content_class": row[0], "count": row[1]} for row in rows] - rows = connection.execute("SELECT strftime('%Y-%m-%dT%H:00:00Z', created_at), COUNT(*) FROM chunks WHERE created_at >= ? AND created_at <= ? GROUP BY 1 ORDER BY 1", (cutoff, now_text)) + rows = connection.execute("SELECT strftime('%Y-%m-%dT%H:00:00Z', created_at), COUNT(*) FROM chunks WHERE datetime(created_at) >= datetime(?) AND datetime(created_at) <= datetime(?) GROUP BY 1 ORDER BY 1", (cutoff, now_text)) by_hour = [{"hour": row[0], "count": row[1]} for row in rows] latest = [] for row in connection.execute( "SELECT id, created_at, source_class, source, sender, source_file, content " - "FROM chunks ORDER BY created_at DESC LIMIT 5" + "FROM chunks ORDER BY datetime(created_at) DESC, id LIMIT 5" ): emitter, _ = derive_emitter(row[3], row[4], row[5]) + scrubbed = scrub_secrets(row[6] or "") latest.append({"chunk_id": str(row[0]), "stored_at": _iso_utc(_parse_time(row[1])), "source_class": row[2], "emitter": emitter, - "preview": scrub_secrets(row[6] or "").text[:80]}) + "preview": "[REDACTED:quarantined]" if scrubbed.quarantine else scrubbed.text[:80]}) return _measured(db_input, total_chunks=total, in_window={"count": sum(item["count"] for item in by_hour), "by_hour": by_hour}, by_content_class=content_classes, latest=latest) - - def _emitters(connection: sqlite3.Connection, columns: set[str], db_input: dict[str, Any], now: datetime) -> dict[str, Any]: required = ("source_class", "source", "sender", "source_file", "created_at") if missing := _missing(columns, required): return _unmeasurable(f"required column missing: chunks.{missing}", db_input) cutoff, now_text = _iso_utc(now - timedelta(hours=WINDOW_HOURS)), _iso_utc(now) - rows = connection.execute("SELECT source_class, COUNT(*), SUM(CASE WHEN created_at >= ? AND created_at <= ? THEN 1 ELSE 0 END) FROM chunks GROUP BY source_class ORDER BY source_class IS NULL, source_class", (cutoff, now_text)) + rows = connection.execute("SELECT source_class, COUNT(*), SUM(CASE WHEN datetime(created_at) >= datetime(?) AND datetime(created_at) <= datetime(?) THEN 1 ELSE 0 END) FROM chunks GROUP BY source_class ORDER BY source_class IS NULL, source_class", (cutoff, now_text)) by_source = [{"source_class": row[0], "count": row[1], "in_window": row[2]} for row in rows] counts: Counter[tuple[str, str]] = Counter() for row in connection.execute( - "SELECT source, sender, source_file FROM chunks WHERE created_at >= ? AND created_at <= ?", + "SELECT source, sender, source_file FROM chunks WHERE datetime(created_at) >= datetime(?) AND datetime(created_at) <= datetime(?)", (cutoff, now_text), ): counts[derive_emitter(*row)] += 1 diff --git a/tests/test_observability_surface.py b/tests/test_observability_surface.py index 2c832bb9..9238cb87 100644 --- a/tests/test_observability_surface.py +++ b/tests/test_observability_surface.py @@ -3,6 +3,7 @@ import json import os import shutil +import sqlite3 import subprocess import sys from datetime import datetime @@ -69,15 +70,14 @@ def test_dev_goldens_for_owned_sections(case: dict[str, object], tmp_path: Path) assert trace.count(str(case["inputs"]["db"])) == 1 -def test_preview_is_secret_scrubbed_and_limited_to_80_chars(tmp_path: Path) -> None: - import sqlite3 - +@pytest.mark.parametrize(("secret", "prefix"), [("sk-ant-" + "A" * 30, "[REDACTED:anthropic]"), ("Q7mV2pL9xR4cT8nW3kY6dF1sH5jB", "[REDACTED:quarantined]")]) # fmt: skip +def test_preview_is_secret_scrubbed_and_limited_to_80_chars(tmp_path: Path, secret: str, prefix: str) -> None: case = next(case for case in _dev_cases() if case["case_id"] == "healthy-dev") db = tmp_path / "preview.sqlite" shutil.copy2(FIXTURES / str(case["inputs"]["db"]), db) - secret = "sk-ant-" + "A" * 30 connection = sqlite3.connect(db) - connection.execute("UPDATE chunks SET content = ? WHERE id = 'synthetic-00'", (secret + "x" * 100,)) + connection.execute("PRAGMA journal_mode=DELETE") + connection.execute("UPDATE chunks SET content = ? WHERE id = 'synthetic-00'", (secret + " " + "x" * 100,)) connection.commit() connection.close() frozen = int(datetime.fromisoformat(str(case["generated_at"]).replace("Z", "+00:00")).timestamp()) @@ -87,10 +87,24 @@ def test_preview_is_secret_scrubbed_and_limited_to_80_chars(tmp_path: Path) -> N actual, _ = _run_case(case, tmp_path) preview = actual["stores"]["latest"][0]["preview"] assert secret not in preview - assert preview.startswith("[REDACTED:anthropic]") + assert preview.startswith(prefix) assert len(preview) <= 80 +def test_offset_timestamps_are_compared_as_utc_instants(tmp_path: Path) -> None: + case = next(case for case in _dev_cases() if case["case_id"] == "healthy-dev") + db = tmp_path / "offset.sqlite" + shutil.copy2(FIXTURES / str(case["inputs"]["db"]), db) + with sqlite3.connect(db) as connection: + connection.execute("PRAGMA journal_mode=DELETE") + connection.executemany("UPDATE chunks SET created_at = ? WHERE id = ?", [("2026-09-13T13:00:00+02:00", "synthetic-00"), ("2026-09-13T11:30:00Z", "synthetic-01"), ("2026-09-12T13:00:00+14:00", "synthetic-02")]) # fmt: skip + case = {**case, "inputs": {**case["inputs"], "db": str(db)}} + actual, _ = _run_case(case, tmp_path) + assert actual["stores"]["latest"][0]["chunk_id"] == "synthetic-01" + assert actual["stores"]["in_window"]["count"] == 11 + assert sum(item["count_in_window"] for item in actual["emitters"]["by_emitter"]) == 11 + + @pytest.mark.parametrize(("source_file", "expected"), [("/Users/x/.claude/projects/-Users-x-Gits-brainlayer/session.jsonl", "brainlayer"), ("/Users/x/.codex/sessions/2026/09/13/rollout.jsonl", "codex"), ("brainbar-store", "brainbar-store"), ("realtime-hook", "realtime-hook"), ("unknown", "unknown"), ("", "unknown")]) # fmt: skip def test_source_file_emitter_derivation(source_file: str, expected: str) -> None: from brainlayer.observability_surface import derive_emitter From c6197b6abf8c5dbe5e2aaa7be7ebc94cdd415dde Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Mon, 14 Sep 2026 01:11:44 +0300 Subject: [PATCH 4/5] fix(observability): close recorder review gaps Co-Authored-By: OpenAI Codex --- src/brainlayer/observability_surface.py | 137 ++++++++++-------------- tests/test_observability_surface.py | 62 +++++++---- 2 files changed, 95 insertions(+), 104 deletions(-) diff --git a/src/brainlayer/observability_surface.py b/src/brainlayer/observability_surface.py index 06cd3b28..377b87ac 100644 --- a/src/brainlayer/observability_surface.py +++ b/src/brainlayer/observability_surface.py @@ -26,11 +26,9 @@ def _parse_time(value: str) -> datetime: return parsed.replace(tzinfo=UTC) if parsed.tzinfo is None else parsed.astimezone(UTC) class InputRecorder: def __init__(self, *, root: Path | None, trace_path: Path | None, now: datetime) -> None: - self.root = root.resolve() if root else None - self.trace_path = trace_path - self.now = now + self.root, self.trace_path, self.now = root.resolve() if root else None, trace_path, now self.trace: list[str] = [] - + self.section_inputs: list[dict[str, Any]] = [] def display_path(self, path: Path) -> str: resolved = path.expanduser().resolve() if self.root is not None: @@ -39,29 +37,30 @@ def display_path(self, path: Path) -> str: except ValueError: pass return str(resolved) - def __call__(self, path: Path | str, *, status: str = "read", rows_or_bytes: int | None = None, skipped_lines: int = 0, in_section_inputs: bool = True) -> dict[str, Any]: - del in_section_inputs # The caller decides whether to retain the returned object. resolved = Path(path).expanduser().resolve() displayed = self.display_path(resolved) self.trace.append(displayed) try: stat = resolved.stat() except FileNotFoundError: - return _input(displayed, "missing", None, None, None, skipped_lines) - effective_status = status - if resolved.is_file() and stat.st_size == 0: - effective_status = "empty" - elif status == "read" and datetime.fromtimestamp(stat.st_mtime, UTC) > self.now: - effective_status = "future" - digest = None - if resolved.is_file() and resolved.suffix not in {".sqlite", ".db"}: - with resolved.open("rb") as handle: - digest = hashlib.sha256(handle.read(65_536)).hexdigest() - return _input(displayed, effective_status, _iso_utc(datetime.fromtimestamp(stat.st_mtime, UTC)), - stat.st_size if rows_or_bytes is None else rows_or_bytes, digest, skipped_lines) - + item = _input(displayed, "missing", None, None, None, skipped_lines) + else: + effective_status = status + if resolved.is_file() and stat.st_size == 0: + effective_status = "empty" + elif status == "read" and datetime.fromtimestamp(stat.st_mtime, UTC) > self.now: + effective_status = "future" + digest = None + if resolved.is_file() and resolved.suffix not in {".sqlite", ".db"}: + with resolved.open("rb") as handle: + digest = hashlib.sha256(handle.read(65_536)).hexdigest() + item = _input(displayed, effective_status, _iso_utc(datetime.fromtimestamp(stat.st_mtime, UTC)), + stat.st_size if rows_or_bytes is None else rows_or_bytes, digest, skipped_lines) + if in_section_inputs: + self.section_inputs.append(item) + return item def write_trace(self) -> None: if self.trace_path is not None: self.trace_path.parent.mkdir(parents=True, exist_ok=True) @@ -74,7 +73,6 @@ def _input(path: str, status: str, mtime: str | None, size: int | None, return {"path": path, "status": status, "mtime": mtime, "rows_or_bytes": size, "sha256_first_64kb": digest, "skipped_lines": skipped} def derive_emitter(source: object, sender: object, source_file: object) -> tuple[str, str]: - """Derive the emitter in source, sender, source_file precedence order.""" for value, origin in ((source, "source"), (sender, "sender")): if cleaned := _clean(value): return cleaned, origin @@ -91,26 +89,21 @@ def _unmeasurable(reason: str, db_input: dict[str, Any]) -> dict[str, Any]: return {"state": "unmeasurable", "reason": reason, "inputs": [db_input]} def _measured(db_input: dict[str, Any], **fields: Any) -> dict[str, Any]: return {"state": "measured", "reason": "", "inputs": [db_input], **fields} - - def _missing(columns: set[str], required: tuple[str, ...]) -> str | None: return next((name for name in required if name not in columns), None) def _stores(connection: sqlite3.Connection, columns: set[str], db_input: dict[str, Any], now: datetime) -> dict[str, Any]: required = ("source_class", "id", "content", "source", "sender", "source_file", "created_at", "content_class") if missing := _missing(columns, required): return _unmeasurable(f"required column missing: chunks.{missing}", db_input) - cutoff = _iso_utc(now - timedelta(hours=WINDOW_HOURS)) - now_text = _iso_utc(now) + cutoff, now_text = _iso_utc(now - timedelta(hours=WINDOW_HOURS)), _iso_utc(now) total = connection.execute("SELECT COUNT(*) FROM chunks").fetchone()[0] rows = connection.execute("SELECT content_class, COUNT(*) FROM chunks GROUP BY content_class ORDER BY content_class IS NULL, content_class") content_classes = [{"content_class": row[0], "count": row[1]} for row in rows] rows = connection.execute("SELECT strftime('%Y-%m-%dT%H:00:00Z', created_at), COUNT(*) FROM chunks WHERE datetime(created_at) >= datetime(?) AND datetime(created_at) <= datetime(?) GROUP BY 1 ORDER BY 1", (cutoff, now_text)) by_hour = [{"hour": row[0], "count": row[1]} for row in rows] latest = [] - for row in connection.execute( - "SELECT id, created_at, source_class, source, sender, source_file, content " - "FROM chunks ORDER BY datetime(created_at) DESC, id LIMIT 5" - ): + for row in connection.execute("SELECT id, created_at, source_class, source, sender, source_file, content " + "FROM chunks ORDER BY datetime(created_at) DESC, id LIMIT 5"): emitter, _ = derive_emitter(row[3], row[4], row[5]) scrubbed = scrub_secrets(row[6] or "") latest.append({"chunk_id": str(row[0]), "stored_at": _iso_utc(_parse_time(row[1])), @@ -137,8 +130,6 @@ def _emitters(connection: sqlite3.Connection, columns: set[str], db_input: dict[ hidden = connection.execute("SELECT COUNT(*) FROM chunks WHERE source_class IN ('desktop', 'brain-worker')").fetchone()[0] return _measured(db_input, by_source_class=by_source, by_emitter=by_emitter, derivation_note=DERIVATION_NOTE, hidden_from_default_search=hidden) - - def _author_unknown(connection: sqlite3.Connection, columns: set[str], db_input: dict[str, Any], now: datetime) -> dict[str, Any]: required = ("source_class", "provenance_class", "archived_at", "created_at", "source_file") if missing := _missing(columns, required): @@ -164,35 +155,26 @@ def _author_unknown(connection: sqlite3.Connection, columns: set[str], db_input: trend.append({"day": day, "never_classified": values[0], "classified_unknown": values[1]}) rows = connection.execute(f"SELECT source_file, COUNT(*) FROM chunks WHERE {never_where} OR ({unknown_where}) GROUP BY source_file ORDER BY COUNT(*) DESC, source_file LIMIT 5") top_files = [{"source_file": row[0] or "", "count": row[1]} for row in rows] - return _measured( - db_input, - never_classified={ - "count": never, - "share": round(never / live, 6) if live else 0.0, - "definition": "provenance_class IS NULL OR source_class IS NULL, archived_at IS NULL", - }, - classified_unknown={ - "count": unknown, - "share": round(unknown / live, 6) if live else 0.0, - "definition": "provenance_class = 'unknown', archived_at IS NULL", - }, - census_2026_09_13=CENSUS_NOTE, - trend_7d=trend, - top_source_files=top_files, - ) - - + never_fields = {"count": never, "share": round(never / live, 6) if live else 0.0, + "definition": "provenance_class IS NULL OR source_class IS NULL, archived_at IS NULL"} + unknown_fields = {"count": unknown, "share": round(unknown / live, 6) if live else 0.0, + "definition": "provenance_class = 'unknown', archived_at IS NULL"} + return _measured(db_input, never_classified=never_fields, classified_unknown=unknown_fields, + census_2026_09_13=CENSUS_NOTE, trend_7d=trend, top_source_files=top_files) def build_document(*, env: Mapping[str, str] = os.environ) -> tuple[dict[str, Any], InputRecorder]: expected_root = env.get("BRAINLAYER_OBSERVABILITY_PRODUCER_ROOT") if expected_root and not Path(__file__).resolve().is_relative_to(Path(expected_root).resolve() / "src"): raise RuntimeError(f"observability_surface imported outside producer root: {Path(__file__).resolve()}") now = _parse_time(env.get("BRAINLAYER_OBSERVABILITY_NOW", datetime.now(UTC).isoformat())) - db_path = Path(env.get("BRAINLAYER_DB", str(get_db_path()))).expanduser().resolve() root = Path(env["BRAINLAYER_OBSERVABILITY_INPUT_ROOT"]).expanduser() if env.get("BRAINLAYER_OBSERVABILITY_INPUT_ROOT") else None - trace_path = ( - Path(env["BRAINLAYER_OBSERVABILITY_TRACE_PATH"]).expanduser() if env.get("BRAINLAYER_OBSERVABILITY_TRACE_PATH") else None - ) + trace_path = Path(env["BRAINLAYER_OBSERVABILITY_TRACE_PATH"]).expanduser() if env.get("BRAINLAYER_OBSERVABILITY_TRACE_PATH") else None recorder = InputRecorder(root=root, trace_path=trace_path, now=now) + try: + return _build_document(env=env, now=now, recorder=recorder) + finally: + recorder.write_trace() +def _build_document(*, env: Mapping[str, str], now: datetime, recorder: InputRecorder) -> tuple[dict[str, Any], InputRecorder]: + db_path = (Path(env["BRAINLAYER_DB"]).expanduser().resolve() if env.get("BRAINLAYER_DB") else get_db_path()) connection: sqlite3.Connection | None = None db_status, db_rows, failure = "read", None, None try: @@ -209,9 +191,14 @@ def build_document(*, env: Mapping[str, str] = os.environ) -> tuple[dict[str, An if failure or connection is None: stores = emitters = author_unknown = _unmeasurable(failure or "database unavailable", db_input) else: - stores = _stores(connection, columns, db_input, now) - emitters = _emitters(connection, columns, db_input, now) - author_unknown = _author_unknown(connection, columns, db_input, now) + malformed_time = connection.execute("SELECT id FROM chunks WHERE created_at IS NOT NULL AND datetime(created_at) IS NULL ORDER BY id LIMIT 1").fetchone() if "created_at" in columns else None # fmt: skip + if malformed_time: + reason = f"malformed created_at: chunks row {malformed_time[0]}" + stores = emitters = author_unknown = _unmeasurable(reason, db_input) + else: + stores = _stores(connection, columns, db_input, now) + emitters = _emitters(connection, columns, db_input, now) + author_unknown = _author_unknown(connection, columns, db_input, now) if connection is not None: connection.close() try: @@ -223,38 +210,24 @@ def build_document(*, env: Mapping[str, str] = os.environ) -> tuple[dict[str, An if path := env.get(f"BRAINLAYER_OBSERVABILITY_{name}"): recorder(path, in_section_inputs=False) backups = {"state": "unmeasurable", "reason": "backups module not installed", "inputs": []} - document = { - "schema_version": 1, - "generated_at": _iso_utc(now), - "db_path": recorder.display_path(db_path), - "window_hours": WINDOW_HOURS, - "stores": stores, - "emitters": emitters, - "author_unknown": author_unknown, - "backups": backups, - } + document = {"schema_version": 1, "generated_at": _iso_utc(now), "db_path": recorder.display_path(db_path), + "window_hours": WINDOW_HOURS, "stores": stores, "emitters": emitters, + "author_unknown": author_unknown, "backups": backups} return document, recorder def write_document(*, env: Mapping[str, str] = os.environ, stdout: bool = False) -> dict[str, Any]: - document, recorder = build_document(env=env) - try: - payload = json.dumps(document, indent=2, sort_keys=True) + "\n" - if stdout: - print(payload, end="") - else: - db_path = Path(env.get("BRAINLAYER_DB", str(get_db_path()))).expanduser().resolve() - output = Path(env.get("BRAINLAYER_OBSERVABILITY_PATH", str(db_path.parent / "observability.json"))).expanduser() - output.parent.mkdir(parents=True, exist_ok=True) - output.write_text(payload, encoding="utf-8") - return document - finally: - recorder.write_trace() - - + document, _ = build_document(env=env) + payload = json.dumps(document, indent=2, sort_keys=True) + "\n" + if stdout: + print(payload, end="") + else: + db_path = (Path(env["BRAINLAYER_DB"]).expanduser().resolve() if env.get("BRAINLAYER_DB") else get_db_path()) + output = Path(env.get("BRAINLAYER_OBSERVABILITY_PATH", str(db_path.parent / "observability.json"))).expanduser() + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(payload, encoding="utf-8") + return document def main() -> int: write_document() return 0 - - if __name__ == "__main__": raise SystemExit(main()) # fmt: on diff --git a/tests/test_observability_surface.py b/tests/test_observability_surface.py index 9238cb87..e028cdb4 100644 --- a/tests/test_observability_surface.py +++ b/tests/test_observability_surface.py @@ -15,23 +15,24 @@ REPO = Path(__file__).resolve().parents[1] FIXTURES = REPO / "tests/fixtures/observability" OWNED_SECTIONS = ("stores", "emitters", "author_unknown") - - def _dev_cases() -> list[dict[str, object]]: manifest = json.loads((FIXTURES / "cases.json").read_text(encoding="utf-8")) return [case for case in manifest["cases"] if case["split"] == "dev"] - - -def _run_case(case: dict[str, object], tmp_path: Path) -> tuple[dict[str, object], list[str]]: +def _stage_db(case: dict[str, object], tmp_path: Path) -> Path: inputs = case["inputs"] assert isinstance(inputs, dict) - source_db = FIXTURES / str(inputs["db"]) - db_relative = Path(str(inputs["db"])) if not Path(str(inputs["db"])).is_absolute() else Path("db") / source_db.name - staged_db = tmp_path / "inputs" / db_relative - staged_db.parent.mkdir(parents=True, exist_ok=True) - shutil.copy2(source_db, staged_db) + source = FIXTURES / str(inputs["db"]) + relative = Path(str(inputs["db"])) if not Path(str(inputs["db"])).is_absolute() else Path("db") / source.name + staged = tmp_path / "inputs" / relative + staged.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(source, staged) fixed = int(datetime.fromisoformat(str(case["generated_at"]).replace("Z", "+00:00")).timestamp()) - os.utime(staged_db, (fixed, fixed)) + os.utime(staged, (fixed, fixed)) + return staged +def _run_case(case: dict[str, object], tmp_path: Path) -> tuple[dict[str, object], list[str]]: + inputs = case["inputs"] + assert isinstance(inputs, dict) + staged_db = _stage_db(case, tmp_path) output = tmp_path / "observability.json" trace = tmp_path / "trace.json" env = {key: os.environ[key] for key in ("HOME", "PATH", "BRAINLAYER_FORBID_EMBEDDING_MODEL") if key in os.environ} @@ -57,8 +58,6 @@ def _run_case(case: dict[str, object], tmp_path: Path) -> tuple[dict[str, object ) assert run.returncode == 0, run.stderr return json.loads(output.read_text(encoding="utf-8")), json.loads(trace.read_text(encoding="utf-8")) - - @pytest.mark.parametrize("case", _dev_cases(), ids=lambda case: str(case["case_id"])) def test_dev_goldens_for_owned_sections(case: dict[str, object], tmp_path: Path) -> None: actual, trace = _run_case(case, tmp_path) @@ -68,8 +67,6 @@ def test_dev_goldens_for_owned_sections(case: dict[str, object], tmp_path: Path) assert actual[section] == expected[section] assert actual["db_path"] == expected["db_path"] assert trace.count(str(case["inputs"]["db"])) == 1 - - @pytest.mark.parametrize(("secret", "prefix"), [("sk-ant-" + "A" * 30, "[REDACTED:anthropic]"), ("Q7mV2pL9xR4cT8nW3kY6dF1sH5jB", "[REDACTED:quarantined]")]) # fmt: skip def test_preview_is_secret_scrubbed_and_limited_to_80_chars(tmp_path: Path, secret: str, prefix: str) -> None: case = next(case for case in _dev_cases() if case["case_id"] == "healthy-dev") @@ -89,8 +86,6 @@ def test_preview_is_secret_scrubbed_and_limited_to_80_chars(tmp_path: Path, secr assert secret not in preview assert preview.startswith(prefix) assert len(preview) <= 80 - - def test_offset_timestamps_are_compared_as_utc_instants(tmp_path: Path) -> None: case = next(case for case in _dev_cases() if case["case_id"] == "healthy-dev") db = tmp_path / "offset.sqlite" @@ -103,20 +98,43 @@ def test_offset_timestamps_are_compared_as_utc_instants(tmp_path: Path) -> None: assert actual["stores"]["latest"][0]["chunk_id"] == "synthetic-01" assert actual["stores"]["in_window"]["count"] == 11 assert sum(item["count_in_window"] for item in actual["emitters"]["by_emitter"]) == 11 +def test_malformed_timestamp_degrades_sections_and_names_row(tmp_path: Path) -> None: + case = next(case for case in _dev_cases() if case["case_id"] == "healthy-dev") + db = tmp_path / "malformed-time.sqlite" + shutil.copy2(FIXTURES / str(case["inputs"]["db"]), db) + with sqlite3.connect(db) as connection: + connection.execute("PRAGMA journal_mode=DELETE") + connection.execute("UPDATE chunks SET created_at = 'not-a-time' WHERE id = 'synthetic-00'") + actual, _ = _run_case({**case, "inputs": {**case["inputs"], "db": str(db)}}, tmp_path) + for section in OWNED_SECTIONS: + assert actual[section]["state"] == "unmeasurable" + assert "synthetic-00" in actual[section]["reason"] +def test_trace_is_written_when_build_fails(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: + import brainlayer.observability_surface as surface - + case = next(case for case in _dev_cases() if case["case_id"] == "healthy-dev") + db, trace = _stage_db(case, tmp_path), tmp_path / "trace.json" + monkeypatch.setattr(surface, "_stores", lambda *_: (_ for _ in ()).throw(RuntimeError("boom"))) + with pytest.raises(RuntimeError, match="boom"): + surface.build_document(env={"BRAINLAYER_DB": str(db), "BRAINLAYER_OBSERVABILITY_INPUT_ROOT": str(tmp_path / "inputs"), "BRAINLAYER_OBSERVABILITY_TRACE_PATH": str(trace), "BRAINLAYER_OBSERVABILITY_NOW": str(case["generated_at"])}) # fmt: skip + assert json.loads(trace.read_text()) == [str(case["inputs"]["db"])] +def test_trace_only_input_is_excluded_from_recorder_section_inputs(tmp_path: Path) -> None: + from brainlayer.observability_surface import InputRecorder + + recorder = InputRecorder(root=tmp_path, trace_path=None, now=datetime.now().astimezone()) + included = recorder(tmp_path / "missing") + recorder(tmp_path, rows_or_bytes=0, in_section_inputs=False) + assert recorder.section_inputs == [included] @pytest.mark.parametrize(("source_file", "expected"), [("/Users/x/.claude/projects/-Users-x-Gits-brainlayer/session.jsonl", "brainlayer"), ("/Users/x/.codex/sessions/2026/09/13/rollout.jsonl", "codex"), ("brainbar-store", "brainbar-store"), ("realtime-hook", "realtime-hook"), ("unknown", "unknown"), ("", "unknown")]) # fmt: skip def test_source_file_emitter_derivation(source_file: str, expected: str) -> None: from brainlayer.observability_surface import derive_emitter assert derive_emitter(None, None, source_file) == (expected, "source_file") - - -def test_cli_observability_stdout(monkeypatch: pytest.MonkeyPatch) -> None: +def test_cli_observability_stdout(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: from brainlayer.cli import app case = next(case for case in _dev_cases() if case["case_id"] == "empty-db-dev") - monkeypatch.setenv("BRAINLAYER_DB", str(FIXTURES / str(case["inputs"]["db"]))) + monkeypatch.setenv("BRAINLAYER_DB", str(_stage_db(case, tmp_path))) monkeypatch.setenv("BRAINLAYER_OBSERVABILITY_NOW", str(case["generated_at"])) result = CliRunner().invoke(app, ["observability", "--stdout"]) assert result.exit_code == 0 From 3ead825aa4de246d1977c9f1858277296a0969f0 Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Mon, 14 Sep 2026 01:15:21 +0300 Subject: [PATCH 5/5] style: format Co-Authored-By: OpenAI Codex --- tests/test_observability_surface.py | 22 ++++++++++++++++++++++ 1 file changed, 22 insertions(+) diff --git a/tests/test_observability_surface.py b/tests/test_observability_surface.py index e028cdb4..38f0fafb 100644 --- a/tests/test_observability_surface.py +++ b/tests/test_observability_surface.py @@ -15,9 +15,13 @@ REPO = Path(__file__).resolve().parents[1] FIXTURES = REPO / "tests/fixtures/observability" OWNED_SECTIONS = ("stores", "emitters", "author_unknown") + + def _dev_cases() -> list[dict[str, object]]: manifest = json.loads((FIXTURES / "cases.json").read_text(encoding="utf-8")) return [case for case in manifest["cases"] if case["split"] == "dev"] + + def _stage_db(case: dict[str, object], tmp_path: Path) -> Path: inputs = case["inputs"] assert isinstance(inputs, dict) @@ -29,6 +33,8 @@ def _stage_db(case: dict[str, object], tmp_path: Path) -> Path: fixed = int(datetime.fromisoformat(str(case["generated_at"]).replace("Z", "+00:00")).timestamp()) os.utime(staged, (fixed, fixed)) return staged + + def _run_case(case: dict[str, object], tmp_path: Path) -> tuple[dict[str, object], list[str]]: inputs = case["inputs"] assert isinstance(inputs, dict) @@ -58,6 +64,8 @@ def _run_case(case: dict[str, object], tmp_path: Path) -> tuple[dict[str, object ) assert run.returncode == 0, run.stderr return json.loads(output.read_text(encoding="utf-8")), json.loads(trace.read_text(encoding="utf-8")) + + @pytest.mark.parametrize("case", _dev_cases(), ids=lambda case: str(case["case_id"])) def test_dev_goldens_for_owned_sections(case: dict[str, object], tmp_path: Path) -> None: actual, trace = _run_case(case, tmp_path) @@ -67,6 +75,8 @@ def test_dev_goldens_for_owned_sections(case: dict[str, object], tmp_path: Path) assert actual[section] == expected[section] assert actual["db_path"] == expected["db_path"] assert trace.count(str(case["inputs"]["db"])) == 1 + + @pytest.mark.parametrize(("secret", "prefix"), [("sk-ant-" + "A" * 30, "[REDACTED:anthropic]"), ("Q7mV2pL9xR4cT8nW3kY6dF1sH5jB", "[REDACTED:quarantined]")]) # fmt: skip def test_preview_is_secret_scrubbed_and_limited_to_80_chars(tmp_path: Path, secret: str, prefix: str) -> None: case = next(case for case in _dev_cases() if case["case_id"] == "healthy-dev") @@ -86,6 +96,8 @@ def test_preview_is_secret_scrubbed_and_limited_to_80_chars(tmp_path: Path, secr assert secret not in preview assert preview.startswith(prefix) assert len(preview) <= 80 + + def test_offset_timestamps_are_compared_as_utc_instants(tmp_path: Path) -> None: case = next(case for case in _dev_cases() if case["case_id"] == "healthy-dev") db = tmp_path / "offset.sqlite" @@ -98,6 +110,8 @@ def test_offset_timestamps_are_compared_as_utc_instants(tmp_path: Path) -> None: assert actual["stores"]["latest"][0]["chunk_id"] == "synthetic-01" assert actual["stores"]["in_window"]["count"] == 11 assert sum(item["count_in_window"] for item in actual["emitters"]["by_emitter"]) == 11 + + def test_malformed_timestamp_degrades_sections_and_names_row(tmp_path: Path) -> None: case = next(case for case in _dev_cases() if case["case_id"] == "healthy-dev") db = tmp_path / "malformed-time.sqlite" @@ -109,6 +123,8 @@ def test_malformed_timestamp_degrades_sections_and_names_row(tmp_path: Path) -> for section in OWNED_SECTIONS: assert actual[section]["state"] == "unmeasurable" assert "synthetic-00" in actual[section]["reason"] + + def test_trace_is_written_when_build_fails(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: import brainlayer.observability_surface as surface @@ -118,6 +134,8 @@ def test_trace_is_written_when_build_fails(monkeypatch: pytest.MonkeyPatch, tmp_ with pytest.raises(RuntimeError, match="boom"): surface.build_document(env={"BRAINLAYER_DB": str(db), "BRAINLAYER_OBSERVABILITY_INPUT_ROOT": str(tmp_path / "inputs"), "BRAINLAYER_OBSERVABILITY_TRACE_PATH": str(trace), "BRAINLAYER_OBSERVABILITY_NOW": str(case["generated_at"])}) # fmt: skip assert json.loads(trace.read_text()) == [str(case["inputs"]["db"])] + + def test_trace_only_input_is_excluded_from_recorder_section_inputs(tmp_path: Path) -> None: from brainlayer.observability_surface import InputRecorder @@ -125,11 +143,15 @@ def test_trace_only_input_is_excluded_from_recorder_section_inputs(tmp_path: Pat included = recorder(tmp_path / "missing") recorder(tmp_path, rows_or_bytes=0, in_section_inputs=False) assert recorder.section_inputs == [included] + + @pytest.mark.parametrize(("source_file", "expected"), [("/Users/x/.claude/projects/-Users-x-Gits-brainlayer/session.jsonl", "brainlayer"), ("/Users/x/.codex/sessions/2026/09/13/rollout.jsonl", "codex"), ("brainbar-store", "brainbar-store"), ("realtime-hook", "realtime-hook"), ("unknown", "unknown"), ("", "unknown")]) # fmt: skip def test_source_file_emitter_derivation(source_file: str, expected: str) -> None: from brainlayer.observability_surface import derive_emitter assert derive_emitter(None, None, source_file) == (expected, "source_file") + + def test_cli_observability_stdout(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> None: from brainlayer.cli import app