#!/usr/bin/env bash
set -uo pipefail

# CASAN H6 — AgentOps hardening tests (D1 live alerting · D2 provider-telemetry
# API + reconciliation · D3 hosted dashboard · V15 sliding-window breaker).
#
# All checks run LIVE against real HTTP endpoints started locally (webhook sink,
# mock provider usage API, dashboard server) — the same "live" standard as the
# Vault-dev KMS tests. Deterministic: no model needed.

SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
source "$SCRIPT_DIR/../scripts/bash/casan-paths.sh"
PROJECT_ROOT="$(cd "$SCRIPT_DIR/../.." && pwd)"
S="$CASAN_HARNESS_ROOT/scripts/bash"
WORK="$(mktemp -d)"
SINK_PID=""; API_PID=""
cleanup() {
  [[ -n "$SINK_PID" ]] && { kill "$SINK_PID" 2>/dev/null; wait "$SINK_PID" 2>/dev/null; }
  [[ -n "$API_PID" ]] && { kill "$API_PID" 2>/dev/null; wait "$API_PID" 2>/dev/null; }
  CASAN_AGENTOPS_DIR="$WORK/agentops" bash "$S/dashboard-serve.sh" stop >/dev/null 2>&1
  rm -rf "$WORK"
}
trap cleanup EXIT

PASS=0; FAIL=0
pass() { echo "PASS: $1"; PASS=$((PASS + 1)); }
fail() { echo "FAIL: $1"; FAIL=$((FAIL + 1)); }
expect_rc() {
  local want="$1" desc="$2"; shift 2
  local got=0; { "$@" >/dev/null 2>&1; } || got=$?
  [[ "$got" -eq "$want" ]] && pass "$desc (rc=$got)" || fail "$desc (got rc=$got, want $want)"
}

free_port() {
  python - <<'PY'
import socket
s = socket.socket()
s.bind(("127.0.0.1", 0))
print(s.getsockname()[1])
s.close()
PY
}

# ── live webhook sink (records every POST body) ────────────────────────────
cat > "$WORK/sink.py" <<'PY'
import sys
from http.server import BaseHTTPRequestHandler, HTTPServer
port, out = int(sys.argv[1]), sys.argv[2]
class H(BaseHTTPRequestHandler):
    def do_POST(self):
        n = int(self.headers.get("Content-Length", 0))
        body = self.rfile.read(n)
        with open(out, "ab") as f:
            f.write(body + b"\n")
        self.send_response(200)
        self.end_headers()
        self.wfile.write(b'{"ok":true}')
    def log_message(self, *a):
        pass
HTTPServer(("127.0.0.1", port), H).serve_forever()
PY

# ── live mock provider usage API (serves a JSON file on GET) ───────────────
cat > "$WORK/mockapi.py" <<'PY'
import sys
from http.server import BaseHTTPRequestHandler, HTTPServer
port, src = int(sys.argv[1]), sys.argv[2]
class H(BaseHTTPRequestHandler):
    def do_GET(self):
        data = open(src, "rb").read()
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
        self.send_header("Content-Length", str(len(data)))
        self.end_headers()
        self.wfile.write(data)
    def log_message(self, *a):
        pass
HTTPServer(("127.0.0.1", port), H).serve_forever()
PY

wait_http() { # <url> — poll until reachable (any status)
  for _ in 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20; do
    curl -sS -m 2 -o /dev/null "$1" 2>/dev/null && return 0
    sleep 0.2
  done
  return 1
}

export CASAN_AGENTOPS_DIR="$WORK/agentops"
mkdir -p "$CASAN_AGENTOPS_DIR"

echo "===== ① H6 live alert dispatch (webhook, dedup, dead-letter) ====="
SINK_PORT="$(free_port)"
SINK_OUT="$WORK/sink-received.jsonl"
: > "$SINK_OUT"
python "$WORK/sink.py" "$SINK_PORT" "$SINK_OUT" &
SINK_PID=$!
wait_http "http://127.0.0.1:$SINK_PORT/" || true

