Spaces:

s-ttp
/

crucible

Paused

crucible / analyst.py
Structured arguments items 1-2: thesis rests_on + two-pass gate with report context, one bounded repair turn, typed trigger labels, evidence printed in full once, identifier and authored-once audits, relaxed evidence limits, past returns withheld
0b22581 verified about 1 month ago
Raw History Blame Contribute Delete
34.4 kB
"""Crucible — AI Equity Analyst engine wrapper.
Runs the vendored `predict_yearly.py <ticker>` as a subprocess. The engine writes ONE record,
`{ticker}_final.json` = {user, qa}:
  * ``user`` is the public report: either PUBLISHED (the audited HTML and markdown, the rating and
    headline read from the frozen analysis state, and the registered claims) or WITHHELD (a
    plain-language message). This wrapper never computes a rating, a band, a rounding or any other
    financial figure; it stores and serves what the engine audited.
  * ``qa`` is the internal QA record (validation results, rejected sentences, the state snapshot).
    It is written only to the private QA store and served only through the token-gated developer
    route.
The browser receives progress over SSE (stage labels and percentages only): no raw engine log
lines and no persona text, since persona prose is published only after ingestion and audit.
Runs are SERIALIZED (one at a time): the engine's Alpha Vantage rate-limiter is per-process."""
import json
import os
import queue
import re
import subprocess
import threading
import time
import uuid

import config

_run_lock = threading.Lock()  # serialize engine runs (report-file race + AV rate limit)
RUN_TIMEOUT_S = 780.0         # hard cap per run; a hung run past this is killed so the lock frees (the structured
                              # debate adds one bounded repair turn per stage and a second gate pass; the repairs
                              # stop at CRUCIBLE_REPAIR_BUDGET_S, well inside this cap)

ANSI_RE = re.compile(r"\x1b\[[0-9;]*m")

# Engine stdout markers → progress. Labels describe work, never results.
STAGES = [
    (re.compile(r"\[1/5\]"), 4, "Collecting data", "data"),
    (re.compile(r"\[reconcile\]"), 14, "Reconciling reported figures with SEC filings", "data"),
    (re.compile(r"\[data\] "), 20, "Validating the data", "data"),
    (re.compile(r"\[3b/5\]"), 30, "Forecasting fundamentals", "forecast"),
    (re.compile(r"\[story\]|Storyteller|story valuation", re.I), 40, "Building the valuation story", "valuation"),
    (re.compile(r"ML ensemble \(fund\+macro"), 50, "Running the quantitative models", "models"),
    (re.compile(r"\[forecast\] "), 60, "Validating the forecast and valuation", "forecast"),
    (re.compile(r"Analyst Panel Debate"), 66, "The analyst panel is deliberating", "debate"),
    (re.compile(r"Round 2 — Bear \(Risk"), 74, "The bear and bull analysts are making their cases", "debate"),
    (re.compile(r"Round 3 — Bear vs Bull"), 82, "Rebuttals", "debate"),
    (re.compile(r"Round 4 — Portfolio"), 88, "The portfolio manager is weighing the scenarios", "debate"),
    (re.compile(r"\[4/5\]"), 92, "Blending the views", "blend"),
    (re.compile(r"\[5/5\]"), 96, "Composing and checking the report", "report"),
]

_jobs: dict[str, dict] = {}
_current: dict = {"job_id": None, "ticker": None, "started": None}   # who holds the run lock

USER_SAFE_FAILURE = ("The analysis could not be completed. Please try again later.")


# Who has finished (never what they said): a storyteller returned, or a debate seat took its position.
STORYTELLER_RE = re.compile(r"^@@progress storyteller\b")
STORY_START_RE = re.compile(r"^@@progress story-start\b")
SEAT_RE = re.compile(r"^\s+(?:\w+ )?\((Neutral|Bear|Bull|PM|fallback)\)\s+(?:[-+]?\d|p_bear)")


# The framework's inputs for this company, shown while the storytellers work ("under the hood"). Only these
# fields cross, each type-checked and bounded; values, ratings and storyteller views never do.
FACT_FIELDS = {"industry": str, "country": str, "beta": float, "beta_basis": str, "erp": float, "erp_month": str,
               "crp": float, "stage": str, "segments": list, "segment_build": bool, "consensus": bool, "ml_rank": str}


