#!/usr/bin/env python3
"""Prowl provider scout hook for Claude Code — opportunistic benchmarking.

Idea: your Claude CLI session is open all day. Between turns, it sits
idle. If you opt in, this hook turns that idle time into a provider
slot — when the model finishes a response, the hook fires, checks for
open benchmark directives on the Prowl network, and (in v0.2) runs
one in a detached subprocess using your existing Claude CLI as the
LLM backend. You earn 70% of the $1 vendors pay; the network gets
fresher scores.

Two modes, gated by env:
  * default (scout)       — polls for directives, writes status, doesn't
                            claim or submit anything. Validates wiring.
  * PROWL_PROVIDER_LIVE=1 — claims one directive per cycle (subject to
                            in-session rate limit), runs the bench via
                            `claude --print`, submits the result, updates
                            earnings in the status file.

Live mode is constrained:
  * Only `platform_profile` template (no creds, no API calls — analyses
    the website + docs only)
  * Only `low` priority directives (no competing for critical paid work)
  * 1 claim per hour per session (lock + state-file timestamp)
  * Hard fail-closed: any error during bench releases the directive
    immediately, so it doesn't get stranded for 24h

Wire it up in `~/.claude/settings.json`:

    {
      "hooks": {
        "Stop": [{
          "hooks": [{
            "type": "command",
            "command": "/abs/path/to/prowl_provider_hook.py",
            "timeout": 5,
            "async": true
          }]
        }]
      }
    }

Required env (in your shell — hooks inherit parent env):

    PROWL_AGENT_KEY=ak_...           # mandatory — anonymous can't be a provider
    PROWL_PROVIDER_MODE=1             # opt-in flag

Optional:

    PROWL_BASE_URL                    # default https://prowl.world
    PROWL_PROVIDER_LIVE=1             # actually claim + run + submit
    PROWL_PROVIDER_PRIORITY=low       # which priorities to scout
                                       # (low|normal|high|critical, comma-sep)
    PROWL_PROVIDER_TEMPLATE=          # filter by template name
                                       # (platform_profile|api_benchmark|...)
    PROWL_PROVIDER_MIN_GAP_SECONDS    # min seconds between live claims
                                       # (default 3600 = 1h)
    PROWL_PROVIDER_CLAUDE_BIN         # path to claude CLI (default 'claude')

Status output: ~/.config/prowl/provider_state.json
  Read at any time with:  cat ~/.config/prowl/provider_state.json
  Tail the log file:      tail -f ~/.config/prowl/provider.log

Contract guarantees:
  * Never blocks the Claude session — always exits 0 within milliseconds
    by detaching the worker to a fully independent subprocess.
  * Never spends LLM tokens in v0.1. v0.2 will require explicit opt-in.
  * Lock file prevents concurrent scouts so a chatty session doesn't
    fan out N parallel polls.
"""
from __future__ import annotations

import json
import os
import sys
import time
import urllib.error
import urllib.request
from pathlib import Path

# ---------------------------------------------------------------------------
# Constants
# ---------------------------------------------------------------------------

DEFAULT_BASE = "https://prowl.world"
HTTP_TIMEOUT = 4.0
STATE_DIR = Path.home() / ".config" / "prowl"
STATE_FILE = STATE_DIR / "provider_state.json"
LOG_FILE = STATE_DIR / "provider.log"
LOCK_FILE = Path("/tmp") / "prowl-provider.lock"
LOCK_STALE_SECONDS = 120  # if the lock is older than this assume it's stale


# ---------------------------------------------------------------------------
# Lightweight logging — append-only, never raises
# ---------------------------------------------------------------------------


def _log(msg: str) -> None:
    try:
        STATE_DIR.mkdir(parents=True, exist_ok=True)
        with open(LOG_FILE, "a") as f:
            f.write(f"{time.strftime('%Y-%m-%d %H:%M:%S')} {msg}\n")
    except Exception:
        pass


# ---------------------------------------------------------------------------
# State file helpers
# ---------------------------------------------------------------------------