cat > "$WORK/alert1.json" <<'EOF'
{"timestamp":"2026-07-05T00:00:00Z","trace_id":"t-1","severity":"WARN","resource":{"service.name":"demo.agent","service.version":"1.0.0"},"body":{"message":"Alert triggered: execution-failed","alert.type":"execution-failed","step.name":"write_code"},"attributes":{"latency_ms":100,"status":"failed"}}
EOF

export CASAN_ALERT_WEBHOOK="http://127.0.0.1:$SINK_PORT/hook"
expect_rc 0 "alert is DELIVERED live to the webhook (ALERT_DISPATCHED)" \
  bash "$S/alert-dispatch.sh" "$WORK/alert1.json"
if grep -q '"alert_type": *"execution-failed"' "$SINK_OUT"; then
  pass "webhook sink received the alert payload (severity routed as CRITICAL)"
else
  fail "webhook sink did not receive the alert payload"
fi

BEFORE_COUNT="$(grep -c . "$SINK_OUT" || true)"
OUT2="$(bash "$S/alert-dispatch.sh" "$WORK/alert1.json" 2>&1)"
AFTER_COUNT="$(grep -c . "$SINK_OUT" || true)"
if echo "$OUT2" | grep -q "ALERT_DEDUP_SUPPRESSED" && [[ "$BEFORE_COUNT" == "$AFTER_COUNT" ]]; then
  pass "duplicate alert inside dedup window is SUPPRESSED (no double page)"
else
  fail "duplicate alert was not suppressed (out=$OUT2 before=$BEFORE_COUNT after=$AFTER_COUNT)"
fi

DEAD_PORT="$(free_port)"   # nothing listens here
cat > "$WORK/alert2.json" <<'EOF'
{"timestamp":"2026-07-05T00:01:00Z","trace_id":"t-2","severity":"WARN","resource":{"service.name":"demo.agent","service.version":"1.0.0"},"body":{"message":"Alert triggered: cost-spike","alert.type":"cost-spike","step.name":"plan"},"attributes":{"latency_ms":100,"status":"success"}}
EOF
expect_rc 1 "webhook DOWN + strict => fail-loud (ALERT_DELIVERY_FAILED, no silent loss)" \
  env CASAN_ALERT_WEBHOOK="http://127.0.0.1:$DEAD_PORT/hook" CASAN_ALERT_STRICT=1 \
  bash "$S/alert-dispatch.sh" "$WORK/alert2.json"
if [[ -s "$CASAN_AGENTOPS_DIR/alert-deadletter.jsonl" ]]; then
  pass "undelivered alert queued to dead-letter (not lost)"
else
  fail "dead-letter queue empty after failed delivery"
fi

FLUSH_OUT="$(bash "$S/alert-dispatch.sh" --flush-deadletter 2>&1)"; FLUSH_RC=$?
if [[ "$FLUSH_RC" -eq 0 ]] && echo "$FLUSH_OUT" | grep -q "redelivered=1 remaining=0"; then
  pass "dead-letter flush REDELIVERS the alert once the channel is back"
else
  fail "dead-letter flush failed (rc=$FLUSH_RC out=$FLUSH_OUT)"
fi

# end-to-end: a failing pipeline step pushes a live alert through agent-metrics
printf 'input\n' > "$WORK/in.txt"
: > "$SINK_OUT"
CASAN_AGENT_NAME="h6.e2e" CASAN_STEP_NAME="e2e_fail_step" \
  bash "$S/agent-metrics.sh" "$WORK/in.txt" "$WORK/out.txt" -- bash -c 'exit 3' >/dev/null 2>&1
if grep -q '"step": *"e2e_fail_step"' "$SINK_OUT"; then
  pass "END-TO-END: failing step -> agent-metrics -> live webhook alert received"