def _facts(payload: str) -> dict:
    try:
        raw = json.loads(payload)
    except (ValueError, TypeError):
        return {}
    out = {}
    for k, typ in FACT_FIELDS.items():
        v = raw.get(k) if isinstance(raw, dict) else None
        if v is None:
            continue
        if typ is float and isinstance(v, (int, float)) and not isinstance(v, bool) and v == v:
            out[k] = round(float(v), 4)
        elif typ is str and isinstance(v, str):
            out[k] = v[:60]
        elif typ is bool and isinstance(v, bool):
            out[k] = v
        elif typ is list and isinstance(v, list):
            out[k] = [str(x)[:40] for x in v if isinstance(x, str)][:6]
    return out


def _classify(line: str, job: dict) -> list[dict]:
    """One engine stdout line → progress events. Nothing from the line itself is forwarded."""
    if STORYTELLER_RE.match(line):
        job["storytellers"] = job.get("storytellers", 0) + 1
        out = []
        if job["pct"] < 40:
            job["pct"] = 40
            out.append({"t": "stage", "pct": 40, "label": "Building the valuation story", "step": "valuation"})
        return out + [{"t": "story", "n": job["storytellers"]}]
    if STORY_START_RE.match(line):
        out = []
        if job["pct"] < 40:
            job["pct"] = 40
            out.append({"t": "stage", "pct": 40, "label": "Building the valuation story", "step": "valuation"})
        f = _facts(line[len("@@progress story-start"):].strip())
        return out + ([{"t": "facts", "facts": f}] if f else [])
    if line.startswith("@@"):
        return []
    for pattern, pct, label, step in STAGES:
        if pattern.search(line) and pct > job["pct"]:
            job["pct"] = pct
            return [{"t": "stage", "pct": pct, "label": label, "step": step}]
    m = SEAT_RE.match(line)
    if m:
        role = {"fallback": "neutral"}.get(m.group(1), m.group(1).lower())
        phase = "done" if role in ("neutral", "pm") else ("rebuttal" if job["pct"] >= 82 else "case")
        return [{"t": "seat", "role": role, "phase": phase}]
    return []


def _read_final(path) -> dict | None:
    try:
        rec = json.loads(path.read_text())
    except (OSError, json.JSONDecodeError, ValueError):
        return None
    if rec.get("schema") != "crucible.final/1" or not isinstance(rec.get("user"), dict):
        return None
    return rec


def _run(job: dict) -> None:
    if not _run_lock.acquire(blocking=False):     # another run in flight → queue behind it
        job["status"] = "queued"
        job["queue"].put({"t": "stage", "pct": 0, "label": "Waiting for another analysis to finish", "step": "queue"})
        _run_lock.acquire()
    proc = None
    watchdog = None
    try:
        job["status"] = "running"
        _current.update(job_id=job.get("id"), ticker=job["ticker"], started=time.time())
        final_path = config.RECOMMENDER_DIR / f"{job['ticker']}_final.json"
        final_path.unlink(missing_ok=True)       # never read a record from a prior run
        proc = subprocess.Popen(
            [config.RECOMMENDER_PYTHON, "predict_yearly.py", job["ticker"], "--llm"],
            cwd=str(config.RECOMMENDER_DIR), stdout=subprocess.PIPE, stderr=subprocess.STDOUT, text=True,
            env={**os.environ, "PYTHONUNBUFFERED": "1"})
        job["proc"] = proc
        # a hung run would hold _run_lock forever; kill it past a hard deadline
        watchdog = threading.Timer(RUN_TIMEOUT_S, lambda: proc.poll() is None and proc.kill())
        watchdog.daemon = True
        watchdog.start()
        assert proc.stdout is not None
        for raw_line in proc.stdout:
            line = ANSI_RE.sub("", raw_line.rstrip("\n"))
            job["lines"].append(line)            # server-side only (private run log)
            for event in _classify(line, job):
                job["events"].append(event)
                job["queue"].put(event)
        proc.wait()
        # read the record BEFORE releasing the lock: a queued same-ticker job unlinks it on start
        rec = _read_final(final_path)
    finally:
        if watchdog is not None:
            watchdog.cancel()
        _current.update(job_id=None, ticker=None, started=None)
        _run_lock.release()
    if rec is None:
        job["status"] = "failed"
        job["user"] = {"schema": "crucible.user_report/1", "status": "FAILED", "ticker": job["ticker"], "message": USER_SAFE_FAILURE}
        _save_private_log(job)
    else:
        job["user"] = scrub_private(rec["user"])
        job["status"] = "done" if job["user"].get("status") == "PUBLISHED" else "withheld"
        qa = dict(rec.get("qa") or {})
        # the engine's own log (last 400 lines) — private QA only, for diagnosing a withheld or odd run
        qa["engine_log_tail"] = [ln for ln in (job.get("lines") or [])[-400:] if not ln.startswith("@@")]
        _save_qa(job["ticker"], qa, job)
        if job["status"] == "done":
            _save_report(job)
    job["queue"].put(None)  # sentinel: stream complete


