feat(h6): AgentOps hardening — live alerting, provider-API reconcile, hosted dashboard, window breaker (79→80)
Close the three gaps the scoring report itself flagged for H6 plus V15, each as a real MVP + fail-able adversarial test (same pattern that lifted H5): - D1 alert-dispatch.sh: alerts POST to a real HTTP webhook (severity routing, dedup window, retry) + dead-letter queue with redelivery; fail-loud in strict. Wired into agent-metrics.sh so a failing step pages live end-to-end. - D2 provider-usage-fetch.sh + telemetry-reconcile.sh: pull usage from a provider usage HTTP API (all-or-nothing schema gate, fail-loud) + reconcile local vs provider ground truth — token under-reporting/hidden runs => TELEMETRY_DISCREPANCY. - D3 dashboard-serve.sh + dashboard-server.py: serve the dashboard over HTTP with a stale-aware /healthz probe (fresh=200 ok, telemetry silent-death=503 stale). - D4 circuit-breaker-check.sh: sliding-window failure-rate breaker (V15) — interleaved successes no longer evade the consecutive-failure breaker (CIRCUIT_OPEN_WINDOW). New suite phase-h6-agentops-tests.sh: 20/20, all live against local HTTP endpoints (webhook sink, mock provider API, dashboard server) — deterministic, no model needed. Also fix sign-policy-bundle.sh key-sync invariant: the local-fallback branch only exported policy-public.pem when generating a NEW key, so a Vault-DOWN run after a Vault-signed run verified a local-key signature against the Vault pubkey (RSA padding error, run-casan4 died mid-suite). Now always re-exports the pubkey before signing — same fix class as tool-audit-lib.sh / governance-check.sh. Full battery re-run sequentially: 175/175 PASS, 0 FAIL across 8 suites (KMS SKIP this run — Vault down; validated live 2026-07-04). Docs synced: scoring-run-report (H6 79→80, no harness below 80, 155→175), CASAN_HARDENING_STATUS (Phase 5 D1–D4), Plan-07, submission README, and run-hardening.sh (H6+ scenes HO1–HO4). Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
co-authored by
Claude Fable 5
parent
2af67ef6a5
commit
da66a36f97
@@ -199,8 +199,18 @@ printf '{"timestamp":"%s","trace_id":"%s","harness":"H6-agentops","agent":"%s","
|
||||
|
||||
for alert in "${ALERTS[@]:-}"; do
|
||||
if [[ -n "$alert" ]]; then
|
||||
printf '{"timestamp":"%s","trace_id":"%s","severity":"WARN","resource":{"service.name":"%s","service.version":"1.0.0"},"body":{"message":"Alert triggered: %s","alert.type":"%s","step.name":"%s"},"attributes":{"latency_ms":%s,"status":"%s"}}\n' \
|
||||
"$START_TS" "$TRACE_ID" "$AGENT_NAME" "$alert" "$alert" "$STEP_NAME" "$LATENCY_MS" "$STATUS" >> "$ALERT_LOG"
|
||||
ALERT_JSON="$(printf '{"timestamp":"%s","trace_id":"%s","severity":"WARN","resource":{"service.name":"%s","service.version":"1.0.0"},"body":{"message":"Alert triggered: %s","alert.type":"%s","step.name":"%s"},"attributes":{"latency_ms":%s,"status":"%s"}}' \
|
||||
"$START_TS" "$TRACE_ID" "$AGENT_NAME" "$alert" "$alert" "$STEP_NAME" "$LATENCY_MS" "$STATUS")"
|
||||
printf '%s\n' "$ALERT_JSON" >> "$ALERT_LOG"
|
||||
# Live dispatch (H6-D1): push to the real alert channel when configured.
|
||||
# Delivery failure is queued to the dead-letter file by alert-dispatch.sh.
|
||||
if [[ -n "${CASAN_ALERT_WEBHOOK:-}" ]]; then
|
||||
ALERT_TMP="$(mktemp)"
|
||||
printf '%s\n' "$ALERT_JSON" > "$ALERT_TMP"
|
||||
bash "$SCRIPT_DIR/alert-dispatch.sh" "$ALERT_TMP" \
|
||||
|| echo "AGENTOPS_ALERT_DISPATCH_FAILED alert=$alert (queued to dead-letter)" >&2
|
||||
rm -f "$ALERT_TMP"
|
||||
fi
|
||||
fi
|
||||
done
|
||||
|
||||
|
||||
@@ -0,0 +1,145 @@
|
||||
#!/usr/bin/env bash
|
||||
set -uo pipefail
|
||||
|
||||
# CASAN H6 — live alert dispatch (D1).
|
||||
# Pushes AgentOps alerts to a real HTTP webhook (Slack/Teams/PagerDuty-style
|
||||
# endpoint) instead of only appending to a local log file. Undelivered alerts
|
||||
# are queued to a dead-letter file so no alert is silently lost.
|
||||
#
|
||||
# Usage:
|
||||
# alert-dispatch.sh <alert-json-file> dispatch one alert (JSON object)
|
||||
# alert-dispatch.sh --flush-deadletter retry alerts that failed delivery
|
||||
#
|
||||
# Env:
|
||||
# CASAN_ALERT_WEBHOOK webhook URL (required to dispatch)
|
||||
# CASAN_ALERT_STRICT=1 delivery failure => exit 1 (fail-loud); default warn
|
||||
# CASAN_ALERT_DEDUP_WINDOW_S suppress same service/step/type within N s (default 300)
|
||||
# CASAN_AGENTOPS_DIR state dir override (default .specify/agentops)
|
||||
#
|
||||
# Greppable outputs:
|
||||
# ALERT_DISPATCHED | ALERT_DEDUP_SUPPRESSED | ALERT_DELIVERY_FAILED |
|
||||
# ALERT_DEADLETTER_FLUSHED | ALERT_WEBHOOK_UNSET
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
PROJECT_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)"
|
||||
AGENTOPS_DIR="${CASAN_AGENTOPS_DIR:-$PROJECT_ROOT/.specify/agentops}"
|
||||
STATE="$AGENTOPS_DIR/alert-dispatch-state.jsonl"
|
||||
DEADLETTER="$AGENTOPS_DIR/alert-deadletter.jsonl"
|
||||
WEBHOOK="${CASAN_ALERT_WEBHOOK:-}"
|
||||
STRICT="${CASAN_ALERT_STRICT:-0}"
|
||||
DEDUP_S="${CASAN_ALERT_DEDUP_WINDOW_S:-300}"
|
||||
mkdir -p "$AGENTOPS_DIR"
|
||||
|
||||
post_payload() { # <json-payload> — 0 = delivered
|
||||
curl -sS -m 5 --retry 2 --retry-delay 1 \
|
||||
-H 'Content-Type: application/json' \
|
||||
-d "$1" "$WEBHOOK" >/dev/null 2>&1
|
||||
}
|
||||
|
||||
if [[ "${1:-}" == "--flush-deadletter" ]]; then
|
||||
if [[ -z "$WEBHOOK" ]]; then
|
||||
echo "ALERT_WEBHOOK_UNSET cannot flush dead-letter queue" >&2
|
||||
exit 1
|
||||
fi
|
||||
if [[ ! -s "$DEADLETTER" ]]; then
|
||||
echo "ALERT_DEADLETTER_FLUSHED redelivered=0 remaining=0"
|
||||
exit 0
|
||||
fi
|
||||
TMP="$DEADLETTER.tmp"
|
||||
: > "$TMP"
|
||||
sent=0; kept=0
|
||||
while IFS= read -r line; do
|
||||
[[ -z "$line" ]] && continue
|
||||
if post_payload "$line"; then
|
||||
sent=$((sent + 1))
|
||||
else
|
||||
printf '%s\n' "$line" >> "$TMP"
|
||||
kept=$((kept + 1))
|
||||
fi
|
||||
done < "$DEADLETTER"
|
||||
mv "$TMP" "$DEADLETTER"
|
||||
echo "ALERT_DEADLETTER_FLUSHED redelivered=$sent remaining=$kept"
|
||||
[[ "$kept" -eq 0 ]] || exit 1
|
||||
exit 0
|
||||
fi
|
||||
|
||||
ALERT_FILE="${1:-}"
|
||||
if [[ -z "$ALERT_FILE" || ! -f "$ALERT_FILE" ]]; then
|
||||
echo "Usage: alert-dispatch.sh <alert-json-file> | --flush-deadletter" >&2
|
||||
exit 64
|
||||
fi
|
||||
|
||||
if [[ -z "$WEBHOOK" ]]; then
|
||||
echo "ALERT_WEBHOOK_UNSET alert not dispatched (set CASAN_ALERT_WEBHOOK)" >&2
|
||||
[[ "$STRICT" == "1" ]] && exit 1
|
||||
exit 0
|
||||
fi
|
||||
|
||||
# Normalize the alert, decide severity, and apply the dedup window.
|
||||
DECISION="$(python - "$ALERT_FILE" "$STATE" "$DEDUP_S" <<'PY'
|
||||
import json, sys, time
|
||||
|
||||
alert_path, state_path, window = sys.argv[1], sys.argv[2], int(sys.argv[3])
|
||||
alert = json.load(open(alert_path, encoding="utf-8"))
|
||||
body = alert.get("body", {}) if isinstance(alert.get("body"), dict) else {}
|
||||
resource = alert.get("resource", {}) if isinstance(alert.get("resource"), dict) else {}
|
||||
atype = body.get("alert.type") or alert.get("alert_type") or "unknown"
|
||||
step = body.get("step.name") or alert.get("step") or "unknown"
|
||||
service = resource.get("service.name") or alert.get("agent") or "unknown"
|
||||
critical = {"execution-failed", "circuit-open", "cost-spike", "token-overuse", "audit-gap"}
|
||||
severity = "CRITICAL" if atype in critical else "WARN"
|
||||
key = f"{service}/{step}/{atype}"
|
||||
now = int(time.time())
|
||||
last = None
|
||||
try:
|
||||
with open(state_path, encoding="utf-8") as fh:
|
||||
for line in fh:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
rec = json.loads(line)
|
||||
if rec.get("key") == key:
|
||||
last = rec.get("ts")
|
||||
except OSError:
|
||||
pass
|
||||
if last is not None and now - last < window:
|
||||
print("SUPPRESS " + key)
|
||||
raise SystemExit(0)
|
||||
payload = json.dumps({
|
||||
"source": "casan-agentops",
|
||||
"severity": severity,
|
||||
"alert_type": atype,
|
||||
"step": step,
|
||||
"service": service,
|
||||
"dedup_key": key,
|
||||
"alert": alert,
|
||||
})
|
||||
with open(state_path, "a", encoding="utf-8") as fh:
|
||||
fh.write(json.dumps({"key": key, "ts": now}) + "\n")
|
||||
print("SEND " + payload)
|
||||
PY
|
||||
)"
|
||||
|
||||
case "$DECISION" in
|
||||
SUPPRESS*)
|
||||
echo "ALERT_DEDUP_SUPPRESSED key=${DECISION#SUPPRESS } window_s=$DEDUP_S"
|
||||
exit 0
|
||||
;;
|
||||
SEND*)
|
||||
PAYLOAD="${DECISION#SEND }"
|
||||
;;
|
||||
*)
|
||||
echo "ALERT_DISPATCH_ERROR unparsable alert file: $ALERT_FILE" >&2
|
||||
exit 1
|
||||
;;
|
||||
esac
|
||||
|
||||
if post_payload "$PAYLOAD"; then
|
||||
echo "ALERT_DISPATCHED webhook=$WEBHOOK"
|
||||
exit 0
|
||||
fi
|
||||
|
||||
printf '%s\n' "$PAYLOAD" >> "$DEADLETTER"
|
||||
echo "ALERT_DELIVERY_FAILED queued=dead-letter webhook=$WEBHOOK" >&2
|
||||
[[ "$STRICT" == "1" ]] && exit 1
|
||||
exit 0
|
||||
@@ -11,14 +11,22 @@ set -uo pipefail
|
||||
# consecutive recent failures. If ≥ CIRCUIT_BREAKER_THRESHOLD consecutive
|
||||
# model calls failed, prints CIRCUIT_OPEN and exits non-zero so the caller
|
||||
# can stop invoking the model (prevents cascading failures / cost runaway).
|
||||
# Also runs a SLIDING-WINDOW breaker (V15): a failure RATE ≥
|
||||
# CIRCUIT_WINDOW_FAIL_PCT over the last CIRCUIT_WINDOW records trips
|
||||
# CIRCUIT_OPEN_WINDOW — interleaving successes between failures no longer
|
||||
# evades the breaker.
|
||||
#
|
||||
# Usage: circuit-breaker-check.sh [--no-bypass-only | --breaker-only]
|
||||
# Env: CASAN_PROVIDER_LOG (log override), CIRCUIT_BREAKER_THRESHOLD,
|
||||
# CIRCUIT_WINDOW (default 10), CIRCUIT_WINDOW_FAIL_PCT (default 50)
|
||||
# Exit: 0 all OK, 1 bypass found or circuit open.
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)"
|
||||
PROVIDER_LOG="$ROOT/.specify/logs/level5/provider-usage.jsonl"
|
||||
PROVIDER_LOG="${CASAN_PROVIDER_LOG:-$ROOT/.specify/logs/level5/provider-usage.jsonl}"
|
||||
CIRCUIT_BREAKER_THRESHOLD="${CIRCUIT_BREAKER_THRESHOLD:-5}"
|
||||
CIRCUIT_WINDOW="${CIRCUIT_WINDOW:-10}"
|
||||
CIRCUIT_WINDOW_FAIL_PCT="${CIRCUIT_WINDOW_FAIL_PCT:-50}"
|
||||
MODE="${1:-both}"
|
||||
FAIL=0
|
||||
|
||||
@@ -110,6 +118,43 @@ PY
|
||||
else
|
||||
ok "Circuit breaker closed: consecutive_failures=$consecutive_fails (threshold=$CIRCUIT_BREAKER_THRESHOLD)"
|
||||
fi
|
||||
|
||||
# Sliding-window failure RATE (V15): interleaved successes reset the
|
||||
# consecutive counter but do not hide a failing provider from the rate.
|
||||
window_stats="$(python - "$PROVIDER_LOG" "$CIRCUIT_WINDOW" << 'PY'
|
||||
import json, sys
|
||||
|
||||
log_file, window = sys.argv[1], int(sys.argv[2])
|
||||
try:
|
||||
lines = [l for l in open(log_file) if l.strip()]
|
||||
recent = lines[-window:]
|
||||
fails = 0
|
||||
total = 0
|
||||
for line in recent:
|
||||
try:
|
||||
r = json.loads(line)
|
||||
except json.JSONDecodeError:
|
||||
continue
|
||||
total += 1
|
||||
if r.get("status") in ("error", "fail", "failed"):
|
||||
fails += 1
|
||||
print(f"{fails} {total}")
|
||||
except Exception:
|
||||
print("0 0")
|
||||
PY
|
||||
)"
|
||||
window_fails="${window_stats%% *}"
|
||||
window_total="${window_stats##* }"
|
||||
if [[ "$window_total" -ge "$CIRCUIT_WINDOW" ]]; then
|
||||
window_pct=$(( window_fails * 100 / window_total ))
|
||||
if [[ "$window_pct" -ge "$CIRCUIT_WINDOW_FAIL_PCT" ]]; then
|
||||
fail "CIRCUIT_OPEN_WINDOW: failure rate ${window_pct}% over last $window_total calls (threshold=${CIRCUIT_WINDOW_FAIL_PCT}%) — interleaved successes do not close the circuit"
|
||||
else
|
||||
ok "Window breaker closed: failure rate ${window_pct}% over last $window_total calls (threshold=${CIRCUIT_WINDOW_FAIL_PCT}%)"
|
||||
fi
|
||||
else
|
||||
ok "Window breaker closed: only $window_total records (< window=$CIRCUIT_WINDOW), rate not evaluated"
|
||||
fi
|
||||
fi
|
||||
fi
|
||||
|
||||
|
||||
@@ -0,0 +1,71 @@
|
||||
#!/usr/bin/env bash
|
||||
set -uo pipefail
|
||||
|
||||
# CASAN H6 — hosted AgentOps dashboard (D3).
|
||||
# Regenerates the dashboard then serves it over HTTP (dashboard-server.py)
|
||||
# with a stale-aware /healthz probe. MVP hosting: local HTTP daemon with a
|
||||
# pid file; production swaps in a real host (nginx/container) same routes.
|
||||
#
|
||||
# Usage:
|
||||
# dashboard-serve.sh start [port] regenerate + serve (default port 8787)
|
||||
# dashboard-serve.sh stop stop the running server
|
||||
# dashboard-serve.sh status curl /healthz of the running server
|
||||
#
|
||||
# Env: CASAN_DASHBOARD_STALE_S, CASAN_DASHBOARD_METRICS, CASAN_DASHBOARD_HTML,
|
||||
# CASAN_AGENTOPS_DIR (pid-file location)
|
||||
#
|
||||
# Greppable outputs: DASHBOARD_HOSTED | DASHBOARD_STOPPED | DASHBOARD_NOT_RUNNING
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
PROJECT_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)"
|
||||
AGENTOPS_DIR="${CASAN_AGENTOPS_DIR:-$PROJECT_ROOT/.specify/agentops}"
|
||||
PID_FILE="$AGENTOPS_DIR/dashboard.pid"
|
||||
PORT_FILE="$AGENTOPS_DIR/dashboard.port"
|
||||
GENERATOR="$PROJECT_ROOT/.specify/tests/generate-agentops-dashboard.py"
|
||||
CMD="${1:-start}"
|
||||
mkdir -p "$AGENTOPS_DIR"
|
||||
|
||||
case "$CMD" in
|
||||
start)
|
||||
PORT="${2:-8787}"
|
||||
if [[ -f "$GENERATOR" ]]; then
|
||||
python "$GENERATOR" >/dev/null 2>&1 || true
|
||||
fi
|
||||
python "$SCRIPT_DIR/dashboard-server.py" "$PORT" &
|
||||
SERVER_PID=$!
|
||||
echo "$SERVER_PID" > "$PID_FILE"
|
||||
echo "$PORT" > "$PORT_FILE"
|
||||
for _ in 1 2 3 4 5 6 7 8 9 10; do
|
||||
if curl -sS -m 2 -o /dev/null "http://127.0.0.1:$PORT/healthz" 2>/dev/null; then
|
||||
echo "DASHBOARD_HOSTED url=http://127.0.0.1:$PORT/ healthz=http://127.0.0.1:$PORT/healthz pid=$SERVER_PID"
|
||||
exit 0
|
||||
fi
|
||||
sleep 0.3
|
||||
done
|
||||
echo "DASHBOARD_START_FAILED port=$PORT (server did not come up)" >&2
|
||||
kill "$SERVER_PID" 2>/dev/null
|
||||
rm -f "$PID_FILE" "$PORT_FILE"
|
||||
exit 1
|
||||
;;
|
||||
stop)
|
||||
if [[ -f "$PID_FILE" ]] && kill "$(cat "$PID_FILE")" 2>/dev/null; then
|
||||
echo "DASHBOARD_STOPPED pid=$(cat "$PID_FILE")"
|
||||
rm -f "$PID_FILE" "$PORT_FILE"
|
||||
exit 0
|
||||
fi
|
||||
echo "DASHBOARD_NOT_RUNNING" >&2
|
||||
rm -f "$PID_FILE" "$PORT_FILE"
|
||||
exit 1
|
||||
;;
|
||||
status)
|
||||
if [[ -f "$PORT_FILE" ]]; then
|
||||
curl -sS -m 3 "http://127.0.0.1:$(cat "$PORT_FILE")/healthz" && echo "" && exit 0
|
||||
fi
|
||||
echo "DASHBOARD_NOT_RUNNING" >&2
|
||||
exit 1
|
||||
;;
|
||||
*)
|
||||
echo "Usage: dashboard-serve.sh start [port] | stop | status" >&2
|
||||
exit 64
|
||||
;;
|
||||
esac
|
||||
@@ -0,0 +1,78 @@
|
||||
#!/usr/bin/env python3
|
||||
"""CASAN H6 — hosted AgentOps dashboard server (D3).
|
||||
|
||||
Serves the generated dashboard over HTTP with a stale-aware /healthz probe so
|
||||
an external monitor (uptime check / load-balancer) can page when telemetry
|
||||
stops flowing — not just when the process dies.
|
||||
|
||||
Routes:
|
||||
GET / -> dashboard HTML (also /dashboard)
|
||||
GET /healthz -> 200 {"status":"ok",...} while metrics are fresh,
|
||||
503 {"status":"stale",...} when metrics are older than
|
||||
CASAN_DASHBOARD_STALE_S (default 3600s) or missing.
|
||||
|
||||
Usage: dashboard-server.py <port>
|
||||
"""
|
||||
import json
|
||||
import os
|
||||
import pathlib
|
||||
import sys
|
||||
import time
|
||||
from http.server import BaseHTTPRequestHandler, HTTPServer
|
||||
|
||||
ROOT = pathlib.Path(__file__).resolve().parents[3]
|
||||
PORT = int(sys.argv[1]) if len(sys.argv) > 1 else 8787
|
||||
DASH = pathlib.Path(os.environ.get(
|
||||
"CASAN_DASHBOARD_HTML", ROOT / "docs" / "output" / "casan" / "central-agentops-dashboard.html"))
|
||||
METRICS = pathlib.Path(os.environ.get(
|
||||
"CASAN_DASHBOARD_METRICS", ROOT / ".specify" / "logs" / "cost" / "metrics.jsonl"))
|
||||
ALERTS = pathlib.Path(os.environ.get(
|
||||
"CASAN_DASHBOARD_ALERTS", ROOT / ".specify" / "agentops" / "alerts.log"))
|
||||
STALE_S = int(os.environ.get("CASAN_DASHBOARD_STALE_S", "3600"))
|
||||
|
||||
|
||||
def count_lines(path: pathlib.Path) -> int:
|
||||
if not path.exists():
|
||||
return 0
|
||||
return sum(1 for line in path.read_text(encoding="utf-8").splitlines() if line.strip())
|
||||
|
||||
|
||||
class Handler(BaseHTTPRequestHandler):
|
||||
def _send(self, code: int, ctype: str, body: str) -> None:
|
||||
data = body.encode("utf-8")
|
||||
self.send_response(code)
|
||||
self.send_header("Content-Type", ctype)
|
||||
self.send_header("Content-Length", str(len(data)))
|
||||
self.end_headers()
|
||||
self.wfile.write(data)
|
||||
|
||||
def do_GET(self): # noqa: N802 (http.server API)
|
||||
if self.path == "/healthz":
|
||||
if METRICS.exists():
|
||||
age = int(time.time() - METRICS.stat().st_mtime)
|
||||
stale = age > STALE_S
|
||||
else:
|
||||
age = -1
|
||||
stale = True
|
||||
body = json.dumps({
|
||||
"status": "stale" if stale else "ok",
|
||||
"metrics_age_s": age,
|
||||
"stale_after_s": STALE_S,
|
||||
"runs": count_lines(METRICS),
|
||||
"alerts": count_lines(ALERTS),
|
||||
})
|
||||
self._send(503 if stale else 200, "application/json", body)
|
||||
elif self.path in ("/", "/dashboard"):
|
||||
if DASH.exists():
|
||||
self._send(200, "text/html; charset=utf-8", DASH.read_text(encoding="utf-8"))
|
||||
else:
|
||||
self._send(404, "text/plain", "dashboard not generated")
|
||||
else:
|
||||
self._send(404, "text/plain", "not found")
|
||||
|
||||
def log_message(self, *args): # silence per-request stderr noise
|
||||
pass
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
HTTPServer(("127.0.0.1", PORT), Handler).serve_forever()
|
||||
@@ -0,0 +1,70 @@
|
||||
#!/usr/bin/env bash
|
||||
set -uo pipefail
|
||||
|
||||
# CASAN H6 — provider usage telemetry API fetch (D2).
|
||||
# Pulls usage records from a provider usage HTTP API (production: the
|
||||
# OpenAI/Anthropic usage endpoints) and imports them into provider-usage.jsonl
|
||||
# through the same schema gate as import-provider-telemetry.sh.
|
||||
# Fail-loud by design: unreachable API or invalid schema => non-zero exit and
|
||||
# NOTHING is imported (all-or-nothing, no partial/dirty telemetry).
|
||||
#
|
||||
# Usage: provider-usage-fetch.sh <api-url> [out-jsonl]
|
||||
#
|
||||
# Greppable outputs:
|
||||
# PROVIDER_TELEMETRY_FETCHED | PROVIDER_USAGE_INVALID | PROVIDER_API_UNREACHABLE
|
||||
|
||||
API_URL="${1:-}"
|
||||
if [[ -z "$API_URL" ]]; then
|
||||
echo "Usage: provider-usage-fetch.sh <api-url> [out-jsonl]" >&2
|
||||
exit 64
|
||||
fi
|
||||
|
||||
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
|
||||
PROJECT_ROOT="$(cd "$SCRIPT_DIR/../../.." && pwd)"
|
||||
OUT="${2:-$PROJECT_ROOT/.specify/logs/level5/provider-usage.jsonl}"
|
||||
mkdir -p "$(dirname "$OUT")"
|
||||
|
||||
BODY="$(mktemp)"
|
||||
trap 'rm -f "$BODY"' EXIT
|
||||
|
||||
if ! curl -sS -m 10 --retry 2 --retry-delay 1 -f "$API_URL" -o "$BODY" 2>/dev/null; then
|
||||
echo "PROVIDER_API_UNREACHABLE url=$API_URL (telemetry NOT imported)" >&2
|
||||
exit 1
|
||||
fi
|
||||
|
||||
python - "$BODY" "$OUT" "$API_URL" <<'PY'
|
||||
import json
|
||||
import sys
|
||||
from datetime import datetime, timezone
|
||||
|
||||
src, out, url = sys.argv[1], sys.argv[2], sys.argv[3]
|
||||
try:
|
||||
data = json.load(open(src, encoding="utf-8"))
|
||||
except (json.JSONDecodeError, UnicodeDecodeError):
|
||||
raise SystemExit("PROVIDER_USAGE_INVALID response is not JSON")
|
||||
records = data if isinstance(data, list) else [data]
|
||||
required = ["provider", "model", "run_id", "step", "input_tokens", "output_tokens",
|
||||
"total_tokens", "cost_usd", "latency_ms", "status"]
|
||||
for idx, rec in enumerate(records):
|
||||
if not isinstance(rec, dict):
|
||||
raise SystemExit(f"PROVIDER_USAGE_INVALID record[{idx}] is not an object")
|
||||
missing = [key for key in required if key not in rec]
|
||||
if missing:
|
||||
raise SystemExit(f"PROVIDER_USAGE_INVALID record[{idx}] missing={missing}")
|
||||
now = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
|
||||
with open(out, "a", encoding="utf-8") as fh:
|
||||
for rec in records:
|
||||
fh.write(json.dumps({
|
||||
"timestamp": now,
|
||||
"harness": "L5-provider-telemetry",
|
||||
"telemetry_source": "provider_api",
|
||||
"api_endpoint": url,
|
||||
**rec,
|
||||
}) + "\n")
|
||||
print(f"PROVIDER_TELEMETRY_FETCHED count={len(records)} url={url} output={out}")
|
||||
PY
|
||||
RC=$?
|
||||
if [[ "$RC" -ne 0 ]]; then
|
||||
echo "PROVIDER_USAGE_INVALID import rejected (nothing written)" >&2
|
||||
exit 1
|
||||
fi
|
||||
@@ -78,8 +78,11 @@ if [[ "$MODE" == "sign" ]]; then
|
||||
# ── Fallback: local key file (dev / no Vault) ─────────────────────────────
|
||||
if [[ ! -f "$PRIVATE_KEY" ]]; then
|
||||
openssl genrsa -out "$PRIVATE_KEY" 2048 >/dev/null 2>&1
|
||||
openssl rsa -in "$PRIVATE_KEY" -pubout -out "$PUBLIC_KEY" >/dev/null 2>&1
|
||||
fi
|
||||
# Key-sync invariant: the on-disk public key must ALWAYS match the key that
|
||||
# signs (a prior Vault-signed run leaves the Vault pubkey here — verifying
|
||||
# a local-key signature against it would fail with an RSA padding error).
|
||||
openssl rsa -in "$PRIVATE_KEY" -pubout -out "$PUBLIC_KEY" >/dev/null 2>&1
|
||||
openssl dgst -sha256 -sign "$PRIVATE_KEY" -out "$SIGNATURE" "$MANIFEST"
|
||||
echo "POLICY_BUNDLE_SIGNED manifest=$MANIFEST signature=$SIGNATURE public_key=$PUBLIC_KEY key_backend=local-file"
|
||||
fi
|
||||
|
||||
@@ -0,0 +1,68 @@
|
||||
#!/usr/bin/env bash
|
||||
set -uo pipefail
|
||||
|
||||
# CASAN H6 — local-vs-provider telemetry reconciliation (D2).
|
||||
# Provider-API usage records are the billing ground truth; local metrics must
|
||||
# not under-report tokens (the cost-hiding attack: trim local metrics so a
|
||||
# runaway/exfil step looks cheap). Per step, local claimed tokens must cover
|
||||
# provider-reported tokens within a tolerance; a provider step entirely absent
|
||||
# from local metrics is also a discrepancy (hidden run).
|
||||
#
|
||||
# Usage: telemetry-reconcile.sh <local-metrics.jsonl> <provider-usage.jsonl> [tolerance_pct]
|
||||
#
|
||||
# Greppable outputs: TELEMETRY_RECONCILED | TELEMETRY_DISCREPANCY
|
||||
|
||||
LOCAL_LOG="${1:-}"
|
||||
PROVIDER_LOG="${2:-}"
|
||||
TOLERANCE_PCT="${3:-10}"
|
||||
|
||||
if [[ -z "$LOCAL_LOG" || -z "$PROVIDER_LOG" || ! -f "$PROVIDER_LOG" ]]; then
|
||||
echo "Usage: telemetry-reconcile.sh <local-metrics.jsonl> <provider-usage.jsonl> [tolerance_pct]" >&2
|
||||
exit 64
|
||||
fi
|
||||
|
||||
python - "$LOCAL_LOG" "$PROVIDER_LOG" "$TOLERANCE_PCT" <<'PY'
|
||||
import json
|
||||
import sys
|
||||
|
||||
local_path, provider_path, tol_pct = sys.argv[1], sys.argv[2], float(sys.argv[3])
|
||||
|
||||
def sums_by_step(path):
|
||||
totals = {}
|
||||
try:
|
||||
with open(path, encoding="utf-8") as fh:
|
||||
for line in fh:
|
||||
line = line.strip()
|
||||
if not line:
|
||||
continue
|
||||
rec = json.loads(line)
|
||||
step = rec.get("step")
|
||||
if step is None:
|
||||
continue
|
||||
totals[step] = totals.get(step, 0) + int(rec.get("total_tokens", 0))
|
||||
except OSError:
|
||||
pass
|
||||
return totals
|
||||
|
||||
local = sums_by_step(local_path)
|
||||
provider = sums_by_step(provider_path)
|
||||
if not provider:
|
||||
raise SystemExit("TELEMETRY_DISCREPANCY provider log empty — nothing to reconcile against")
|
||||
|
||||
issues = []
|
||||
for step, prov_tokens in sorted(provider.items()):
|
||||
loc_tokens = local.get(step)
|
||||
if loc_tokens is None:
|
||||
issues.append(f"step={step} local=MISSING provider={prov_tokens}")
|
||||
continue
|
||||
floor = prov_tokens * (1 - tol_pct / 100.0)
|
||||
if loc_tokens < floor:
|
||||
issues.append(f"step={step} local={loc_tokens} provider={prov_tokens} (under-reported beyond {tol_pct}%)")
|
||||
|
||||
if issues:
|
||||
for issue in issues:
|
||||
print(f"TELEMETRY_DISCREPANCY {issue}", file=sys.stderr)
|
||||
raise SystemExit(1)
|
||||
print(f"TELEMETRY_RECONCILED steps={len(provider)} tolerance_pct={tol_pct}")
|
||||
PY
|
||||
exit $?
|
||||
Reference in New Issue
Block a user