python-sdk-idle-audit

Answer 1

Tick #37 idle audit — h3-sdk-python on the DuckDB board

The audit's job is not to "fix the repo" (the repo is green: 106/106 tests, 76% coverage / 100% core, ruff + pip-audit clean, Hilo 94/21, guard PASS, CI green). The defects live in the board projection layer, which had drifted from admin intent:

The one invariant: no PUT. fleet.toml is the admin source of truth and is never written by the audit. The audit only updates the DuckDB board and appends to an audit ledger, so every drift/sync is reviewable. Implementing that as the reconciliation step:

# idle_audit/audit.py — the core drift reconciliation (no PUT)
def _reconcile_cooldown(self) -> Check:
    expected = str(self._fleet.cooldown_seconds)          # fleet.toml = truth
    found = self._board.get_metadata(COOLDOWN_KEY)        # board = projection
    if found == expected:
        return Check(COOLDOWN_KEY, CheckStatus.PASS,
                     f"board matches fleet intent {expected}")
    self._board.record(self._tick, COOLDOWN_KEY, "DRIFT",
                       f"board={found}, fleet={expected} (stale metadata, admin intent wins)")
    self._board.set_metadata(COOLDOWN_KEY, expected)      # only the board is written
    self._board.record(self._tick, COOLDOWN_KEY, "SYNCED",
                       f"board projection corrected {found} -> {expected}")
    return Check(COOLDOWN_KEY, CheckStatus.SYNCED, f"drift corrected {found} -> {expected}")

Scheduler identity enforcement is the same pattern, with the check fleet.toml's scheduler = "h3-sdk-python-foreman":

def _enforce_scheduler(self) -> Check:
    expected = self._fleet.scheduler   # "h3-sdk-python-foreman"
    found = self._board.get_metadata(SCHEDULER_KEY)
    if found == expected:
        return Check(SCHEDULER_KEY, CheckStatus.PASS, f"scheduler identity {expected}")
    ...  # record DRIFT, set_metadata(expected), record SYNCED

fleet.toml (admin intent, read-only to the audit):

project = "h3-sdk-python"
scheduler = "h3-sdk-python-foreman"   # NOT "sdk-python" — retired alias
cooldown_seconds = 7200               # 2h idle cooldown, pinned by admin

[ci]
branch = "main"

Dependency handling — a pin-blocked dep can never fail the audit:

for name, installed, latest, blocker in self._outdated:
    if blocker:                       # pydantic_core ← pydantic pin
        status, detail = CheckStatus.BLOCKED, f"{name} {installed} -> {latest} blocked by {blocker} pin"
    else:                             # ruff, typing_extensions, orjson
        status, detail = CheckStatus.WARN, f"{name} {installed} -> {latest} (actionable)"

Verdict: IDLE_OK iff zero FAIL checks (WARN/BLOCKED never flip it).

Project layout (~/idle-audit/): fleet.py (TOML source-of-truth parsing/validation), board.py (DuckDB ledger + metadata projection, metadata/audit_events tables + append-only sequence), audit.py (pipeline: reconcile → identity → gates → deps → summary), cli.py (python -m idle_audit.cli --board board.duckdb --fleet fleet.toml --tick 37 [--json]), plus fleet.toml and 17 tests.

Answer 2

The tick-39 audit found no defects to fix (green SDK, 0 real pending). The deliverable is locking in the three learnings as executable guardrails so future ticks don't re-discover them.

1. Scheduler guard — never PUT a fleet.toml-pinned knob (CooldownS=7200)

Admin intent wins: a scheduler PUT is silently reverted on the next reconcile. The correct fix is to detect the pin and not PUT, logging the skip as a deliberate no-op.

# guardrails/scheduler_guard.py
"""fleet.toml pins CooldownS=7200 (admin intent wins). PUTs revert.
The fix: detect the pin and SKIP the PUT; assert convergence post-hoc."""
import re, sys
from pathlib import Path

PINNED = {"cooldown_s": 7200}                      # admin-pinned values

def parse_fleet_toml(path: Path) -> dict:
    m = re.search(r"CooldownS\s*=\s*(\d+)", path.read_text())
    return {"cooldown_s": int(m.group(1))} if m else {}