# ── symbol resolution: accept tickers OR company names, fail clearly on junk ──────
# AV SYMBOL_SEARCH does naive prefix matching ("TSMC" -> Tesmec SpA, "Coca Cola" -> the
# HBC bottler), so names are resolved by an LLM (which knows aliases cold) and the answer
# is then VERIFIED against a real quote — a hallucinated ticker can't slip through.
import os as _os
import urllib.request as _url
import urllib.parse as _uparse

_AV_KEY = _os.getenv("ALPHAVANTAGE_API_KEY") or _os.getenv("ALPHA_VANTAGE_API_KEY", "")
_GEM_KEY = _os.getenv("GEMINI_API_KEY") or _os.getenv("GOOGLE_API_KEY", "")
_quote_cache: dict = {}


def _quote_exists(sym: str):
    """Does this symbol return a real price? True / False / None (= could not tell: provider
    error or rate limit). Only POSITIVE answers are cached — a cached negative from one bad
    Alpha Vantage call rejected SPGI for the rest of the container's life."""
    sym = sym.upper()
    if _quote_cache.get(sym):
        return True
    try:
        u = (f"https://www.alphavantage.co/query?function=GLOBAL_QUOTE"
             f"&symbol={_uparse.quote(sym)}&apikey={_AV_KEY}")
        with _url.urlopen(u, timeout=15) as r:
            payload = json.loads(r.read())
        q = payload.get("Global Quote") or {}
        ok = bool(q.get("05. price"))
        if not ok and (payload.get("Note") or payload.get("Information") or payload.get("Error Message") is None and not payload):
            return None                     # throttled / empty payload: unknown, not "no such stock"
    except Exception:
        return None
    if ok:
        _quote_cache[sym] = True
    return ok


def _llm_ticker(query: str) -> dict | None:
    """Company name/alias -> {'ticker','name'} via Gemini (strict JSON), or None."""
    if not _GEM_KEY:
        return None
    try:
        prompt = (
            "Map this company reference to its primary US-listed stock ticker (NYSE/Nasdaq; "
            "for foreign companies use the US ADR). Reply with STRICT JSON only:\n"
            '{"ticker": "<symbol or null>", "name": "<official company name>"}\n'
            'If it is not a real public company, reply {"ticker": null, "name": ""}.\n\n'
            f"Company reference: {query!r}")
        body = json.dumps({"contents": [{"parts": [{"text": prompt}]}],
                           "generationConfig": {"temperature": 0.0, "maxOutputTokens": 100}}).encode()
        u = ("https://generativelanguage.googleapis.com/v1beta/models/"
             f"gemini-flash-lite-latest:generateContent?key={_GEM_KEY}")
        req = _url.Request(u, data=body, headers={"Content-Type": "application/json"})
        with _url.urlopen(req, timeout=20) as r:
            out = json.loads(r.read())
        txt = out["candidates"][0]["content"]["parts"][0]["text"]
        txt = txt[txt.find("{"): txt.rfind("}") + 1]
        d = json.loads(txt)
        t = (d.get("ticker") or "").strip().upper()
        if t and re.fullmatch(r"[A-Z.\-]{1,6}", t):
            return {"ticker": t, "name": (d.get("name") or "").strip()[:60]}
    except Exception:
        pass
    return None