def _load_state() -> dict:
    if not STATE_FILE.exists():
        return {}
    try:
        return json.loads(STATE_FILE.read_text())
    except Exception:
        return {}


def _save_state(state: dict) -> None:
    try:
        STATE_DIR.mkdir(parents=True, exist_ok=True)
        tmp = STATE_FILE.with_suffix(".json.tmp")
        tmp.write_text(json.dumps(state, indent=2))
        tmp.replace(STATE_FILE)
    except Exception as exc:
        _log(f"_save_state failed: {exc}")


# ---------------------------------------------------------------------------
# Lock — file-based, PID + timestamp; auto-reclaims if stale
# ---------------------------------------------------------------------------


def _acquire_lock() -> bool:
    """Return True if we got the lock. Otherwise False (another runner is
    already in flight, OR the lock is held by a live process).
    """
    now = time.time()
    if LOCK_FILE.exists():
        try:
            data = json.loads(LOCK_FILE.read_text())
            pid = int(data.get("pid", 0))
            ts = float(data.get("ts", 0))
            age = now - ts
            # Check if process is still alive
            try:
                os.kill(pid, 0)
                alive = True
            except OSError:
                alive = False
            if alive and age < LOCK_STALE_SECONDS:
                return False
        except Exception:
            pass
        # Stale or unreadable — overwrite
        try:
            LOCK_FILE.unlink()
        except Exception:
            pass
    try:
        LOCK_FILE.write_text(json.dumps({"pid": os.getpid(), "ts": now}))
        return True
    except Exception:
        return False


def _release_lock() -> None:
    try:
        LOCK_FILE.unlink()
    except Exception:
        pass


# ---------------------------------------------------------------------------
# Prowl API
# ---------------------------------------------------------------------------


def _api_get(path: str, agent_key: str) -> object | None:
    base = os.environ.get("PROWL_BASE_URL", DEFAULT_BASE)
    url = base.rstrip("/") + path
    req = urllib.request.Request(
        url,
        headers={"X-Agent-Key": agent_key, "User-Agent": "prowl-provider-hook/0.1"},
        method="GET",
    )
    try:
        with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        _log(f"GET {path} → HTTP {exc.code}")
        return None
    except (urllib.error.URLError, TimeoutError, OSError) as exc:
        _log(f"GET {path} → {type(exc).__name__}: {exc}")
        return None


def _api_post(path: str, agent_key: str, body: dict | None = None) -> object | None:
    base = os.environ.get("PROWL_BASE_URL", DEFAULT_BASE)
    url = base.rstrip("/") + path
    data = json.dumps(body or {}).encode("utf-8")
    req = urllib.request.Request(
        url,
        data=data,
        headers={
            "X-Agent-Key": agent_key,
            "Content-Type": "application/json",
            "User-Agent": "prowl-provider-hook/0.1",
        },
        method="POST",
    )
    try:
        with urllib.request.urlopen(req, timeout=HTTP_TIMEOUT) as resp:
            return json.loads(resp.read().decode("utf-8"))
    except urllib.error.HTTPError as exc:
        body_excerpt = ""
        try:
            body_excerpt = exc.read().decode("utf-8", errors="ignore")[:200]
        except Exception:
            pass
        _log(f"POST {path} → HTTP {exc.code}: {body_excerpt}")
        return None
    except (urllib.error.URLError, TimeoutError, OSError) as exc:
        _log(f"POST {path} → {type(exc).__name__}: {exc}")
        return None


# ---------------------------------------------------------------------------
# Scout — the only thing v0.1 does
# ---------------------------------------------------------------------------


def _filter_priority(env_value: str) -> list[str]:
    valid = {"low", "normal", "high", "critical"}
    out: list[str] = []
    for p in (env_value or "low").split(","):
        p = p.strip().lower()
        if p in valid:
            out.append(p)
    return out or ["low"]


def _is_live() -> bool:
    return os.environ.get("PROWL_PROVIDER_LIVE", "").strip() not in ("", "0", "false", "False")


def _min_gap_seconds() -> int:
    try:
        return max(60, int(os.environ.get("PROWL_PROVIDER_MIN_GAP_SECONDS", "3600")))
    except ValueError:
        return 3600