else
  fail "end-to-end live alert not received by webhook sink"
fi
unset CASAN_ALERT_WEBHOOK

echo "===== ② H6 provider-telemetry API + reconciliation ====="
cat > "$WORK/usage-valid.json" <<'EOF'
[
 {"provider":"ollama","model":"ornith:9b","run_id":"r1","step":"plan","input_tokens":200,"output_tokens":300,"total_tokens":500,"cost_usd":0.0,"latency_ms":900,"status":"success"},
 {"provider":"ollama","model":"ornith:9b","run_id":"r1","step":"code","input_tokens":400,"output_tokens":600,"total_tokens":1000,"cost_usd":0.0,"latency_ms":1200,"status":"success"}
]
EOF
API_PORT="$(free_port)"
python "$WORK/mockapi.py" "$API_PORT" "$WORK/usage-valid.json" &
API_PID=$!
wait_http "http://127.0.0.1:$API_PORT/usage" || true

FETCH_LOG="$WORK/provider-usage.jsonl"
expect_rc 0 "usage records FETCHED live from provider API (PROVIDER_TELEMETRY_FETCHED)" \
  bash "$S/provider-usage-fetch.sh" "http://127.0.0.1:$API_PORT/usage" "$FETCH_LOG"
[[ "$(grep -c . "$FETCH_LOG" 2>/dev/null)" == "2" ]] \
  && pass "both API records imported with api_endpoint provenance" \
  || fail "expected 2 imported records in $FETCH_LOG"

cat > "$WORK/usage-invalid.json" <<'EOF'
[{"provider":"ollama","model":"ornith:9b","step":"plan","total_tokens":500}]
EOF
kill "$API_PID" 2>/dev/null; wait "$API_PID" 2>/dev/null
python "$WORK/mockapi.py" "$API_PORT" "$WORK/usage-invalid.json" &
API_PID=$!
wait_http "http://127.0.0.1:$API_PORT/usage" || true
BAD_LOG="$WORK/provider-usage-bad.jsonl"
expect_rc 1 "invalid API schema is REJECTED all-or-nothing (PROVIDER_USAGE_INVALID)" \
  bash "$S/provider-usage-fetch.sh" "http://127.0.0.1:$API_PORT/usage" "$BAD_LOG"
[[ ! -s "$BAD_LOG" ]] \
  && pass "nothing imported from an invalid API response (no dirty telemetry)" \
  || fail "invalid response leaked records into $BAD_LOG"

UNREACH_PORT="$(free_port)"
expect_rc 1 "unreachable provider API => fail-loud (PROVIDER_API_UNREACHABLE)" \
  bash "$S/provider-usage-fetch.sh" "http://127.0.0.1:$UNREACH_PORT/usage" "$WORK/never.jsonl"

# reconciliation: local metrics vs provider ground truth
cat > "$WORK/local-ok.jsonl" <<'EOF'
{"step":"plan","total_tokens":495}
{"step":"code","total_tokens":1000}
EOF
expect_rc 0 "local metrics reconcile with provider API (TELEMETRY_RECONCILED)" \
  bash "$S/telemetry-reconcile.sh" "$WORK/local-ok.jsonl" "$FETCH_LOG" 10

cat > "$WORK/local-under.jsonl" <<'EOF'
{"step":"plan","total_tokens":50}
{"step":"code","total_tokens":1000}
EOF
expect_rc 1 "local UNDER-REPORTING (cost-hiding) is caught (TELEMETRY_DISCREPANCY)" \
  bash "$S/telemetry-reconcile.sh" "$WORK/local-under.jsonl" "$FETCH_LOG" 10