def resolve_symbol(query: str) -> dict:
    """Ticker or company name -> {'symbol','name','resolved_from'}. Raises ValueError
    with a user-facing message when nothing real matches."""
    q = (query or "").strip()
    if not q or len(q) > 40:
        raise ValueError("Enter a ticker (e.g. NVDA) or a company name (e.g. Nvidia).")
    qU = q.upper()
    if re.fullmatch(r"[A-Z.\-]{1,6}", qU):
        chk = _quote_exists(qU)
        if chk is True:
            return {"symbol": qU, "name": None, "resolved_from": None}
        if chk is None:                     # provider could not answer: accept a ticker-shaped symbol
            return {"symbol": qU, "name": None, "resolved_from": None, "unverified": True}
    # not a live ticker — treat as a company name/alias (this is how "TSMC" -> TSM)
    r = _llm_ticker(q)
    if r and r.get("ticker") and _quote_exists(r["ticker"]) is not False:
        return {"symbol": r["ticker"], "name": r.get("name"), "resolved_from": q}
    raise ValueError(
        f"Couldn't find a US-listed stock for {q!r}. Check the spelling, or enter the "
        f"ticker symbol directly (e.g. TSM for Taiwan Semiconductor / TSMC).")


# ── report cache: persist completed runs for instant replays + the Recent chips ──
# Local dir survives Space sleeps; the private HF dataset repo (s-ttp/crucible-reports)
# survives restarts/redeploys. Both are soft dependencies — no token, no mirror, no crash.
from pathlib import Path as _Path

_REPORTS_DIR = _Path(__file__).resolve().parent / "cache" / "reports"
_REPORTS_DIR.mkdir(parents=True, exist_ok=True)
_REPORTS_REPO = _os.getenv("REPORTS_REPO", "s-ttp/crucible-reports")
# Prefer a dedicated name so it can't collide with other tooling's HF_TOKEN usage.
_HF_TOKEN = _os.getenv("REPORTS_HF_TOKEN") or _os.getenv("HF_TOKEN", "")
_REQLOG = _REPORTS_DIR / "_requests.json"
_CACHE_MAX_AGE_S = 7 * 86400


def _price_now(sym: str):
    """Current price via GLOBAL_QUOTE (used for the staleness flag on cached serves)."""
    try:
        u = (f"https://www.alphavantage.co/query?function=GLOBAL_QUOTE"
             f"&symbol={_uparse.quote(sym.upper())}&apikey={_AV_KEY}")
        with _url.urlopen(u, timeout=15) as r:
            q = json.loads(r.read()).get("Global Quote") or {}
        p = float(q.get("05. price") or 0)
        return p if p > 0 else None
    except Exception:
        return None


# ── privacy: requester metadata never enters a report ─────────────────────────────
# IP address, User-Agent and Referer were stored in every saved report and returned by the
# public /api/report endpoint. They now live only in a private, server-side access log that
# is not served, not mirrored to the report dataset, and pruned after 30 days. Every report
# object is scrubbed on save, on load and before it is served, so reports saved before this
# change cannot leak them either.
_PRIVATE_DIR = _Path(__file__).resolve().parent / "cache" / "private"
_ACCESS_LOG = _PRIVATE_DIR / "access_log.jsonl"
_PRIVATE_KEYS = frozenset({"client", "ip", "ua", "user_agent", "user-agent", "referer", "referrer",
                           "x-forwarded-for", "remote_addr"})
_ACCESS_RETENTION_S = 30 * 86400


def scrub_private(obj):
    """Remove requester metadata anywhere in a report object (recursive, in place)."""
    if isinstance(obj, dict):
        for k in [k for k in obj if str(k).lower() in _PRIVATE_KEYS]:
            obj.pop(k, None)
        for v in obj.values():
            scrub_private(v)
    elif isinstance(obj, list):
        for v in obj:
            scrub_private(v)
    return obj


def log_access(job_id: str, ticker: str, via: str, request) -> None:
    """Private operational record of who started a run (abuse forensics only)."""
    try:
        _PRIVATE_DIR.mkdir(parents=True, exist_ok=True)
        row = {"ts": time.time(), "job_id": job_id, "ticker": ticker, "via": via,
               "ip": (request.headers.get("x-forwarded-for") or (request.client.host if request.client else "") or "")[:64],
               "ua": (request.headers.get("user-agent") or "")[:120],
               "referer": (request.headers.get("referer") or "")[:120]}
        keep = []
        if _ACCESS_LOG.exists():
            now = time.time()
            for line in _ACCESS_LOG.read_text().splitlines():
                try:
                    if now - json.loads(line).get("ts", 0) <= _ACCESS_RETENTION_S:
                        keep.append(line)
                except Exception:
                    continue
        keep.append(json.dumps(row))
        _ACCESS_LOG.write_text("\n".join(keep) + "\n")
    except Exception:
        pass