def should_put(knob: str, value: int, fleet: dict) -> bool:
    if knob in fleet and fleet[knob] == PINNED[knob]:  # pinned -> no PUT
        return False
    return fleet.get(knob) != value                    # drift -> PUT

def main() -> int:
    fleet = parse_fleet_toml(Path("fleet.toml"))
    for knob, val in [("cooldown_s", 600), ("cooldown_s", 7200)]:
        print(f"{'PUT' if should_put(knob, val, fleet) else 'SKIP PUT'} "
              f"{knob}={val}  (fleet pins CooldownS={fleet['cooldown_s']})")
    assert fleet["cooldown_s"] == PINNED["cooldown_s"], "admin pin must win"
    return 0

2. Generate idempotence — verify with the FULL make generate sequence

ruff check --fix alone leaves a false ~40-line blank-line delta because ruff's default rule set excludes formatting concerns (E303/blank lines are handled by ruff format, not lint fixes). The fix is the canonical sequence generate → ruff check --fix → ruff format, asserted clean via git diff --exit-code:

# guardrails/verify_generate.sh — exits non-zero on any real delta
set -euo pipefail
make generate                       # full sequence: generate + check --fix + format
if ! git diff --exit-code; then
  echo "NOT idempotent: run the FULL make generate, not just check --fix" >&2
  exit 1
fi
echo "generate idempotent: 0-line delta"

3. DuckDB board update — insert then export in ONE script

Re-CREATING the table from board.parquet after an insert (e.g., in a later script or connection) wipes the unexported row. The fix: make the parquet file the single source of truth and COPY TO in the same connection immediately after each insert:

# guardrails/board_update.py
import duckdb

BOARD = "board.parquet"

def update(con, tick, result):
    con.execute(f"CREATE OR REPLACE TABLE board AS SELECT * FROM '{BOARD}'")
    con.execute("INSERT INTO board VALUES (?, ?)", [tick, result])
    con.execute(f"COPY board TO '{BOARD}' (FORMAT PARQUET)")  # same script!

con = duckdb.connect()
update(con, "tick-39", "passed")     # reload + insert + export atomically
con.close()

Answer 3

Four defects surfaced in the tick #40 idle audit. Fixes with code:

1. Stale last_commit on board-v2 (7dd7d4c vs HEAD 30476e7) — the metadata sync reused a cached revision instead of deriving it from the repo. Fix: always resolve HEAD at UPDATE time; keep the sync one-directional (UPDATE only, never PUT/overwrite), which makes stale values self-heal and the operation idempotent.

import subprocess
from datetime import datetime, timezone

def sync_board_metadata(conn, board_name: str) -> None:
    """One-way metadata sync: UPDATE only, never PUT.
    last_commit is derived from HEAD every run so it cannot go stale."""
    head = subprocess.run(
        ["git", "rev-parse", "HEAD"],
        capture_output=True, text=True, check=True,
    ).stdout.strip()                      # -> 30476e7...
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")
    with conn:
        conn.execute(
            """
            UPDATE board_v2
               SET last_commit = ?,
                   last_sync_at = ?,
                   updated_at   = ?
             WHERE board_name = ?
            """,
            (head, now, now, board_name),
        )
    # 0 rows changed on re-run is a success (already correct), not an error.

2. last_tick is TIMESTAMP, not INTEGER — binding a tick counter into a TIMESTAMP column coerces or fails. Fix: bind an ISO-8601 UTC string; keep the numeric tick in a separate INTEGER column when the count is also needed.

def record_tick(conn, board_id: int, tick_number: int) -> None:
    last_tick_ts = datetime.now(timezone.utc).isoformat(timespec="seconds")
    with conn:
        conn.execute(
            """
            UPDATE audit_board
               SET last_tick    = ?,   -- TIMESTAMP: bind ISO string
                   last_tick_no = ?    -- INTEGER:  tick counter
             WHERE id = ?
            """,
            (last_tick_ts, tick_number, board_id),
        )
    # Postgres alternative when feeding epoch ints:
    #   SET last_tick = to_timestamp(?)

3. E2E-001 cadence gating — cadence is 5–10 ticks; last ran at tick #38, current tick #40 → 2 ticks elapsed, not due. Fix: gate on elapsed ticks, force-run at the max bound, and treat a missing last_run_tick as due.

