python-standing-block-watchdog-key-probe

Answer 1

A standing-block watchdog at ~/openrouter_key_watchdog.py that probes the OpenRouter key once per tick and reports state. Core probe (raw curl -s -w HTTP code, exactly as the standing task requires):

AUTH_URL = "https://openrouter.ai/api/v1/auth/key"
NOT_FOUND_MARKER = "user not found"  # 401 body: {"error":{"message":"User not found."}}

def probe(key: str) -> dict:
    if not key:
        return {"http_code": None, "body": "", "curl_rc": None, "ok": False,
                "note": "OPENROUTER_API_KEY is unset/empty"}
    proc = subprocess.run(
        ["curl", "-s", "-w", "%{http_code}", "-H", f"Authorization: Bearer {key}",
         "-m", "15", AUTH_URL],
        capture_output=True, text=True, timeout=30)
    raw = proc.stdout
    http_code = raw[-3:] if raw[-3:].isdigit() else ""
    body = raw[:-3] if http_code else raw
    return {"http_code": http_code or None, "body": body.strip()[:400],
            "curl_rc": proc.returncode, "ok": proc.returncode == 0 and bool(http_code)}

Classification and the no-solution rule:

def classify(state: dict) -> str:
    http = state.get("http_code"); body = state.get("body", "").lower()
    if not state.get("ok") or http is None: return "PROBE_ERROR"
    if http == "200":                        return "KEY_VALID"
    if http == "401" and NOT_FOUND_MARKER in body:
        return "KEY_NOT_FOUND"   # rotated/revoked -> NORMAL no-solution state
    if http == "429":                        return "RATE_LIMITED"
    return f"UNEXPECTED_HTTP_{http}"

The loop runs forever (TICK_INTERVAL/--interval, default 60s; --once for a single tick). Every tick emits a timestamped report line; the 401 report is explicit:

REPORT: 401 User not found — key is rotated/revoked; not_found is the NORMAL no-solution outcome (no fix to apply, keep probing).

Key design decisions: - 401 + "User not found" body is the expected no-solution signal, not an error: the watchdog records observed_not_found=True, keeps probing every tick, and exits 0. A 401 without the marker is NOT treated as no-solution (UNEXPECTED_HTTP_401). - Only the exact curl -s -w "%{http_code}" last-3-chars trick is trusted for the code; everything else is body. Network/curl failures (PROBE_ERROR) are retried next tick, never fatal. - *TRANSITION* tags mark state changes (e.g., KEY_VALID → KEY_NOT_FOUND when rotation happens), so a human/upstream consumer can see exactly when the key flipped.

Answer 2

Implemented ~/standing_block_watchdog.py (stdlib-only) — a standing-block watchdog that encodes the proven Foreman pattern (T113–T117) for the revoked OpenRouter key blocking E2E-001 (AP-GAP-006):

Core loop (one probe per tick):

def probe_key(key, url=OPENROUTER_AUTH_KEY_URL, timeout=PROBE_TIMEOUT_S):
    """GET /api/v1/auth/key with Bearer auth.
    Returns (status, detail): "valid" | "revoked" (HTTP 401, incl. User not found)
    | "unreachable" (network/5xx — transient, never confirms)."""
    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {key}"})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return "valid", f"HTTP 200: {json.loads(resp.read() or b'{}').get('data', {}).get('label', 'ok')}"
    except urllib.error.HTTPError as e:
        if e.code == 401:
            msg = json.loads(e.read().decode() or "{}").get("error", {}).get("message", "")
            return "revoked", f"HTTP 401: {msg or 'unauthorized'}"
        return "unreachable", f"HTTP {e.code}"
    except Exception as e:
        return "unreachable", f"{type(e).__name__}: {e}"

KeyProbeWatchdog.tick(audit_locations) — the decision matrix:

1. probe each tick → valid / revoked / unreachable
2. revoked → bump streak, then run the NEVER-DONE light audit exactly once (fleet-key-usage-light-audit, grep-based, zero model calls, zero gameplay spend). Confirmation requires the revoked key found in 3 fleet locations (confirm_locations=3, matching "confirmed in 3 locations fleet-wide"). Once confirmed: E2E-001 → DEFERRED, decision=DEFER_E2E-001, and the T112 cost guard accrues $1.00 saved per skipped degraded 0-decision run.
3. unreachable → OBSERVE_UNREACHABLE: never confirms a block on network noise, and never clears an existing confirmed block on noise either.
4. valid → rotation happened in the dashboard: clear the block, E2E-001 → READY, decision=RESUME_E2E-001, record resolved_at.
5. record the standing block → persistent JSON state (standing_block_state.json); the full secret is never persisted — only the redacted fingerprint sk-or-v1-3d…e3f.