def _scrub_existing_files() -> None:
    """Rewrite every locally cached report and the ledger without requester metadata."""
    for f in list(_REPORTS_DIR.glob("*.json")):
        try:
            d = json.loads(f.read_text())
            before = json.dumps(d, sort_keys=True)
            scrub_private(d)
            if json.dumps(d, sort_keys=True) != before:
                f.write_text(json.dumps(d))
        except Exception:
            continue
    try:
        if _LEDGER.exists():
            rows = []
            for line in _LEDGER.read_text().splitlines():
                if not line.strip():
                    continue
                try:
                    rows.append(json.dumps(scrub_private(json.loads(line))))
                except Exception:
                    continue
            _LEDGER.write_text("\n".join(rows) + ("\n" if rows else ""))
    except Exception:
        pass


# ── public schema: an explicit whitelist ────────────────────────────────────────────
PUBLIC_REPORT_KEYS = ("schema", "status", "ticker", "company", "as_of", "version", "rating", "headline", "price",
                      "limitations", "html", "markdown", "claims", "message")
PUBLIC_META_KEYS = ("cached_at", "cache_age_days", "stale", "price_now", "source")


def public_view(user: dict, **meta) -> dict:
    """The only shape any public endpoint returns. Unknown keys are dropped, so an internal field
    added upstream can never leak by default."""
    out = {k: user.get(k) for k in PUBLIC_REPORT_KEYS if k in user}
    out.update({k: v for k, v in meta.items() if k in PUBLIC_META_KEYS and v is not None})
    return scrub_private(out)


_QA_DIR = _PRIVATE_DIR / "qa"


def _save_qa(ticker: str, qa: dict, job: dict) -> None:
    """Internal QA: private disk + the private dataset under qa/. Never served publicly."""
    try:
        _QA_DIR.mkdir(parents=True, exist_ok=True)
        rec = scrub_private(dict(qa, job_id=job.get("id"), saved_at=time.time(), user_status=(job.get("user") or {}).get("status")))
        f = _QA_DIR / f"{ticker.upper()}.json"
        f.write_text(json.dumps(rec, default=str))
        if _HF_TOKEN:
            try:
                from huggingface_hub import HfApi
                HfApi().upload_file(path_or_fileobj=str(f), path_in_repo=f"qa/{ticker.upper()}/{int(time.time())}.json",
                                    repo_id=_REPORTS_REPO, repo_type="dataset", token=_HF_TOKEN)
            except Exception:
                pass
    except Exception:
        pass


def _save_private_log(job: dict) -> None:
    """A run that produced no record: keep its log tail privately (disk + private dataset qa/) for diagnosis."""
    try:
        _QA_DIR.mkdir(parents=True, exist_ok=True)
        f = _QA_DIR / f"{job['ticker'].upper()}.failed.log"
        f.write_text("\n".join(ln for ln in (job.get("lines") or [])[-400:] if not ln.startswith("@@")))
        if _HF_TOKEN:
            try:
                from huggingface_hub import HfApi
                HfApi().upload_file(path_or_fileobj=str(f), path_in_repo=f"qa/{job['ticker'].upper()}/{int(time.time())}.failed.log",
                                    repo_id=_REPORTS_REPO, repo_type="dataset", token=_HF_TOKEN)
            except Exception:
                pass
    except Exception:
        pass


def qa_record(ticker: str) -> dict | None:
    """Latest internal QA record for a ticker (developer route only)."""
    f = _QA_DIR / f"{ticker.upper()}.json"
    try:
        return json.loads(f.read_text()) if f.exists() else None
    except Exception:
        return None


def _save_report(job: dict) -> None:
    """Persist a PUBLISHED report (the audited user record) for replays, history and the ledger."""
    try:
        user = job.get("user") or {}
        if user.get("status") != "PUBLISHED":
            return
        payload = dict(public_view(user), cached_at=time.time(), source=job.get("source", "user"))
        f = _REPORTS_DIR / f"{job['ticker']}.json"
        f.write_text(json.dumps(payload))
        _ledger_append(payload)
        if _HF_TOKEN:
            try:
                from huggingface_hub import HfApi
                HfApi().upload_file(path_or_fileobj=str(f), path_in_repo=f"reports/{job['ticker']}.json",
                                    repo_id=_REPORTS_REPO, repo_type="dataset", token=_HF_TOKEN)
            except Exception:
                pass  # mirror is best-effort
    except Exception:
        pass


_LEDGER = _REPORTS_DIR / "_ledger.jsonl"