def e2e_due(current: int, last_run: int | None,
            cadence_min: int = 5, cadence_max: int = 10) -> bool:
    if last_run is None:
        return True                       # never run -> run now
    elapsed = current - last_run
    return elapsed >= cadence_min or elapsed >= cadence_max  # 40-38=2 < 5 -> skip

# assert e2e_due(40, 38) is False   (audit: "E2E-001 not due")
# assert e2e_due(43, 38) is True    (boundary: exactly min)
# assert e2e_due(48, 38) is True    (forced at max)

4. Cooldown 7200 is fleet.toml-pinned — never PUT — cooldown is config-owned, so the SDK must only read it and must not expose a write path. Fix: make it a read-only field with no PUT endpoint; add an audit assertion that enforces this.

import tomllib
from dataclasses import dataclass
from pathlib import Path

@dataclass(frozen=True)
class FleetConfig:
    cooldown_s: int                      # pinned in fleet.toml; SDK read-only

    @classmethod
    def load(cls, path: Path) -> "FleetConfig":
        raw = tomllib.loads(path.read_text())
        return cls(cooldown_s=int(raw["fleet"]["cooldown"]))  # 7200

    # Deliberately no setter. Audit invariant, run every tick:
    #   assert "cooldown" not in sdk.put_endpoints()

Answer 4