def _scout(agent_key: str) -> None:
    """Poll for available directives; write to state file."""
    state = _load_state()
    started_at = time.time()

    priorities = _filter_priority(os.environ.get("PROWL_PROVIDER_PRIORITY", "low"))
    template_filter = os.environ.get("PROWL_PROVIDER_TEMPLATE", "").strip()

    all_seen: list[dict] = []
    for pri in priorities:
        path = f"/v1/provider/directives?directive_status=open&priority={pri}&limit=5"
        if template_filter:
            path += f"&template={template_filter}"
        result = _api_get(path, agent_key)
        if not isinstance(result, list):
            # API said no — could be 403 (not registered), 401 (bad key), etc.
            # _api_get already logged the failure
            state["last_error"] = f"directives({pri}) returned non-list"
            state["last_check"] = int(started_at)
            _save_state(state)
            return
        all_seen.extend(result)

    # Dedupe by id (priorities can be redundant if env says e.g. "low,normal")
    seen_ids = set()
    unique: list[dict] = []
    for d in all_seen:
        did = d.get("id")
        if did and did not in seen_ids:
            seen_ids.add(did)
            unique.append(d)

    state["last_check"] = int(started_at)
    state["last_check_iso"] = time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime(started_at))
    state["mode"] = "scout"  # v0.1: never actually claims
    state["directives_seen"] = [
        {
            "id": d.get("id"),
            "service_name": d.get("service_name"),
            "priority": d.get("priority"),
            "template": (d.get("instructions") or {}).get("template"),
            "reward_usd": d.get("reward_usd"),
            "expires_at": d.get("expires_at"),
        }
        for d in unique
    ]
    state["would_have_claimed"] = unique[0]["id"] if unique else None
    state.setdefault("history", [])
    state["history"].append({
        "ts": int(started_at),
        "found": len(unique),
        "priorities": priorities,
    })
    state["history"] = state["history"][-50:]  # keep last 50
    state.pop("last_error", None)
    _save_state(state)

    _log(f"scout: found {len(unique)} directive(s) across {priorities}")


# ---------------------------------------------------------------------------
# v0.2 · live mode — claim + bench + submit
# ---------------------------------------------------------------------------


def _gap_ok(state: dict) -> bool:
    """True if enough time has elapsed since the last successful claim."""
    last = state.get("last_claim_at")
    if not last:
        return True
    return (time.time() - float(last)) >= _min_gap_seconds()


def _pick_directive(directives: list[dict]) -> dict | None:
    """Pick a single safe directive for live execution. v0.2 limits:
    platform_profile only, status open, reward >= 0 (i.e. anything)."""
    for d in directives:
        if (d.get("instructions") or {}).get("template") != "platform_profile":
            continue
        if d.get("status") != "open":
            continue
        return d
    return None


def _claim(directive_id: str, agent_key: str) -> dict | None:
    return _api_post(f"/v1/provider/directives/{directive_id}/claim", agent_key)  # type: ignore[return-value]


def _release(directive_id: str, agent_key: str, reason: str) -> None:
    """Release a claimed directive back to the pool. We pass the reason
    purely to log it; the API doesn't take a body field."""
    _api_post(f"/v1/provider/directives/{directive_id}/release", agent_key)
    _log(f"released {directive_id}: {reason}")


def _fetch_public(url: str, timeout: float = 8.0, max_bytes: int = 32 * 1024) -> tuple[int | None, int, str]:
    """Fetch a public URL via stdlib. Returns (status, latency_ms, body_excerpt).
    Body is truncated to `max_bytes` so we don't pass huge HTML blobs to Claude."""
    start = time.monotonic()
    try:
        req = urllib.request.Request(url, headers={"User-Agent": "prowl-provider-hook/0.2"})
        with urllib.request.urlopen(req, timeout=timeout) as resp:
            status = resp.getcode()
            body = resp.read(max_bytes).decode("utf-8", errors="ignore")
            latency_ms = int((time.monotonic() - start) * 1000)
            return status, latency_ms, body
    except urllib.error.HTTPError as exc:
        latency_ms = int((time.monotonic() - start) * 1000)
        try:
            body = exc.read().decode("utf-8", errors="ignore")[:1024]
        except Exception:
            body = ""
        return exc.code, latency_ms, body
    except (urllib.error.URLError, TimeoutError, OSError) as exc:
        latency_ms = int((time.monotonic() - start) * 1000)
        return None, latency_ms, f"<<fetch error: {type(exc).__name__}: {exc}>>"