def _ledger_append(payload: dict) -> None:
    """Append-only record of every call, so realized accuracy can be measured later
    (the 7-day report cache prunes; the ledger never does). Mirrored to the dataset repo."""
    try:
        h = (payload.get("headline") or {}).get("value")
        row = {"ticker": payload["ticker"], "ts": payload["cached_at"],
               "date": dt_date_str(payload["cached_at"]),
               "price": (payload.get("price") or {}).get("value"),
               # percent, as every earlier ledger row; the value is the audited headline, not recomputed
               "expected_return": (round(h * 100, 1) if h is not None else None), "rating": payload.get("rating"),
               "version": payload.get("version"), "source": payload.get("source", "user")}
        if not row["price"] or row["expected_return"] is None:
            return
        with open(_LEDGER, "a") as fh:
            fh.write(json.dumps(row) + "\n")
        if _HF_TOKEN:
            try:
                from huggingface_hub import HfApi
                HfApi().upload_file(path_or_fileobj=str(_LEDGER), path_in_repo="ledger/_ledger.jsonl",
                                    repo_id=_REPORTS_REPO, repo_type="dataset", token=_HF_TOKEN)
            except Exception:
                pass
    except Exception:
        pass


def dt_date_str(ts):
    import datetime as _dt
    return _dt.datetime.fromtimestamp(ts, _dt.timezone.utc).strftime("%Y-%m-%d")


def track_record(min_age_days: int = 30) -> dict:
    """Realized accuracy of past calls: for ledger rows older than min_age_days, compare the
    predicted direction/return with the price move since. Empty until the ledger ages."""
    rows = []
    try:
        if _LEDGER.exists():
            rows = [json.loads(l) for l in _LEDGER.read_text().splitlines() if l.strip()]
    except Exception:
        rows = []
    now = time.time()
    total = len(rows)
    # Readings at the horizons that matter for a 12-month call: 30d is an early sanity check,
    # 365d is the one the headline is actually calibrated against.
    horizons = {}
    for h in (30, 90, 180, 365):
        _aged = [r for r in rows if now - r["ts"] >= h * 86400]
        if not _aged:
            horizons[str(h)] = {"evaluated": 0}
            continue
        _latest = {}
        for r in _aged:
            if r["ticker"] not in _latest or r["ts"] > _latest[r["ticker"]]["ts"]:
                _latest[r["ticker"]] = r
        _hits = 0; _errs = []; _n = 0
        for r in _latest.values():
            p1 = _price_now(r["ticker"])
            if not p1:
                continue
            real = p1 / r["price"] - 1.0
            pred = (r["expected_return"] or 0) / 100.0
            # scale the 12-month call to the horizon so early readings aren't judged on the full-year number
            pred_h = pred * min(1.0, h / 365.0)
            ok = (real >= 0) == (pred_h >= 0) if abs(pred_h) >= 0.02 else abs(real) < 0.05
            _hits += 1 if ok else 0; _errs.append(abs(real - pred_h)); _n += 1
        horizons[str(h)] = {"evaluated": _n,
                            "direction_hit_rate": round(_hits / _n, 3) if _n else None,
                            "mean_abs_error_pct": round(sum(_errs) / _n * 100, 1) if _n else None}
    aged = [r for r in rows if now - r["ts"] >= min_age_days * 86400]
    if not aged:
        return {"calls_logged": total, "evaluated": 0, "min_age_days": min_age_days, "horizons": horizons,
                "note": f"track record builds automatically — first reading when calls are {min_age_days} days old"}
    # dedupe to latest aged call per ticker to avoid overweighting repeats
    latest = {}
    for r in aged:
        if r["ticker"] not in latest or r["ts"] > latest[r["ticker"]]["ts"]:
            latest[r["ticker"]] = r
    hits = 0; errs = []; det = []
    for r in latest.values():
        p1 = _price_now(r["ticker"])
        if not p1:
            continue
        real = p1 / r["price"] - 1.0
        pred = (r["expected_return"] or 0) / 100.0
        ok = (real >= 0) == (pred >= 0) if abs(pred) >= 0.02 else abs(real) < 0.05
        hits += 1 if ok else 0
        errs.append(abs(real - pred))
        det.append({"ticker": r["ticker"], "date": r["date"], "rating": r["rating"],
                    "predicted": round(pred * 100, 1), "realized_to_date": round(real * 100, 1), "direction_ok": ok})
    n = len(det)
    return {"calls_logged": total, "evaluated": n, "min_age_days": min_age_days, "horizons": horizons,
            "direction_hit_rate": round(hits / n, 3) if n else None,
            "mean_abs_error_pct": round(sum(errs) / n * 100, 1) if n else None,
            "detail": sorted(det, key=lambda x: x["date"])[-25:]}


