deploy_check.py

"""Read-only deployment preflight for the Readthrough Hugging Face Space."""

from __future__ import annotations

import sqlite3
import subprocess
import tempfile
from pathlib import Path
from typing import Callable, Iterable

from agent.company_profile import profile_source_coverage


REPO_ROOT = Path(__file__).resolve().parent
METRICS_DB = REPO_ROOT / "data" / "metrics.db"


class GitCommandError(RuntimeError):
    """A local, read-only git inspection command failed."""


def _run_git(args: list[str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["git", *args],
        cwd=REPO_ROOT,
        check=False,
        capture_output=True,
        text=True,
        encoding="utf-8",
        errors="replace",
    )


def git_status_data() -> str:
    """Return porcelain status for data/ without writing LFS objects to .git/."""
    with tempfile.TemporaryDirectory(prefix="readthrough-deploy-check-lfs-") as tmp:
        lfs_storage = Path(tmp)
        (lfs_storage / "tmp").mkdir()
        result = _run_git(
            [
                "-c",
                f"lfs.storage={lfs_storage}",
                "status",
                "--porcelain",
                "--",
                "data/",
            ]
        )
    if result.returncode != 0:
        detail = result.stderr.strip() or result.stdout.strip() or "unknown git error"
        raise GitCommandError(f"git status --porcelain -- data/ failed: {detail}")
    return result.stdout


def parse_data_status(porcelain_output: str) -> list[str]:
    """Extract changed paths from simulated or real porcelain-v1 output."""
    paths: list[str] = []
    for line in porcelain_output.splitlines():
        if len(line) < 4 or line[2] != " ":
            continue
        path = line[3:]
        if " -> " in path:
            path = path.rsplit(" -> ", 1)[1]
        paths.append(path.strip('"'))
    return paths


def load_metric_tickers(db_path: Path = METRICS_DB) -> list[str]:
    """Read distinct metric tickers without allowing SQLite writes."""
    database_uri = f"{db_path.resolve().as_uri()}?mode=ro"
    with sqlite3.connect(database_uri, uri=True) as conn:
        rows = conn.execute(
            """
            SELECT DISTINCT UPPER(TRIM(ticker))
            FROM metrics
            WHERE ticker IS NOT NULL AND TRIM(ticker) <> ''
            ORDER BY 1
            """
        ).fetchall()
    return [row[0] for row in rows]


def collect_coverage(
    tickers: Iterable[str],
    coverage_fn: Callable[[str], dict] | None = None,
) -> tuple[list[dict], list[str]]:
    """Collect coverage rows and tickers that would show the backfill banner."""
    check = coverage_fn or profile_source_coverage
    rows: list[dict] = []
    incomplete: list[str] = []
    for ticker in tickers:
        coverage = check(ticker)
        row = {
            "ticker": ticker,
            "status": coverage.get("status", "UNKNOWN"),
            "business_sections": coverage.get("business_sections", 0),
            "segments_geography_sections": coverage.get(
                "segments_geography_sections", 0
            ),
            "recent_transcripts": coverage.get("recent_transcripts", 0),
        }
        rows.append(row)
        if (
            coverage.get("business_sections_attempted", 0) == 0
            or coverage.get("segments_geography_sections_attempted", 0) == 0
        ):
            incomplete.append(ticker)
    return rows, incomplete


def format_coverage_table(rows: list[dict]) -> str:
    headers = (
        "ticker",
        "status",
        "business_sections",
        "segments_geography_sections",
        "recent_transcripts",
    )
    text_rows = [[str(row.get(header, "")) for header in headers] for row in rows]
    widths = [
        max(len(header), *(len(row[index]) for row in text_rows))
        for index, header in enumerate(headers)
    ]

    def render(values: Iterable[str]) -> str:
        return "  ".join(value.ljust(width) for value, width in zip(values, widths))

    separator = "  ".join("-" * width for width in widths)
    return "\n".join([render(headers), separator, *(render(row) for row in text_rows)])



# ── Runtime cache coverage ────────────────────────────────────────────────────
# The Space serves the committed database, cache tables included. Whatever is
# warm at commit time is what production can answer with, so a cache entry can
# make a feature look healthy while the credential or endpoint it depends on is
# unreachable from the Space. This section makes that surface visible instead of
# incidental — see scripts/warm_runtime_cache.py.

# Feature → cache-key template. {t} is the upper-case ticker.
CACHE_FEATURES: list[tuple[str, str]] = [
    ("valuation",   "VALUATION:v3:{t}"),
    ("events 18m",  "FILING_EVENTS:v1:{t}:18"),
    ("events 36m",  "FILING_EVENTS:v1:{t}:36"),
    ("work capital", "WORKING_CAPITAL:v1:{t}"),
    ("adj. EPS",    "EARNINGS:{t}"),
]

# Secrets the Space needs at runtime. Their absence cannot be detected from
# here — a local .env says nothing about the deployment — so this is a
# checklist, deliberately not a pass/fail gate.
REQUIRED_SPACE_SECRETS: list[tuple[str, str]] = [
    ("ANTHROPIC_API_KEY", "brief generation and Ask AI"),
    ("ALPHAVANTAGE_API_KEY_1", "adjusted EPS, surprise history, transcripts"),
    ("TAVILY_API_KEY", "live news"),
]


def cached_keys(db_path: Path = METRICS_DB) -> set[str]:
    """Every cache key present in the shipped database."""
    if not db_path.exists():
        return set()
    with sqlite3.connect(db_path) as conn:
        tables = {row[0] for row in conn.execute(
            "SELECT name FROM sqlite_master WHERE type='table'"
        )}
        if "av_cache" not in tables:
            return set()
        return {row[0] for row in conn.execute("SELECT cache_key FROM av_cache")}


def format_cache_table(tickers: list[str], keys: set[str]) -> str:
    """A ticker x feature grid of what production will be able to serve."""
    headers = ["ticker"] + [name for name, _ in CACHE_FEATURES]
    widths = [max(len(headers[0]), *(len(t) for t in tickers))] if tickers else [len(headers[0])]
    widths += [len(name) for name, _ in CACHE_FEATURES]

    lines = ["  ".join(h.ljust(w) for h, w in zip(headers, widths))]
    lines.append("  ".join("-" * w for w in widths))
    for ticker in tickers:
        cells = [ticker.ljust(widths[0])]
        for (name, template), width in zip(CACHE_FEATURES, widths[1:]):
            warm = template.format(t=ticker.upper()) in keys
            cells.append(("warm" if warm else "cold").ljust(width))
        lines.append("  ".join(cells))
    return "\n".join(lines)


def resolve_ref(ref: str, *, optional: bool = False) -> str | None:
    result = _run_git(["rev-parse", "--verify", ref])
    if result.returncode == 0:
        return result.stdout.strip()
    if optional:
        return None
    detail = result.stderr.strip() or result.stdout.strip() or "ref not found"
    raise GitCommandError(f"cannot resolve {ref}: {detail}")


def main() -> int:
    failed = False
    print("Readthrough Hugging Face deployment preflight")
    print("============================================")

    try:
        changes = parse_data_status(git_status_data())
    except GitCommandError as exc:
        print(f"\nERROR: {exc}")
        changes = []
        failed = True
    else:
        if changes:
            failed = True
            print("\nERROR: uncommitted data/ changes detected:")
            for path in changes:
                print(f"  - {path}")
            print(
                "The Space will serve the committed data snapshot, not this local "
                "working-tree state."
            )
        else:
            print("\nOK: no uncommitted data/ changes.")

    try:
        tickers = load_metric_tickers()
        coverage_rows, incomplete = collect_coverage(tickers)
    except (OSError, sqlite3.Error) as exc:
        print(f"\nERROR: could not read coverage inputs: {exc}")
        failed = True
    else:
        print("\nCompany Primer source coverage:")
        print(format_coverage_table(coverage_rows))
        if incomplete:
            failed = True
            joined = ", ".join(incomplete)
            print(f"\nERROR: Company Primer coverage incomplete for: {joined}")
            print(
                'Once deployed, the Company Primer will show the "coverage incomplete" '
                "banner for these tickers."
            )
        else:
            print("\nOK: required Company Primer sections were attempted for every ticker.")

    tickers_for_cache = load_metric_tickers()
    if tickers_for_cache:
        keys = cached_keys()
        print("\nRuntime cache shipped in data/metrics.db:")
        print(format_cache_table(tickers_for_cache, keys))
        cold = [
            f"{ticker} {name}"
            for ticker in tickers_for_cache
            for name, template in CACHE_FEATURES
            if template.format(t=ticker.upper()) not in keys
        ]
        if cold:
            print(
                f"\n{len(cold)} feature/ticker pair(s) ship cold. Production will "
                "call the provider live for those, which is fine when the Space "
                "can reach it and empty when it cannot."
            )
            print("Warm them with: python scripts/warm_runtime_cache.py")
        else:
            print("\nOK: every tracked feature ships warm for every ticker.")

    print("\nSpace secrets required at runtime (cannot be verified from here):")
    for name, purpose in REQUIRED_SPACE_SECRETS:
        print(f"  {name:24s} {purpose}")
    print(
        "  Set these under Settings -> Variables and secrets on the Space.\n"
        "  A warm cache will mask a missing key until it expires."
    )

    try:
        local_main = resolve_ref("main")
        hf_main = resolve_ref("hf/main", optional=True)
    except GitCommandError as exc:
        print(f"\nERROR: {exc}")
        failed = True
    else:
        print("\nLocal refs (no network fetch):")
        print(f"  main:    {local_main}")
        if hf_main is None:
            print("  hf/main: not available locally (non-blocking)")
        else:
            print(f"  hf/main: {hf_main}")
            if local_main == hf_main:
                print("  refs:    identical")
            else:
                print("  refs:    different (the local hf/main ref may be stale)")

    if failed:
        print("\nVERDICT: deployment preflight failed; do not push yet.")
        return 1

    print("\nVERDICT: deployment preflight passed.")
    print("Deploy manually with: git push hf main:main")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())