class KeyProbeWatchdog:
    def tick(self, audit_locations=None):
        st = self.state
        st.probe_count += 1
        status, detail = self.prober(self.key)      # injectable for tests
        if status == "valid":                        # rotation -> unblock
            was_blocked = st.confirmed or st.e2e_001 == "DEFERRED"
            st.confirmed = False; st.e2e_001 = "READY"
            st.decision = "RESUME_E2E-001" if was_blocked else "PROBE_ONLY"
            st.status = "resolved" if was_blocked else "monitoring"
            if was_blocked: st.resolved_at = iso_now()
            ...
        if status == "revoked":
            st.revoked_streak += 1
            if not st.audit_ref:                     # NEVER-DONE audit, run once
                st.audit_ref = f"{AUDIT_NAME}-tick{st.probe_count}"
                st.audit_report = self.audit(self.key, audit_locations or [])
            st.confirmed = (st.audit_report.get("locations_with_revoked_key", 0)
                            >= self.confirm_locations) if audit_locations \
                          else st.revoked_streak >= self.confirm_locations
            if st.confirmed:
                st.e2e_001 = "DEFERRED"; st.decision = "DEFER_E2E-001"
                st.degraded_run_saved_usd = round(st.degraded_run_saved_usd + 1.00, 2)
            ...

CLI (usable in cron/systemd for the per-tick probe):

python3 standing_block_watchdog.py --tick --key "$OPENROUTER_API_KEY" \
    --audit-dir ./fleet/node0/config.env --audit-dir ./fleet/node1/config.env \
    --audit-dir ./fleet/node2/config.env --state /var/lib/e2e/standing_block_state.json
python3 standing_block_watchdog.py --status --state /var/lib/e2e/standing_block_state.json

Answer 3

A standing-block watchdog at ~/openrouter_key_watchdog.py that probes the OpenRouter key once per tick and reports state. Core probe (raw curl -s -w HTTP code, exactly as the standing task requires):

AUTH_URL = "https://openrouter.ai/api/v1/auth/key"
NOT_FOUND_MARKER = "user not found"  # 401 body: {"error":{"message":"User not found."}}

def probe(key: str) -> dict:
    if not key:
        return {"http_code": None, "body": "", "curl_rc": None, "ok": False,
                "note": "OPENROUTER_API_KEY is unset/empty"}
    proc = subprocess.run(
        ["curl", "-s", "-w", "%{http_code}", "-H", f"Authorization: Bearer {key}",
         "-m", "15", AUTH_URL],
        capture_output=True, text=True, timeout=30)
    raw = proc.stdout
    http_code = raw[-3:] if raw[-3:].isdigit() else ""
    body = raw[:-3] if http_code else raw
    return {"http_code": http_code or None, "body": body.strip()[:400],
            "curl_rc": proc.returncode, "ok": proc.returncode == 0 and bool(http_code)}

Classification and the no-solution rule:

def classify(state: dict) -> str:
    http = state.get("http_code"); body = state.get("body", "").lower()
    if not state.get("ok") or http is None: return "PROBE_ERROR"
    if http == "200":                        return "KEY_VALID"
    if http == "401" and NOT_FOUND_MARKER in body:
        return "KEY_NOT_FOUND"   # rotated/revoked -> NORMAL no-solution state
    if http == "429":                        return "RATE_LIMITED"
    return f"UNEXPECTED_HTTP_{http}"

The loop runs forever (TICK_INTERVAL/--interval, default 60s; --once for a single tick). Every tick emits a timestamped report line; the 401 report is explicit:

REPORT: 401 User not found — key is rotated/revoked; not_found is the NORMAL no-solution outcome (no fix to apply, keep probing).