def _hydrate_reports() -> None:
    """On startup, pull recent reports AND the call ledger from the dataset repo so the cache and
    the track record survive redeploys (the landing page read "0 calls logged" after each rebuild)."""
    if not _HF_TOKEN:
        return
    try:
        from huggingface_hub import HfApi, hf_hub_download
        api = HfApi()
        try:
            if not _LEDGER.exists() or _LEDGER.stat().st_size == 0:
                got = hf_hub_download(_REPORTS_REPO, "ledger/_ledger.jsonl", repo_type="dataset", token=_HF_TOKEN)
                _LEDGER.parent.mkdir(parents=True, exist_ok=True)
                _LEDGER.write_text(open(got).read())
        except Exception:
            pass
        for fp in api.list_repo_files(_REPORTS_REPO, repo_type="dataset", token=_HF_TOKEN):
            if not (fp.startswith("reports/") and fp.endswith(".json")):
                continue
            local = _REPORTS_DIR / _Path(fp).name
            if local.exists():
                continue
            try:
                got = hf_hub_download(_REPORTS_REPO, fp, repo_type="dataset", token=_HF_TOKEN)
                d = scrub_private(json.loads(open(got).read()))
                if time.time() - (d.get("cached_at") or 0) <= _CACHE_MAX_AGE_S:
                    local.write_text(json.dumps(d))
            except Exception:
                continue
    except Exception:
        pass
    finally:
        _scrub_existing_files()


def _startup_maintenance() -> None:
    _scrub_existing_files()          # always, even without a dataset token
    _hydrate_reports()


threading.Thread(target=_startup_maintenance, daemon=True).start()


def _log_request(ticker: str) -> None:
    try:
        d = json.loads(_REQLOG.read_text()) if _REQLOG.exists() else {}
        now = time.time()
        d[ticker] = [t for t in (d.get(ticker) or []) if now - t < _CACHE_MAX_AGE_S] + [now]
        _REQLOG.write_text(json.dumps(d))
    except Exception:
        pass


def _request_counts() -> dict:
    try:
        d = json.loads(_REQLOG.read_text()) if _REQLOG.exists() else {}
        now = time.time()
        return {t: len([x for x in v if now - x < _CACHE_MAX_AGE_S]) for t, v in d.items()}
    except Exception:
        return {}


def _hydrate_one(ticker: str) -> bool:
    """Pull ONE ticker's report from the dataset repo on demand. After a rebuild the container
    starts with an empty cache and the background hydration takes minutes; without this, a
    Recent-chip click in that window found nothing and the client fell back to a full live run
    (observed: an unrequested AAPL run after every deploy today)."""
    if not _HF_TOKEN:
        return False
    try:
        from huggingface_hub import hf_hub_download
        got = hf_hub_download(_REPORTS_REPO, f"reports/{ticker.upper()}.json", repo_type="dataset", token=_HF_TOKEN)
        d = scrub_private(json.loads(open(got).read()))
        if time.time() - (d.get("cached_at") or 0) <= _CACHE_MAX_AGE_S:
            _REPORTS_DIR.mkdir(parents=True, exist_ok=True)
            (_REPORTS_DIR / f"{ticker.upper()}.json").write_text(json.dumps(d))
            return True
    except Exception:
        pass
    return False


def _stored(ticker: str) -> dict | None:
    f = _REPORTS_DIR / f"{ticker.upper()}.json"
    if not f.exists() and not _hydrate_one(ticker):
        return None
    try:
        d = scrub_private(json.loads(f.read_text()))
    except Exception:
        return None
    # only audited reports from the current pipeline are served; older records predate the audit
    if d.get("schema") != "crucible.user_report/1" or d.get("status") != "PUBLISHED" or not d.get("html"):
        return None
    if time.time() - (d.get("cached_at") or 0) > _CACHE_MAX_AGE_S:
        return None
    return d


