fix: make watcher rewind archival APSW-safe

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"))
 

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_<X> <card> PR: <url>` once the orchestrator supplies the slice/card mapping and the open PR URL exists.

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)

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}