# System prompt mirrors src/services/templates.py PLATFORM_INTERPRET_SYSTEM
# so the output shape matches what the backend's quality-scorer wants.
_BENCH_SYSTEM_PROMPT = (
    "You are Prowl's platform scorer. Score the given platform on agent-efficiency — "
    "how useful is this platform for AI agents to recommend to users?\n\n"
    "Respond with VALID JSON ONLY. No prose, no markdown fences. Schema:\n"
    "{\n"
    '  "overall": 0-100,\n'
    '  "dimensions": {\n'
    '    "token_efficiency": 0.0-10.0,\n'
    '    "first_try_success": 0.0-10.0,\n'
    '    "response_parseability": 0.0-10.0,\n'
    '    "error_clarity": 0.0-10.0,\n'
    '    "doc_quality": 0.0-10.0,\n'
    '    "auth_simplicity": 0.0-10.0,\n'
    '    "latency": 0.0-10.0,\n'
    '    "consistency": 0.0-10.0\n'
    "  },\n"
    '  "issues": [{"severity":"low|medium|high","detail":"..."}],\n'
    '  "recommendations": ["..."]\n'
    "}\n"
    "Be conservative: if you cannot evaluate a dimension, give 5.0. "
    "Output ONE JSON object and nothing else."
)


def _run_claude_print(user_msg: str) -> str | None:
    """Invoke `claude --print` and capture stdout. Returns None on failure."""
    import subprocess
    bin_path = os.environ.get("PROWL_PROVIDER_CLAUDE_BIN", "claude")
    try:
        proc = subprocess.run(
            [bin_path, "--print", "--system-prompt", _BENCH_SYSTEM_PROMPT, user_msg],
            capture_output=True,
            text=True,
            timeout=180,
        )
        if proc.returncode != 0:
            _log(f"claude --print exit {proc.returncode}: {proc.stderr[:200]}")
            return None
        out = proc.stdout.strip()
        if not out:
            _log("claude --print produced empty stdout")
            return None
        return out
    except FileNotFoundError:
        _log(f"claude CLI not found at '{bin_path}'")
        return None
    except subprocess.TimeoutExpired:
        _log("claude --print timed out after 180s")
        return None
    except Exception as exc:
        _log(f"claude --print failed: {type(exc).__name__}: {exc}")
        return None


def _extract_json(blob: str) -> dict | None:
    """Be liberal about what the model emits — strip optional fences,
    then look for the first { ... } that parses."""
    s = blob.strip()
    # Strip ```json ... ``` if present
    if s.startswith("```"):
        s = s.split("\n", 1)[1] if "\n" in s else s
        if s.endswith("```"):
            s = s.rsplit("```", 1)[0]
    s = s.strip()
    try:
        return json.loads(s)
    except json.JSONDecodeError:
        pass
    # Try to find the largest balanced { ... } substring
    start = s.find("{")
    if start == -1:
        return None
    depth = 0
    for i in range(start, len(s)):
        if s[i] == "{":
            depth += 1
        elif s[i] == "}":
            depth -= 1
            if depth == 0:
                try:
                    return json.loads(s[start : i + 1])
                except json.JSONDecodeError:
                    continue
    return None