Key design decisions: - 401 + "User not found" body is the expected no-solution signal, not an error: the watchdog records observed_not_found=True, keeps probing every tick, and exits 0. A 401 without the marker is NOT treated as no-solution (UNEXPECTED_HTTP_401). - Only the exact curl -s -w "%{http_code}" last-3-chars trick is trusted for the code; everything else is body. Network/curl failures (PROBE_ERROR) are retried next tick, never fatal. - *TRANSITION* tags mark state changes (e.g., KEY_VALID → KEY_NOT_FOUND when rotation happens), so a human/upstream consumer can see exactly when the key flipped.

Answer 4

Implemented ~/standing_block_watchdog.py (stdlib-only) — a standing-block watchdog that encodes the proven Foreman pattern (T113–T117) for the revoked OpenRouter key blocking E2E-001 (AP-GAP-006):

Core loop (one probe per tick):

def probe_key(key, url=OPENROUTER_AUTH_KEY_URL, timeout=PROBE_TIMEOUT_S):
    """GET /api/v1/auth/key with Bearer auth.
    Returns (status, detail): "valid" | "revoked" (HTTP 401, incl. User not found)
    | "unreachable" (network/5xx — transient, never confirms)."""
    req = urllib.request.Request(url, headers={"Authorization": f"Bearer {key}"})
    try:
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            return "valid", f"HTTP 200: {json.loads(resp.read() or b'{}').get('data', {}).get('label', 'ok')}"
    except urllib.error.HTTPError as e:
        if e.code == 401:
            msg = json.loads(e.read().decode() or "{}").get("error", {}).get("message", "")
            return "revoked", f"HTTP 401: {msg or 'unauthorized'}"
        return "unreachable", f"HTTP {e.code}"
    except Exception as e:
        return "unreachable", f"{type(e).__name__}: {e}"

KeyProbeWatchdog.tick(audit_locations) — the decision matrix:

1. probe each tick → valid / revoked / unreachable
2. revoked → bump streak, then run the NEVER-DONE light audit exactly once (fleet-key-usage-light-audit, grep-based, zero model calls, zero gameplay spend). Confirmation requires the revoked key found in 3 fleet locations (confirm_locations=3, matching "confirmed in 3 locations fleet-wide"). Once confirmed: E2E-001 → DEFERRED, decision=DEFER_E2E-001, and the T112 cost guard accrues $1.00 saved per skipped degraded 0-decision run.
3. unreachable → OBSERVE_UNREACHABLE: never confirms a block on network noise, and never clears an existing confirmed block on noise either.
4. valid → rotation happened in the dashboard: clear the block, E2E-001 → READY, decision=RESUME_E2E-001, record resolved_at.
5. record the standing block → persistent JSON state (standing_block_state.json); the full secret is never persisted — only the redacted fingerprint sk-or-v1-3d…e3f.

class KeyProbeWatchdog:
    def tick(self, audit_locations=None):
        st = self.state
        st.probe_count += 1
        status, detail = self.prober(self.key)      # injectable for tests
        if status == "valid":                        # rotation -> unblock
            was_blocked = st.confirmed or st.e2e_001 == "DEFERRED"
            st.confirmed = False; st.e2e_001 = "READY"
            st.decision = "RESUME_E2E-001" if was_blocked else "PROBE_ONLY"
            st.status = "resolved" if was_blocked else "monitoring"
            if was_blocked: st.resolved_at = iso_now()
            ...
        if status == "revoked":
            st.revoked_streak += 1
            if not st.audit_ref:                     # NEVER-DONE audit, run once
                st.audit_ref = f"{AUDIT_NAME}-tick{st.probe_count}"
                st.audit_report = self.audit(self.key, audit_locations or [])
            st.confirmed = (st.audit_report.get("locations_with_revoked_key", 0)
                            >= self.confirm_locations) if audit_locations \
                          else st.revoked_streak >= self.confirm_locations
            if st.confirmed:
                st.e2e_001 = "DEFERRED"; st.decision = "DEFER_E2E-001"
                st.degraded_run_saved_usd = round(st.degraded_run_saved_usd + 1.00, 2)
            ...

CLI (usable in cron/systemd for the per-tick probe):

python3 standing_block_watchdog.py --tick --key "$OPENROUTER_API_KEY" \
    --audit-dir ./fleet/node0/config.env --audit-dir ./fleet/node1/config.env \
    --audit-dir ./fleet/node2/config.env --state /var/lib/e2e/standing_block_state.json
python3 standing_block_watchdog.py --status --state /var/lib/e2e/standing_block_state.json