def cached_report(ticker: str) -> dict | None:
    """Fresh (<7d) published report with a price-drift staleness note."""
    d = _stored(ticker)
    if d is None:
        return None
    age = time.time() - (d.get("cached_at") or 0)
    meta = {"cached_at": d.get("cached_at"), "cache_age_days": round(age / 86400, 1), "source": d.get("source")}
    p0 = (d.get("price") or {}).get("value")
    p1 = _price_now(ticker) if p0 else None
    if p0 and p1:
        drift = p1 / p0 - 1.0
        meta["price_now"] = p1
        if abs(drift) > 0.07:
            meta["stale"] = "The share price has moved materially since this analysis; a new analysis is recommended."
    return public_view(d, **meta)


def cached_markdown(ticker: str) -> str | None:
    """The stored, audited markdown — the same fragments as the HTML page."""
    d = _stored(ticker)
    return d.get("markdown") if d else None


def recent_reports(limit: int = 8) -> list[dict]:
    """Latest PUBLISHED run per ticker (<=7d), ranked by request count then recency."""
    counts = _request_counts()
    rows = []
    for f in _REPORTS_DIR.glob("*.json"):
        if f.name.startswith("_"):
            continue
        try:
            d = json.loads(f.read_text())
        except Exception:
            continue
        if d.get("schema") != "crucible.user_report/1" or d.get("status") != "PUBLISHED":
            continue
        age = time.time() - (d.get("cached_at") or 0)
        if age > _CACHE_MAX_AGE_S:
            continue
        rows.append({"ticker": d["ticker"], "company": d.get("company"), "rating": d.get("rating"),
                     "headline": (d.get("headline") or {}).get("rendered"),
                     "age_days": round(age / 86400, 1), "source": d.get("source", "user"),
                     "requests": counts.get(d["ticker"], 0)})
    rows.sort(key=lambda r: (-r["requests"], r["age_days"]))
    return rows[:limit]


def start(ticker: str) -> str:
    ticker = ticker.upper().strip()
    if not re.fullmatch(r"[A-Z.\-]{1,6}", ticker):
        raise ValueError(f"invalid ticker {ticker!r}")
    # Same ticker already in flight (running or queued): attach to that job instead of
    # spending a second engine run on an identical request (double-click, page reload).
    for _j in list(_jobs.values()):
        if _j.get("ticker") == ticker and _j.get("status") in ("running", "queued") \
                and time.time() - (_j.get("created") or 0) < 1800:
            return _j["id"]
    job_id = uuid.uuid4().hex[:12]
    job = {
        "ticker": ticker,
        "status": "running",
        "lines": [],
        "events": [],
        "pct": 0,
        "queue": queue.Queue(),
        "user": None,
    }
    job["id"] = job_id; job["created"] = time.time()
    _jobs[job_id] = job
    threading.Thread(target=_run, args=(job,), daemon=True).start()
    return job_id


def jobs_view() -> dict:
    """Observability: who holds the engine, who is waiting, how long."""
    now = time.time()
    cur = dict(_current)
    if cur.get("started"):
        cur["age_s"] = int(now - cur["started"])
    queued = [{"job_id": j, "ticker": v["ticker"], "waiting_s": int(now - v.get("created", now))}
              for j, v in _jobs.items() if v["status"] == "queued"]
    recent = sorted(((v.get("created", 0), j, v["ticker"], v["status"]) for j, v in _jobs.items()), reverse=True)[:10]
    return {"running": cur, "queued": queued,
            "recent": [{"job_id": j, "ticker": t, "status": st} for _, j, t, st in recent]}


def exists(job_id: str) -> bool:
    return job_id in _jobs


def stream(job_id: str):
    """SSE progress for a job; replays history first. Stage labels and percentages only."""
    job = _jobs[job_id]
    replayed = len(job["events"])
    for event in list(job["events"]):
        yield f"data: {json.dumps(event)}\n\n"
    while job["status"] in ("queued", "running"):
        try:
            event = job["queue"].get(timeout=30)
        except queue.Empty:
            yield ": keepalive\n\n"
            continue
        if event is None:
            break
        if replayed:
            replayed -= 1
            continue
        yield f"data: {json.dumps(event)}\n\n"
    yield f"event: done\ndata: {json.dumps({'status': job['status']})}\n\n"


def result(job_id: str) -> dict:
    job = _jobs[job_id]
    if job["status"] in ("queued", "running"):
        return {"schema": "crucible.user_report/1", "status": job["status"].upper(), "ticker": job["ticker"]}
    return public_view(job.get("user") or {"status": "FAILED", "ticker": job["ticker"], "message": USER_SAFE_FAILURE},
                       source=job.get("source"))