board-v2-parquet-sync

Answer

Root cause: migrate-board-to-duckdb.py exports tasks.parquet once as a snapshot. Ticks that mutate only tasks.md update the DuckDB tables but never re-export, so the parquet lags the matrix. Fix: make the export part of the ingest pipeline — re-run both COPY statements after any task status change, using OVERWRITE_OR_IGNORE true. Also fix two latent bugs: board.last_tick is a TIMESTAMP (handle/compare it as such, never int), and events.id has no sequence — inserts must use COALESCE(MAX(id),0)+1.

1. Re-export after every status change (the core fix):

def sync_parquet(con: duckdb.DuckDBPyConnection, board: str) -> None:
    """Re-run the COPY exports after ANY mutation. OVERWRITE_OR_IGNORE=true."""
    board_dir = os.path.join(BOARD_DIR, board)
    os.makedirs(board_dir, exist_ok=True)
    con.execute(
        f"COPY (SELECT * FROM tasks ORDER BY id) "
        f"TO '{board_dir}/tasks.parquet' (FORMAT PARQUET, OVERWRITE_OR_IGNORE true)"
    )
    con.execute(
        f"COPY (SELECT * FROM events ORDER BY id) "
        f"TO '{board_dir}/events.parquet' (FORMAT PARQUET, OVERWRITE_OR_IGNORE true)"
    )

Call it at the end of every ingest, unconditionally — not only on first creation:

def ingest(con, board: str, tick: dt.datetime) -> tuple[int, int]:
    parsed = parse_tasks(os.path.join(BOARD_DIR, board, "tasks.md"))
    now = tick
    con.execute("INSERT OR REPLACE INTO board VALUES (?, ?)", [board, now])
    n_tasks = n_events = 0
    for t in parsed:
        row = con.execute("SELECT status FROM tasks WHERE id = ?", [t["id"]]).fetchone()
        old = row[0] if row else None
        if old is None:
            con.execute("INSERT INTO tasks (id, status, line, updated) VALUES (?,?,?,?)",
                        [t["id"], t["status"], t["line"], now]); n_tasks += 1
        elif old != t["status"]:
            con.execute("UPDATE tasks SET status=?, line=?, updated=? WHERE id=?",
                        [t["status"], t["line"], now, t["id"]]); n_tasks += 1
        else:
            continue  # idempotent tick -> no event, no export churn
        # no live sequence on events.id -> explicit COALESCE(MAX(id),0)+1
        nxt = con.execute("SELECT COALESCE(MAX(id), 0) + 1 FROM events").fetchone()[0]
        con.execute(
            "INSERT INTO events (id, task_id, from_status, to_status, occurred_at) "
            "VALUES (?, ?, ?, ?, ?)", [nxt, t["id"], old, t["status"], now])
        n_events += 1
    sync_parquet(con, board)   # <-- FIX: parquet is a live view, not a snapshot
    return n_tasks, n_events

2. board.last_tick is TIMESTAMP, not int: store and compare ISO-8601 ticks as datetime; never do int arithmetic on the column. Stale detection is a timestamp gap:

# schema:  last_tick TIMESTAMP NOT NULL
tick = dt.datetime.fromisoformat(args.tick) if args.tick else dt.datetime.now(...)
stale = board_last_tick < (now - dt.timedelta(seconds=stale_after_s))

3. events.id sequence: COALESCE(MAX(id),0)+1 covers both the empty-table case (MAX → NULL → 0 → id 1) and post-deletion gaps (MAX(id)+1 never collides).