def _build_results(
    *,
    service_name: str,
    url: str,
    status: int | None,
    latency_ms: int,
    body: str,
    llm_json: dict,
) -> dict:
    """Wrap the LLM's per-dimension scores with the http-call evidence so
    the backend quality-scorer has both signals."""
    dims = llm_json.get("dimensions") or {}
    # Clamp into [0, 10]
    clean_dims: dict = {}
    for k, v in dims.items():
        try:
            f = float(v)
        except (TypeError, ValueError):
            continue
        clean_dims[k] = max(0.0, min(10.0, f))

    overall = llm_json.get("overall")
    try:
        overall = float(overall) if overall is not None else None
    except (TypeError, ValueError):
        overall = None
    if overall is None and clean_dims:
        # Heuristic fallback: average × 10
        overall = round(sum(clean_dims.values()) / len(clean_dims) * 10, 1)
    overall = max(0.0, min(100.0, float(overall or 0)))

    issues = llm_json.get("issues") or []
    if not isinstance(issues, list):
        issues = []
    recs = llm_json.get("recommendations") or []
    if not isinstance(recs, list):
        recs = []

    return {
        "overall_score": overall,
        "dimensions": clean_dims,
        "issues": issues,
        "recommendations": recs,
        "evidence": {
            "http_calls": [
                {"endpoint": url, "status": status, "latency_ms": latency_ms}
            ],
            "total_tests": 1,
            "passed": 1 if (status and 200 <= status < 400) else 0,
            "body_chars": len(body),
        },
    }


def _submit(directive_id: str, agent_key: str, results: dict) -> dict | None:
    return _api_post(
        "/v1/provider/submit",
        agent_key,
        {"directive_id": directive_id, "results": results},
    )  # type: ignore[return-value]


def _live_cycle(agent_key: str) -> None:
    """One end-to-end claim → bench → submit cycle. Pessimistic about
    every step — releases the directive on any failure path so it doesn't
    get stuck for 24h."""
    state = _load_state()
    if not _gap_ok(state):
        _log(f"live: skipping (gap not met, last claim {state.get('last_claim_at')})")
        return

    # 1) Scout. We pick the safest open directive.
    path = "/v1/provider/directives?directive_status=open&priority=low&limit=10&template=platform_profile"
    listing = _api_get(path, agent_key)
    if not isinstance(listing, list) or not listing:
        _log("live: no eligible directives")
        return
    target = _pick_directive(listing)
    if target is None:
        _log("live: no platform_profile directive in the listing")
        return

    directive_id = target["id"]
    service_name = target.get("service_name", "?")
    service_url = (target.get("instructions") or {}).get("service_url") or ""
    if not service_url:
        _log(f"live: directive {directive_id} has no service_url; skipping")
        return

    # 2) Claim. Backend returns DirectiveClaimResponse with directive_id +
    # status="claimed" on success; logs the HTTP error on failure inside
    # _api_post.
    claim_resp = _claim(directive_id, agent_key)
    if not isinstance(claim_resp, dict) or claim_resp.get("status") != "claimed":
        _log(f"live: claim of {directive_id} failed (resp={claim_resp}); aborting cycle")
        return
    state["last_claim_at"] = time.time()
    _save_state(state)
    _log(f"live: claimed {directive_id} ({service_name})")

    # 3) Bench: fetch + score.
    try:
        status, latency_ms, body = _fetch_public(service_url)
        excerpt = body[:6000]
        user_msg = (
            f"Service name: {service_name}\n"
            f"URL: {service_url}\n"
            f"HTTP status fetched: {status}\n"
            f"Round-trip latency: {latency_ms} ms\n\n"
            f"Page excerpt (UTF-8, truncated to 6KB):\n```\n{excerpt}\n```\n\n"
            "Score this platform per the schema in the system prompt. JSON only."
        )
        llm_out = _run_claude_print(user_msg)
        if llm_out is None:
            _release(directive_id, agent_key, "claude --print returned nothing")
            return
        parsed = _extract_json(llm_out)
        if parsed is None:
            _log(f"live: could not parse LLM JSON (head: {llm_out[:200]!r})")
            _release(directive_id, agent_key, "LLM output not parseable as JSON")
            return

        results = _build_results(
            service_name=service_name,
            url=service_url,
            status=status,
            latency_ms=latency_ms,
            body=body,
            llm_json=parsed,
        )

        # 4) Submit. Backend status is one of {accepted, pending_review}.
        # We treat both as "submitted OK" for bookkeeping; only a missing
        # contribution_id (or HTTP failure → None) is a real submit failure.
        submit_resp = _submit(directive_id, agent_key, results)
        if not isinstance(submit_resp, dict) or "contribution_id" not in submit_resp:
            _release(directive_id, agent_key, f"submit failed: {submit_resp}")
            return
        contrib_status = submit_resp.get("status", "unknown")
        quality = submit_resp.get("quality_score")
        _log(
            f"live: submitted {directive_id} ({service_name}) · "
            f"overall={results['overall_score']:.1f} · "
            f"contribution_status={contrib_status} · "
            f"quality={quality}"
        )

        # 5) Bookkeep. Reward only counts when contribution was auto-accepted
        # (quality >= 50). pending_review contributions earn retroactively.
        state.setdefault("benches_completed", 0)
        state["benches_completed"] += 1
        # The directive's reward_usd is what the provider earns IF the vendor
        # later pays. For low-priority unclaimed directives that's $0 upfront
        # — we track potential, not realised, earnings here.
        earned = float(target.get("reward_usd") or 0.0)
        state.setdefault("total_earned_usd", 0.0)
        state["total_earned_usd"] += earned
        state.setdefault("recent_benches", [])
        state["recent_benches"].append({
            "ts": int(time.time()),
            "directive_id": directive_id,
            "service_name": service_name,
            "overall": results["overall_score"],
            "quality_score": quality,
            "contribution_status": contrib_status,
            "reward_potential_usd": earned,
        })
        state["recent_benches"] = state["recent_benches"][-20:]
        _save_state(state)
    except Exception as exc:
        _log(f"live: unhandled error during bench: {type(exc).__name__}: {exc}")
        _release(directive_id, agent_key, f"unhandled error: {exc}")


