From c3be151608093247057aa09b09733eeda5a687ca Mon Sep 17 00:00:00 2001 From: Etan Joseph Heyman Date: Tue, 14 Jul 2026 12:49:07 +0300 Subject: [PATCH] fix: make watcher rewind archival APSW-safe --- .../BrainDatabaseWindowedBucketsTests.swift | 50 +++++++++ .../plans/2026-07-14-brainbar-truth-rewind.md | 104 ++++++++++++++++++ src/brainlayer/cli/__init__.py | 11 +- tests/test_rewind_batch_archival.py | 84 ++++++++++++++ 4 files changed, 248 insertions(+), 1 deletion(-) create mode 100644 docs/plans/2026-07-14-brainbar-truth-rewind.md diff --git a/brain-bar/Tests/BrainBarTests/BrainDatabaseWindowedBucketsTests.swift b/brain-bar/Tests/BrainBarTests/BrainDatabaseWindowedBucketsTests.swift index 1997b2a4..ffbb743f 100644 --- a/brain-bar/Tests/BrainBarTests/BrainDatabaseWindowedBucketsTests.swift +++ b/brain-bar/Tests/BrainBarTests/BrainDatabaseWindowedBucketsTests.swift @@ -12,6 +12,7 @@ import XCTest import SQLite3 +import Darwin @testable import BrainBar final class BrainDatabaseWindowedBucketsTests: XCTestCase { @@ -80,6 +81,20 @@ final class BrainDatabaseWindowedBucketsTests: XCTestCase { ) } + private func insertWrite( + id: String, + source: String, + localCreatedAtText: String + ) throws { + try sqliteExecWriteLocal( + path: tempDBPath, + sql: """ + INSERT INTO chunks (id, content, source, created_at, enriched_at, enrich_status, status) + VALUES ('\(id)', 'local wall-clock bucket probe \(id)', '\(source)', '\(localCreatedAtText)', NULL, NULL, 'active'); + """ + ) + } + func testWiderWindowReturnsMoreHistoricalDataThanLiveWindow() throws { // Force the schema to exist (constructor opens + ensures schema, but make // it explicit so an empty-table call cannot be misread). @@ -195,6 +210,41 @@ final class BrainDatabaseWindowedBucketsTests: XCTestCase { XCTAssertEqual(buckets.allWriteTotal - buckets.agentTotal - buckets.watcherTotal, 2) } + func testAgentWindowBucketsCountNaiveLocalWallClockTimestamp() throws { + XCTAssertTrue(try db.tableExists("chunks")) + + let previousTimeZone = getenv("TZ").map { String(cString: $0) } + XCTAssertEqual(setenv("TZ", "Asia/Jerusalem", 1), 0) + tzset() + defer { + if let previousTimeZone { + setenv("TZ", previousTimeZone, 1) + } else { + unsetenv("TZ") + } + tzset() + } + + let now = Date(timeIntervalSince1970: 1_784_031_600) + let formatter = DateFormatter() + formatter.locale = Locale(identifier: "en_US_POSIX") + formatter.timeZone = TimeZone(identifier: "Asia/Jerusalem") + formatter.dateFormat = "yyyy-MM-dd'T'HH:mm:ss.SSSSSS" + try insertWrite( + id: "agent-local-wall-clock", + source: "mcp", + localCreatedAtText: formatter.string(from: now.addingTimeInterval(-30)) + ) + + let buckets = try db.pipelineWindowBuckets(activityWindowMinutes: 30, bucketCount: 6, now: now) + + XCTAssertEqual( + buckets.agentWriteBuckets, + [0, 0, 0, 0, 0, 1], + "Agent-store metrics must interpret naive local T timestamps as local wall-clock time." + ) + } + func testWatcherWindowBucketsUseIngestionLivenessInsteadOfTranscriptCreatedAt() throws { XCTAssertTrue(try db.tableExists("chunks")) diff --git a/docs/plans/2026-07-14-brainbar-truth-rewind.md b/docs/plans/2026-07-14-brainbar-truth-rewind.md new file mode 100644 index 00000000..fed7af53 --- /dev/null +++ b/docs/plans/2026-07-14-brainbar-truth-rewind.md @@ -0,0 +1,104 @@ +# BrainBar Truth and Rewind Archive Implementation Plan + +> **For Claude:** REQUIRED SUB-SKILL: Use superpowers:executing-plans to implement this plan task-by-task. + +**Goal:** Preserve truthful BrainBar ingest metrics for local wall-clock timestamps and make watcher rewind archival commit safely through APSW. + +**Architecture:** Keep BrainBar's existing normalized timestamp and watcher-liveness paths; add focused coverage that exercises a local wall-clock agent-store fixture instead of duplicating the already-landed truth fix. Change the rewind archive batcher to own an explicit APSW transaction (`BEGIN IMMEDIATE` / `COMMIT`, with `ROLLBACK` on failure), matching the repository's write convention. + +**Tech Stack:** Swift/XCTest/SQLite3 for BrainBar dashboard coverage; Python/pytest/APSW for watcher archival. + +--- + +### Task 1: Lock the existing local-time ingest truth + +**Files:** +- Modify: `brain-bar/Tests/BrainBarTests/BrainDatabaseWindowedBucketsTests.swift` + +**Step 1: Add the local wall-clock fixture test** + +Add a test that inserts an agent-source chunk with a naive local `created_at` string and asserts `pipelineWindowBuckets` counts it in the current agent-store bucket. + +**Step 2: Run the focused test** + +Run: `swift test --filter BrainDatabaseWindowedBucketsTests` + +Expected: PASS because current `normalizedUnixEpochSQL` already interprets naive `T` timestamps as local wall time. This is regression coverage for already-landed source behavior, not a reason to manufacture another production change. + +**Step 3: Record truth-source evidence** + +Document in the PR that the ad-hoc `created_at >= datetime('now', '-10 minutes')` comparison is invalid for mixed timestamp formats, while BrainBar uses normalized epochs plus `watcher_liveness_events`. Document that the installed 1.4.2 app at commit `a2fd98d1` predates the source truth fix and needs a BrainBar restart after a later build is installed. + +### Task 2: Reproduce and fix APSW rewind archival + +**Files:** +- Modify: `tests/test_rewind_batch_archival.py` +- Modify: `src/brainlayer/cli/__init__.py` + +**Step 1: Write the real APSW regression** + +Add a test that uses `_RewindArchiveBatcher` with its default `VectorStore`, archives one rewound session, closes the store, and verifies the two realtime watcher rows are durable. + +**Step 2: Run the test to verify RED** + +Run: `PYTHONPATH=src /Users/etanheyman/Gits/brainlayer/.venv/bin/python -m pytest tests/test_rewind_batch_archival.py::test_rewind_archiver_commits_with_real_apsw_connection -vv` + +Expected: FAIL with `AttributeError: 'apsw.Connection' object has no attribute 'commit'`. + +**Step 3: Implement the minimal APSW transaction** + +In `_RewindArchiveBatcher.flush`, execute `BEGIN IMMEDIATE` before the archive update, execute `COMMIT` after reading `changes()`, and attempt `ROLLBACK` when the transaction was started and any operation fails. Preserve telemetry outcome and pending-session retry semantics. + +**Step 4: Run the regression to verify GREEN** + +Run the same focused pytest command. + +Expected: PASS. + +**Step 5: Run the rewind archive module** + +Run: `PYTHONPATH=src /Users/etanheyman/Gits/brainlayer/.venv/bin/python -m pytest tests/test_rewind_batch_archival.py -vv` + +Expected: all tests PASS. + +### Task 3: Verify and publish an open PR + +**Files:** +- Modify: `/Users/etanheyman/Gits/orchestrator/docs.local/collab/driver-buddy-2026-07-12.md` after PR completion + +**Step 1: Run scoped quality gates** + +Run: + +```bash +swift test --package-path brain-bar --filter BrainDatabaseWindowedBucketsTests +PYTHONPATH=src /Users/etanheyman/Gits/brainlayer/.venv/bin/python -m pytest tests/test_rewind_batch_archival.py -vv +/Users/etanheyman/Gits/brainlayer/.venv/bin/ruff check src/brainlayer/cli/__init__.py tests/test_rewind_batch_archival.py +git diff --check +``` + +Expected: all commands exit 0. + +**Step 2: Run repository-wide gates** + +Run: `ulimit -n 4096 && PYTHONPATH=src /Users/etanheyman/Gits/brainlayer/.venv/bin/python -m pytest` + +Run: `swift test --package-path brain-bar` + +Expected: both suites complete with zero failures. + +**Step 3: Review, commit, and push** + +Run the bounded local CodeRabbit review. Fix any real issues, then create focused commits and push `fix/brainbar-truth-rewind`. + +**Step 4: Open but never merge the PR** + +Open a ready-for-review PR and request `@codex`, `@cursor`, and `@bugbot` review. The PR must remain OPEN. Do not run `gh pr merge`. + +PR deploy note: restart `com.brainlayer.watch` for the Python rewind fix; rebuild/install and restart BrainBar for the dashboard truth code already present on current source. + +Blocker: BrainLayer has no remote `develop` branch and no `feature/854-ux-batch-integration` branch. Resolve the requested base before PR creation; do not invent or publish a shared base branch. + +**Step 5: Post the builder receipt** + +Under the collab file's builders section, append exactly: `DONE: SLICE_ PR: ` once the orchestrator supplies the slice/card mapping and the open PR URL exists. diff --git a/src/brainlayer/cli/__init__.py b/src/brainlayer/cli/__init__.py index a171e155..4af6abc7 100644 --- a/src/brainlayer/cli/__init__.py +++ b/src/brainlayer/cli/__init__.py @@ -3310,7 +3310,10 @@ def flush(self, reason: str) -> int: "sessions_planned": len(self._pending_session_ids), }, ) + transaction_started = False try: + cursor.execute("BEGIN IMMEDIATE") + transaction_started = True cursor.execute( f""" UPDATE chunks @@ -3322,8 +3325,14 @@ def flush(self, reason: str) -> int: params, ) affected = vector_store.conn.changes() - vector_store.conn.commit() + cursor.execute("COMMIT") + transaction_started = False except Exception as exc: + if transaction_started: + try: + cursor.execute("ROLLBACK") + except Exception: + pass telemetry_span.finish("error", error=f"{type(exc).__name__}: {exc}") raise telemetry_span.finish("commit", rows_touched=affected) diff --git a/tests/test_rewind_batch_archival.py b/tests/test_rewind_batch_archival.py index 76117bc3..d9d7adf0 100644 --- a/tests/test_rewind_batch_archival.py +++ b/tests/test_rewind_batch_archival.py @@ -1,6 +1,10 @@ +import json import sqlite3 import time +import apsw +import pytest + from brainlayer.cli import _RewindArchiveBatcher from brainlayer.vector_store import VectorStore @@ -94,6 +98,86 @@ def _prepare_rewind_archive_db(tmp_path): return db_path +def test_rewind_archiver_commits_with_real_apsw_connection(tmp_path): + db_path = _prepare_rewind_archive_db(tmp_path) + archiver = _RewindArchiveBatcher( + db_path=db_path, + batch_size=1, + flush_interval_ms=60_000, + ) + + try: + archiver.add("s1") + assert archiver.flush("rewind") == 2 + finally: + archiver.close() + + with sqlite3.connect(db_path) as conn: + archived = conn.execute( + """ + SELECT COUNT(*) + FROM chunks + WHERE conversation_id = 's1' + AND source = 'realtime_watcher' + AND archived_at IS NOT NULL + AND value_type = 'ARCHIVED' + """ + ).fetchone()[0] + + assert archived == 2 + + +def test_rewind_archiver_rolls_back_and_preserves_pending_retry(tmp_path, monkeypatch): + db_path = _prepare_rewind_archive_db(tmp_path) + telemetry_path = tmp_path / "writer-telemetry.jsonl" + monkeypatch.setenv("BRAINLAYER_WRITER_TELEMETRY", "1") + monkeypatch.setenv("BRAINLAYER_WRITER_TELEMETRY_PATH", str(telemetry_path)) + + with VectorStore(db_path) as store: + store.conn.execute( + """ + CREATE TRIGGER fail_rewind_archive + BEFORE UPDATE OF archived_at ON chunks + WHEN NEW.archived_at IS NOT NULL + BEGIN + SELECT RAISE(ABORT, 'forced rewind archive failure'); + END + """ + ) + + archiver = _RewindArchiveBatcher( + db_path=db_path, + batch_size=1, + flush_interval_ms=60_000, + ) + archiver.add("s1") + + try: + with pytest.raises(apsw.ConstraintError, match="forced rewind archive failure"): + archiver.flush("rewind") + + assert archiver._get_vector_store().conn.getautocommit() is True + assert archiver.pending_count == 1 + with sqlite3.connect(db_path) as conn: + archived = conn.execute( + "SELECT COUNT(*) FROM chunks WHERE conversation_id = 's1' AND archived_at IS NOT NULL" + ).fetchone()[0] + assert archived == 0 + + finished_events = [ + event + for event in (json.loads(line) for line in telemetry_path.read_text().splitlines()) + if event.get("event") == "txn_finished" and event.get("operation") == "rewind_archive" + ] + assert finished_events[-1]["outcome"] == "error" + + archiver._get_vector_store().conn.execute("DROP TRIGGER fail_rewind_archive") + assert archiver.flush("retry") == 2 + assert archiver.pending_count == 0 + finally: + archiver.close() + + def test_rewind_archiver_batches_multiple_sessions(tmp_path): db_path = _prepare_rewind_archive_db(tmp_path) counters = {"opens": 0, "updates": 0}