A twice-daily batch that only ran once — reconstructing run counts from artifacts

"""Decide whether effective throughput is falling short of plan, using backlog alone."""
 
def drain_rate(backlog: list[int]) -> float:
    """Average decrease per day. Positive means draining, negative means piling up."""
    deltas = [backlog[i] - backlog[i + 1] for i in range(len(backlog) - 1)]
    return sum(deltas) / len(deltas)
 
 
def days_to_empty(backlog: list[int]) -> float | None:
    rate = drain_rate(backlog)
    return None if rate <= 0 else backlog[-1] / rate
 
 
def plan_ratio(planned_per_run: int, runs_per_day: int, backlog: list[int]) -> dict:
    """Observed throughput over planned. A value near 0.5 points at missing runs."""
    planned = planned_per_run * runs_per_day
    observed = drain_rate(backlog)
    return {
        "planned_per_day": planned,
        "observed_per_day": round(observed, 1),
        "ratio": round(observed / planned, 2) if planned else None,
    }
 
 
if __name__ == "__main__":
    # Eight days of backlog, sampled every morning at the same time, no new inflow
    backlog = [1180, 1108, 1036, 962, 890, 818, 744, 672]
    print("drain per day:", round(drain_rate(backlog), 1))
    print("days to empty:", round(days_to_empty(backlog), 1))
    print(plan_ratio(planned_per_run=72, runs_per_day=2, backlog=backlog))

drain per day: 72.6
days to empty: 9.3
{'planned_per_day': 144, 'observed_per_day': 72.6, 'ratio': 0.5}

"""Detect missing run slots by working backwards from artifacts."""
from __future__ import annotations
from dataclasses import dataclass
from datetime import datetime, timedelta
from zoneinfo import ZoneInfo
 
JST = ZoneInfo("Asia/Tokyo")
 
 
@dataclass(frozen=True)
class Slot:
    day: str        # "2026-08-26"
    label: str      # "04:30"
    start: datetime
    end: datetime   # deadline including grace
 
 
def expected_slots(days: list[str], times: list[str], grace_min: int = 90) -> list[Slot]:
    """Turn a schedule into acceptance windows with grace."""
    out = []
    for d in days:
        y, m, dd = map(int, d.split("-"))
        for t in times:
            hh, mm = map(int, t.split(":"))
            start = datetime(y, m, dd, hh, mm, tzinfo=JST)
            out.append(Slot(d, t, start, start + timedelta(minutes=grace_min)))
    return sorted(out, key=lambda s: s.start)
 
 
def observed_runs(records: list[dict]):
    """Pull run_id and completion time out of artifacts as evidence a run occurred."""
    seen = set()
    for r in records:
        rid = r.get("run_id")
        if not rid or rid in seen:
            continue
        seen.add(rid)
        yield datetime.fromisoformat(r["finished_at"]).astimezone(JST)
 
 
def reconcile(slots: list[Slot], runs: list[datetime]) -> dict:
    """Assign evidence to windows, keeping leftovers on both sides separate."""
    remaining = sorted(runs)
    missed, matched = [], []
    for s in slots:
        hit = next((r for r in remaining if s.start <= r <= s.end), None)
        if hit is None:
            missed.append(s)
        else:
            matched.append((s, hit))
            remaining.remove(hit)
    return {"missed": missed, "matched": matched, "orphans": remaining}
 
 
if __name__ == "__main__":
    days = [f"2026-08-{d:02d}" for d in range(20, 28)]
    slots = expected_slots(days, ["04:30", "16:30"])
 
    # Completion times recovered from artifacts — only the 04:30 slot was represented
    records = [
        {"run_id": f"r{i}", "finished_at": f"{d}T04:52:00+09:00"}
        for i, d in enumerate(days)
    ]
 
    res = reconcile(slots, list(observed_runs(records)))
    print(f"expected: {len(slots)}  observed: {len(res['matched'])}  missed: {len(res['missed'])}")
 
    by_label: dict[str, int] = {}
    for s in res["missed"]:
        by_label[s.label] = by_label.get(s.label, 0) + 1
    for label, n in sorted(by_label.items()):
        print(f"  missing slot {label}: {n}/{len(days)} days")

expected: 16  observed: 8  missed: 8
  missing slot 16:30: 8/8 days

"""Record the fact that a run started, decoupled from what it processed."""
import sqlite3
import uuid
from datetime import datetime, timezone
 
DDL = """
CREATE TABLE IF NOT EXISTS run_heartbeat (
    run_id      TEXT PRIMARY KEY,
    slot_label  TEXT NOT NULL,     -- "04:30", matching the schedule
    started_at  TEXT NOT NULL,
    finished_at TEXT,
    item_count  INTEGER            -- zero is a legitimate recorded value
);
"""
 
 
def open_ledger(path: str = "runs.db") -> sqlite3.Connection:
    conn = sqlite3.connect(path)
    conn.execute(DDL)
    conn.commit()
    return conn
 
 
def begin_run(conn: sqlite3.Connection, slot_label: str) -> str:
    """Call this before the Gemini request. Write it later and the blind spot returns."""
    run_id = uuid.uuid4().hex
    conn.execute(
        "INSERT INTO run_heartbeat (run_id, slot_label, started_at) VALUES (?, ?, ?)",
        (run_id, slot_label, datetime.now(timezone.utc).isoformat()),
    )
    conn.commit()
    return run_id
 
 
def end_run(conn: sqlite3.Connection, run_id: str, item_count: int) -> None:
    conn.execute(
        "UPDATE run_heartbeat SET finished_at = ?, item_count = ? WHERE run_id = ?",
        (datetime.now(timezone.utc).isoformat(), item_count, run_id),
    )
    conn.commit()
 
 
def missing_slots(conn: sqlite3.Connection, day: str, expected: list[str]) -> list[str]:
    rows = conn.execute(
        "SELECT slot_label FROM run_heartbeat WHERE started_at LIKE ? || '%'", (day,)
    ).fetchall()
    seen = {r[0] for r in rows}
    return [s for s in expected if s not in seen]