# ---------------------------------------------------------------------------
# Detached worker entrypoint
# ---------------------------------------------------------------------------


def _run_detached() -> int:
    """The actual worker body. Runs in a fully independent subprocess so
    the parent (the Claude Stop hook) can exit immediately."""
    if not _acquire_lock():
        _log("lock held — skipping cycle")
        return 0
    try:
        agent_key = os.environ.get("PROWL_AGENT_KEY", "").strip()
        if not agent_key:
            _log("no PROWL_AGENT_KEY set; cannot scout")
            return 0
        _scout(agent_key)
        if _is_live():
            _live_cycle(agent_key)
    finally:
        _release_lock()
    return 0


# ---------------------------------------------------------------------------
# Hook entrypoint (parent process) — read stdin, detach, exit 0 fast
# ---------------------------------------------------------------------------


def _spawn_detached_self() -> None:
    """Re-exec this same script with --detached so the parent (Claude Code
    hook subprocess) returns immediately. The grandchild is fully
    independent: own session, no stdin/stdout/stderr, won't block Claude."""
    import subprocess
    try:
        subprocess.Popen(
            [sys.executable, os.path.abspath(__file__), "--detached"],
            env=os.environ.copy(),
            stdin=subprocess.DEVNULL,
            stdout=subprocess.DEVNULL,
            stderr=subprocess.DEVNULL,
            start_new_session=True,
            close_fds=True,
        )
    except Exception as exc:
        _log(f"spawn failed: {exc}")


def main(argv: list[str]) -> int:
    # Detached mode: do the work and exit.
    if "--detached" in argv:
        return _run_detached()

    # Hook mode: opt-in gate, then spawn detached worker.
    if not os.environ.get("PROWL_PROVIDER_MODE"):
        return 0  # not enabled; no-op
    agent_key = os.environ.get("PROWL_AGENT_KEY", "").strip()
    if not agent_key:
        return 0  # nothing we can do without auth

    # Drain stdin (Claude Code may have sent JSON; we don't need it but
    # closing it cleanly avoids SIGPIPE in some shells).
    try:
        sys.stdin.read()
    except Exception:
        pass

    _spawn_detached_self()
    return 0


if __name__ == "__main__":
    sys.exit(main(sys.argv))