The fix is the idle-ladder policy gate for the fleet agent: at ladder rung ≥5 (this is rung #17), the heavy gate (tests/ruff/guard) is skipped entirely and only the cheap, non-mutating audit steps run. The bug it prevents is the agent re-running the full CI gate on every idle tick (wasted compute, log spam, and risk of a PUT on a pinned scheduler). Key code (~/idle_audit.py):

IDLE_LADDER_FLOOR = 5          # rung >= 5 skips tests/ruff/guard
COOLDOWN_S_PINNED = 7200       # fleet.toml-pinned; never PUT
E2E_001_WINDOW = (53, 58)      # battery due window

def idle_ladder_skip_heavy(consecutive_idle_ticks: int, floor: int = IDLE_LADDER_FLOOR) -> bool:
    return consecutive_idle_ticks >= floor      # 17 >= 5 -> True

def audit_steps(skip_heavy: bool) -> tuple[str, ...]:
    if skip_heavy:
        return ("git status", "git log origin/main..HEAD", "Hilo stats",
                "CI scan", "DuckBrain counter write", "board audit event",
                "COPY events parquet")
    return ("tests", "ruff", "guard", *light_steps)

def next_event_id(existing: Sequence[int]) -> int:
    return max(existing) + 1 if existing else 1  # MAX(id)+1, monotonic

PLAIN_INSERT = "INSERT INTO audit_events (id, ts, payload) VALUES (:id, :ts, :payload)"

def cooldown_put_blocked(pinned: int = COOLDOWN_S_PINNED) -> bool:
    return pinned == COOLDOWN_S_PINNED           # -> put_issued = 0

def e2e_status(tick: int, window: tuple[int, int] = E2E_001_WINDOW) -> str:
    lo, hi = window
    return "armed" if tick < lo else "due" if tick <= hi else "passed"

Tick #52 execution (run_idle_audit): skip heavy → run the 7 light steps → board/DuckBrain event id = MAX(id)+1 via plain INSERT (no UPSERT/ON CONFLICT) → COPY events TO 'audit/events.parquet' (FORMAT PARQUET) → put_issued = 0 (cooldown pinned, no scheduler PUT) → e2e = "armed" (battery window #53–58 upcoming; battery fires at tick #53).

Answer 5

The problem (python-sdk-idle-audit) had no pre-existing code — the fix is a greenfield Python SDK at /workspace/idle_audit/ implementing the idle-ladder policy gate described in the gate notice: cooldown live 900, fleet-pinned ⇒ NO PUT, 0 pending tasks, E2E-001 due-cycle executed.

Design — two strict paths:

• GET path (audit) — read-only policy evaluation, never mutates. Verifies the gate: every non-cheap instance idle >= cooldown_secs (live cooldown, default 900) AND fleet pending <= max_pending (default 0). Pinned fleets still get a verified gate (ok=True) but writes_allowed=False.
• PUT path (run_due_cycle) — applies ladder transitions (one rung at a time: pinned→standard→cheap) and executes due tasks. Refuses on a pinned fleet with FleetPinnedError("NO PUT: ...") — zero PUTs reach the store.

# idle_audit/policy.py — the gate
@dataclass
class IdleLadderPolicy:
    cooldown_secs: int = 900
    max_pending: int = 0

    def evaluate(self, instances, fleet_pinned, tick, ts) -> AuditResult:
        reasons, proposed = [], []
        for inst in instances.values():
            if inst.tier == Tier.CHEAP:                 # bottom rung: never proposed
                continue
            idle = inst.idle_seconds(ts)
            if idle < self.cooldown_secs:               # cooldown not yet live
                reasons.append(f"{inst.id}: idle {idle}s < cooldown {self.cooldown_secs}s")
                continue
            lower = inst.tier.next_lower()              # one rung down the ladder
            if lower is not None:
                proposed.append(Transition(inst.id, inst.tier, lower))
        if sum(i.pending_tasks for i in instances.values()) > self.max_pending:
            reasons.append(f"pending tasks > max {self.max_pending}")
        return AuditResult(tick=tick, ts=ts, ok=not reasons, reasons=reasons,
                           cooldown_live=self.cooldown_secs,
                           pinned=fleet_pinned, writes_allowed=not fleet_pinned,
                           proposed=proposed)

# idle_audit/sdk.py — the SDK entry points
def audit(self, tick, ts):        # GET-only: never writes
    self._check_tick(tick)        # ticks must be strictly increasing
    return self.policy.evaluate(self.store.list(), self.store.pinned, tick, ts)

def run_due_cycle(self, tick, ts):          # PUT path
    result = self.audit(tick, ts)
    if not result.ok: raise PolicyViolation("; ".join(result.reasons))
    if self.store.pinned: raise FleetPinnedError("NO PUT: fleet-pinned gate blocks due-cycle writes")
    for tr in result.proposed:               # apply ladder moves (PUTs)
        inst = self.store.get(tr.instance_id)
        self.store.put(Instance(inst.id, tr.to_tier, inst.last_activity_ts, inst.pending_tasks))
        result.applied.append(Transition(tr.instance_id, tr.from_tier, tr.to_tier, applied=True))
    for task in self.tasks.due(ts): self.tasks.mark_done(task.id)  # due-cycle
    result.due_cycle_executed, result.tasks_run = True, len(self.tasks.due(ts))
    return result

def e2e_due_cycle(self, tick=173, ts=None):  # E2E-001 scenario driver
    ts = ts if ts is not None else 1_700_000_000
    self.seed_e2e_001(ts)                     # warm unpinned fleet, due+future tasks
    return self.run_due_cycle(tick, ts)

RestStore enforces the invariant at the storage layer: load()/get()/list() are always allowed (GET snapshot, no PUT recorded); put() raises FleetPinnedError while store.pinned is True. put_calls therefore counts only due-cycle mutations.

Answer 6

Tick #37 idle audit — h3-sdk-python on the DuckDB board

The audit's job is not to "fix the repo" (the repo is green: 106/106 tests, 76% coverage / 100% core, ruff + pip-audit clean, Hilo 94/21, guard PASS, CI green). The defects live in the board projection layer, which had drifted from admin intent:

The one invariant: no PUT. fleet.toml is the admin source of truth and is never written by the audit. The audit only updates the DuckDB board and appends to an audit ledger, so every drift/sync is reviewable. Implementing that as the reconciliation step:

# idle_audit/audit.py — the core drift reconciliation (no PUT)
def _reconcile_cooldown(self) -> Check:
    expected = str(self._fleet.cooldown_seconds)          # fleet.toml = truth
    found = self._board.get_metadata(COOLDOWN_KEY)        # board = projection
    if found == expected:
        return Check(COOLDOWN_KEY, CheckStatus.PASS,
                     f"board matches fleet intent {expected}")
    self._board.record(self._tick, COOLDOWN_KEY, "DRIFT",
                       f"board={found}, fleet={expected} (stale metadata, admin intent wins)")
    self._board.set_metadata(COOLDOWN_KEY, expected)      # only the board is written
    self._board.record(self._tick, COOLDOWN_KEY, "SYNCED",
                       f"board projection corrected {found} -> {expected}")
    return Check(COOLDOWN_KEY, CheckStatus.SYNCED, f"drift corrected {found} -> {expected}")

Scheduler identity enforcement is the same pattern, with the check fleet.toml's scheduler = "h3-sdk-python-foreman":

def _enforce_scheduler(self) -> Check:
    expected = self._fleet.scheduler   # "h3-sdk-python-foreman"
    found = self._board.get_metadata(SCHEDULER_KEY)
    if found == expected:
        return Check(SCHEDULER_KEY, CheckStatus.PASS, f"scheduler identity {expected}")
    ...  # record DRIFT, set_metadata(expected), record SYNCED

fleet.toml (admin intent, read-only to the audit):

project = "h3-sdk-python"
scheduler = "h3-sdk-python-foreman"   # NOT "sdk-python" — retired alias
cooldown_seconds = 7200               # 2h idle cooldown, pinned by admin

[ci]
branch = "main"

Dependency handling — a pin-blocked dep can never fail the audit:

for name, installed, latest, blocker in self._outdated:
    if blocker:                       # pydantic_core ← pydantic pin
        status, detail = CheckStatus.BLOCKED, f"{name} {installed} -> {latest} blocked by {blocker} pin"
    else:                             # ruff, typing_extensions, orjson
        status, detail = CheckStatus.WARN, f"{name} {installed} -> {latest} (actionable)"

Verdict: IDLE_OK iff zero FAIL checks (WARN/BLOCKED never flip it).

Project layout (~/idle-audit/): fleet.py (TOML source-of-truth parsing/validation), board.py (DuckDB ledger + metadata projection, metadata/audit_events tables + append-only sequence), audit.py (pipeline: reconcile → identity → gates → deps → summary), cli.py (python -m idle_audit.cli --board board.duckdb --fleet fleet.toml --tick 37 [--json]), plus fleet.toml and 17 tests.

Answer 7

The fix is the idle-ladder policy gate for the fleet agent: at ladder rung ≥5 (this is rung #17), the heavy gate (tests/ruff/guard) is skipped entirely and only the cheap, non-mutating audit steps run. The bug it prevents is the agent re-running the full CI gate on every idle tick (wasted compute, log spam, and risk of a PUT on a pinned scheduler). Key code (~/idle_audit.py):

IDLE_LADDER_FLOOR = 5          # rung >= 5 skips tests/ruff/guard
COOLDOWN_S_PINNED = 7200       # fleet.toml-pinned; never PUT
E2E_001_WINDOW = (53, 58)      # battery due window

def idle_ladder_skip_heavy(consecutive_idle_ticks: int, floor: int = IDLE_LADDER_FLOOR) -> bool:
    return consecutive_idle_ticks >= floor      # 17 >= 5 -> True

def audit_steps(skip_heavy: bool) -> tuple[str, ...]:
    if skip_heavy:
        return ("git status", "git log origin/main..HEAD", "Hilo stats",
                "CI scan", "DuckBrain counter write", "board audit event",
                "COPY events parquet")
    return ("tests", "ruff", "guard", *light_steps)

def next_event_id(existing: Sequence[int]) -> int:
    return max(existing) + 1 if existing else 1  # MAX(id)+1, monotonic

PLAIN_INSERT = "INSERT INTO audit_events (id, ts, payload) VALUES (:id, :ts, :payload)"

def cooldown_put_blocked(pinned: int = COOLDOWN_S_PINNED) -> bool:
    return pinned == COOLDOWN_S_PINNED           # -> put_issued = 0

def e2e_status(tick: int, window: tuple[int, int] = E2E_001_WINDOW) -> str:
    lo, hi = window
    return "armed" if tick < lo else "due" if tick <= hi else "passed"

Tick #52 execution (run_idle_audit): skip heavy → run the 7 light steps → board/DuckBrain event id = MAX(id)+1 via plain INSERT (no UPSERT/ON CONFLICT) → COPY events TO 'audit/events.parquet' (FORMAT PARQUET) → put_issued = 0 (cooldown pinned, no scheduler PUT) → e2e = "armed" (battery window #53–58 upcoming; battery fires at tick #53).