stale-gate-janitor

stale-gate-janitor

Why this exists

Active approval-gates dir accumulates resolved + expired entries that audit reads (hermes-daily-audit) keep tripping on. Today's session (2026-04-29) caught 3 stale Apr-28 gates Cooper had to mentally filter every morning. This skill removes that toll.

Inputs

{
  "dry_run": false,
  "test_telegram_chat_id": null
}

dry_run=true walks + classifies but doesn't write or Telegram.

test_telegram_chat_id (env var TELEGRAM_TEST_CHAT_ID) — if set, sends summary to that chat instead of Cooper's. Used by self-test to avoid pinging Cooper during ops verification.

Procedure

Step 1 — Walk + classify

import json, datetime, pathlib, shutil, os, subprocess
home = pathlib.Path.home()
repo = home / "Dev/daily-ai-agent-os"
gates_dir = repo / "data/approval-gates"
archive_root = gates_dir / "archive"
now = datetime.datetime.now(datetime.timezone.utc)

if not gates_dir.exists():
    print("[stale-gate-janitor] no approval-gates dir; nothing to do")
    raise SystemExit(0)

ARCHIVE_AGE_S = 7 * 86400

expired_now = []
archived = []
errors = []

# Pass 1 — expire stale-pending gates
for p in gates_dir.glob("*.json"):
    if p.parent != gates_dir:  # skip archive subdir
        continue
    try:
        g = json.loads(p.read_text())
    except Exception as e:
        errors.append({"file": str(p), "err": f"parse: {e}"})
        continue
    if g.get("status") in (None, "pending", "open"):
        ex_iso = g.get("expires_at", "")
        try:
            ex_dt = datetime.datetime.fromisoformat(ex_iso.replace("Z", "+00:00")) if ex_iso else None
        except Exception:
            ex_dt = None
        if ex_dt and ex_dt < now:
            g["status"] = "expired"
            g["expired_at"] = now.isoformat()
            g["expired_by"] = "stale-gate-janitor"
            p.write_text(json.dumps(g, indent=2))
            expired_now.append(g.get("id", p.stem))

# Pass 2 — archive >7d files
for p in list(gates_dir.glob("*.json")):
    if p.parent != gates_dir:
        continue
    age_s = now.timestamp() - p.stat().st_mtime
    if age_s <= ARCHIVE_AGE_S:
        continue
    try:
        g = json.loads(p.read_text())
        ts = g.get("created_at") or g.get("expires_at") or now.isoformat()
        ym = ts[:7]  # YYYY-MM
    except Exception:
        ym = now.strftime("%Y-%m")
    bucket = archive_root / ym
    bucket.mkdir(parents=True, exist_ok=True)
    target = bucket / p.name
    shutil.move(str(p), str(target))
    archived.append(p.stem)

print(f"[stale-gate-janitor] expired={len(expired_now)} archived={len(archived)} errors={len(errors)}")