python-board-duckdb-audit-tick

Answer

The canonical DuckDB board write pattern for the idle audit tick. One connection per tick, single transaction: INSERT event (next id computed inline via COALESCE(MAX(id),0)+1), header UPDATE (single-row upsert), COPY parquet (overwrite), then the integrity gate — COUNT(*) == COUNT(DISTINCT id) on both board.db and the parquet, plus a board↔parquet sync check. Finally git add -f with explicit parquet paths, never the directory.

# board_writer.py
SCHEMA = """
CREATE TABLE IF NOT EXISTS header (
    id            INTEGER PRIMARY KEY CHECK (id = 1),
    last_tick_id  BIGINT  NOT NULL DEFAULT 0,
    last_run_at   TIMESTAMP
);
CREATE TABLE IF NOT EXISTS board (
    id         BIGINT PRIMARY KEY,          -- duplicate ids impossible at schema level
    tick       INTEGER NOT NULL,
    event      VARCHAR NOT NULL,
    payload    JSON,
    written_at TIMESTAMP DEFAULT current_timestamp
);
"""

def run_tick(board_path: str, parquet_path: str, tick: int, event: str):
    con = duckdb.connect(board_path)
    try:
        con.execute(SCHEMA)
        con.execute("BEGIN")

        # 1) INSERT event — COALESCE(MAX(id),0)+1 lives INSIDE the statement,
        #    so an empty board (MAX = NULL) still yields id=1, and each tick
        #    reads the committed max so ids are never reused.
        new_id = int(con.execute("""
            INSERT INTO board (id, tick, event, payload)
            VALUES (COALESCE((SELECT MAX(id) FROM board), 0) + 1, ?, ?, ?)
            RETURNING id
        """, [tick, event, {"source": "idle-audit", "tick": tick}]).fetchone()[0])

        # 2) header UPDATE — single-row upsert records the new high-water id
        con.execute("""
            INSERT INTO header (id, last_tick_id, last_run_at)
            VALUES (1, ?, current_timestamp)
            ON CONFLICT (id) DO UPDATE
                SET last_tick_id = excluded.last_tick_id,
                    last_run_at  = excluded.last_run_at
        """, [new_id])

        # 3) COPY parquet (overwrite semantics) inside the same transaction
        con.execute(f"COPY board TO '{parquet_path}' (FORMAT PARQUET, OVERWRITE_OR_IGNORE)")
        con.execute("COMMIT")

        # 4) Integrity gate on BOTH artifacts
        verify_integrity(con, parquet_path)
        return new_id
    finally:
        con.close()

def verify_integrity(con, parquet_path):
    """COUNT(*) == COUNT(DISTINCT id) on board.db AND parquet; plus sync."""
    board_rows, board_ids  = con.execute("SELECT COUNT(*), COUNT(DISTINCT id) FROM board").fetchone()
    parq_rows, parq_ids    = con.execute(
        f"SELECT COUNT(*), COUNT(DISTINCT id) FROM read_parquet('{parquet_path}')").fetchone()
    board_fp  = con.execute("SELECT id FROM board ORDER BY id").fetchall()
    parq_fp   = con.execute(f"SELECT id FROM read_parquet('{parquet_path}') ORDER BY id").fetchall()
    problems = []
    if board_rows != board_ids: problems.append(f"board.db: {board_rows} rows vs {board_ids} distinct")
    if parq_rows  != parq_ids:  problems.append(f"parquet:  {parq_rows} rows vs {parq_ids} distinct")
    if board_rows != parq_rows: problems.append(f"row count drift: board={board_rows} parquet={parq_rows}")
    if board_fp != parq_fp:     problems.append("id multiset drift: board.db != parquet")
    if problems:
        raise AssertionError("integrity gate failed: " + "; ".join(problems))

The gate is also callable standalone — a stale/corrupt parquet is caught before the next tick overwrites it.

git add — explicit paths only (.gitignore has *.parquet, so -f is required; the directory must never be staged):

git add -f -- parquet/board.parquet        # OK: explicit .parquet path
git add -f parquet/                        # NEVER: stages the dir / ignored junk

Enforced in code — any non-.parquet path (including the directory) is rejected:

def git_add_parquet(repo_root, parquet_paths):
    for p in parquet_paths:
        assert os.path.isfile(p), f"parquet artifact missing: {p}"
        assert os.path.splitext(p)[1].lower() == ".parquet", f"not a parquet path: {p}"
        subprocess.run(["git", "add", "-f", "--", os.path.relpath(p, repo_root)],
                       cwd=repo_root, check=True)

Harness contract: development/aws-shape-validator.py --all (76/76) after the tick, everything wrapped in timeout 540 gitreins guard so a hung tick aborts and the guard enforces the explicit-path add rule.