echo "===== ③ H6 hosted dashboard (/healthz stale-aware) ====="
DASH_PORT="$(free_port)"
FRESH_METRICS="$WORK/metrics-fresh.jsonl"
printf '{"trace_id":"t","step":"s","status":"success","latency_ms":10,"total_tokens":5,"cost_estimate":0}\n' > "$FRESH_METRICS"
CASAN_DASHBOARD_METRICS="$FRESH_METRICS" bash "$S/dashboard-serve.sh" start "$DASH_PORT" >/dev/null 2>&1
HEALTH="$(curl -sS -m 3 -w '\n%{http_code}' "http://127.0.0.1:$DASH_PORT/healthz" 2>/dev/null)"
if [[ "${HEALTH##*$'\n'}" == "200" ]] && echo "$HEALTH" | grep -q '"status": *"ok"'; then
  pass "dashboard HOSTED: /healthz live returns 200 ok with fresh metrics"
else
  fail "healthz not ok (got: $HEALTH)"
fi
if curl -sS -m 3 "http://127.0.0.1:$DASH_PORT/" 2>/dev/null | grep -q "AgentOps Dashboard"; then
  pass "dashboard HTML is served live over HTTP (no static-file-only)"
else
  fail "dashboard HTML not served over HTTP"
fi
bash "$S/dashboard-serve.sh" stop >/dev/null 2>&1

STALE_METRICS="$WORK/metrics-stale.jsonl"
printf '{"step":"s","total_tokens":5}\n' > "$STALE_METRICS"
touch -t 202601010000 "$STALE_METRICS"
CASAN_DASHBOARD_METRICS="$STALE_METRICS" bash "$S/dashboard-serve.sh" start "$DASH_PORT" >/dev/null 2>&1
HEALTH2="$(curl -sS -m 3 -w '\n%{http_code}' "http://127.0.0.1:$DASH_PORT/healthz" 2>/dev/null)"
if [[ "${HEALTH2##*$'\n'}" == "503" ]] && echo "$HEALTH2" | grep -q '"status": *"stale"'; then
  pass "STALE telemetry => /healthz 503 stale (silent telemetry death is page-able)"
else
  fail "stale metrics not detected by healthz (got: $HEALTH2)"
fi
bash "$S/dashboard-serve.sh" stop >/dev/null 2>&1

echo "===== ④ H6 sliding-window circuit breaker (V15 interleave evasion) ====="
ALT_LOG="$WORK/alt-usage.jsonl"
: > "$ALT_LOG"
for i in 1 2 3 4 5; do
  printf '{"step":"s%s","status":"error","total_tokens":10}\n' "$i" >> "$ALT_LOG"
  printf '{"step":"s%s","status":"success","total_tokens":10}\n' "$i" >> "$ALT_LOG"
done
expect_rc 1 "alternating fail/success EVADES consecutive counter but TRIPS window breaker (CIRCUIT_OPEN_WINDOW)" \
  env CASAN_PROVIDER_LOG="$ALT_LOG" bash "$S/circuit-breaker-check.sh" --breaker-only
grep -q "CIRCUIT_OPEN_WINDOW" <(env CASAN_PROVIDER_LOG="$ALT_LOG" bash "$S/circuit-breaker-check.sh" --breaker-only 2>&1) \
  && pass "window breaker reason is CIRCUIT_OPEN_WINDOW (rate-based, not consecutive)" \
  || fail "window breaker reason missing"

OK_LOG="$WORK/ok-usage.jsonl"
: > "$OK_LOG"
for i in 1 2 3 4 5 6 7 8 9 10; do
  printf '{"step":"s%s","status":"success","total_tokens":10}\n' "$i" >> "$OK_LOG"
done
expect_rc 0 "healthy provider log keeps BOTH breakers closed (no false trip)" \
  env CASAN_PROVIDER_LOG="$OK_LOG" bash "$S/circuit-breaker-check.sh" --breaker-only

echo ""
echo "===== H6 AGENTOPS SUMMARY: PASS=$PASS FAIL=$FAIL ====="
[[ "$FAIL" -eq 0 ]] || exit 1