cursor_bridge.py

"""Cursor SDK bridge — shape 4a (fire-and-forget reasoning/drafting).

The lightweight text-in/text-out backend every department loop, chief-agent
tick, and the goal engine call for a single reasoning/drafting call. Kept
deliberately additive: callers should try this bridge first, then whatever
local fallback (e.g. Ollama) they have configured, then a canned stub —
never depend on this bridge alone until a key is live and proven, so a
missing/invalid key degrades gracefully instead of breaking every loop that
depends on it.

Spend enforcement: no governance/spend-mediation layer is assumed to exist
for these calls — each caller is expected to self-enforce its own
``HARD_STOP`` token/dollar dict in-process (see the department-loop
template). This bridge holds itself to that same honest bar: a hard,
in-process per-call token ceiling, clearly labeled as a stand-in for real
spend mediation, not a claim that such mediation already exists.

Usage (library):
    from cursor_bridge import cursor_available, cursor_prompt, CursorBridgeError
    if cursor_available():
        try:
            result = cursor_prompt("...", caller="example_loop")
        except CursorBridgeError:
            ...  # fall back

Usage (CLI diagnostic):
    python3 cursor_bridge.py --selfcheck
    python3 cursor_bridge.py --prompt "hello" --caller manual_test
"""

from __future__ import annotations

import argparse
import json
import os
import sys
import time
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path

ROOT = Path(os.environ.get("PHANTOM_BOX_ROOT", Path.home() / "phantom-box"))
BRIDGE_LOG = ROOT / "logs" / "dogfood" / "cursor_bridge.jsonl"

DEFAULT_MODEL = "auto"
# The SDK's local ``RunResult`` has no ``cost``/``cost_usd`` field at all —
# it returns ``usage: TokenUsage`` (input/output/total tokens), not a
# synchronous dollar figure. A dollar ceiling checked against a field that
# doesn't exist is silent dead code, so the real enforced ceiling here is
# tokens, not dollars. Real $ spend enforcement lives on Cursor's own
# team/org dashboard (Settings -> Usage & Billing spend limits) — that is
# the correct place for it and this bridge cannot substitute for it.
#
# Calibrate this against your own real measured calls once you have some —
# the default below is a reasonable starting point, not a guarantee it fits
# every prompt shape (a call with a lot of retrieved context will use more
# tokens than a trivial one).
MAX_TOKENS_PER_CALL = int(os.environ.get("PHANTOM_BOX_CURSOR_MAX_TOKENS_PER_CALL", "400000"))
DEFAULT_TIMEOUT_S = float(os.environ.get("PHANTOM_BOX_CURSOR_TIMEOUT_S", "60"))

# Model-tier routing: cheaper/faster model for high-frequency low-stakes
# department-loop ticks, best-available for leadership/high-stakes work.
# "strategic" stays on "auto" (Cursor's own best-available routing) so
# leadership/chief reasoning quality is never downgraded by tier selection.
# Adjust the "routine" model id to whatever fast/cheap tier is current when
# you wire this up — check ``Client.list_models()`` for real available ids.
MODEL_TIERS: dict[str, str] = {
    "strategic": DEFAULT_MODEL,  # leadership/chief scripts, high-stakes work — unchanged behavior
    "routine": "claude-haiku-4-5",  # high-frequency, low-stakes dept-loop ticks
}
DEFAULT_TIER = "strategic"

CURSOR_ENV_CANDIDATES = (
    lambda: os.environ.get("PHANTOM_BOX_CURSOR_ENV"),
    lambda: str(Path.home() / ".config" / "phantom-box" / "cursor.env"),
)


class CursorBridgeError(Exception):
    """Raised when a call can't even be attempted, or exceeds this bridge's guardrails."""


@dataclass
class CursorBridgeResult:
    text: str
    model: str  # the model string this bridge requested (tier-resolved or caller-explicit)
    status: str
    duration_s: float
    total_tokens: int | None = None
    input_tokens: int | None = None
    output_tokens: int | None = None
    agent_id: str | None = None
    run_id: str | None = None
    tier: str | None = None  # which tier resolved `model`, if any (None when caller passed an explicit model=)
    resolved_model: str | None = None  # the real model id echoed back by the server (RunResult.model.id) — never fabricated; None if the SDK response didn't carry one


def _utc_now() -> str:
    return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%SZ")


def _load_api_key() -> str | None:
    key = os.environ.get("CURSOR_API_KEY")
    if key and key.strip():
        return key.strip()
    for get_path in CURSOR_ENV_CANDIDATES:
        raw = get_path()
        if not raw:
            continue
        candidate = Path(raw)
        if not candidate.is_file():
            continue
        try:
            lines = candidate.read_text(encoding="utf-8").splitlines()
        except OSError:
            continue
        for line in lines:
            line = line.strip()
            if not line or line.startswith("#") or "=" not in line:
                continue
            k, _, v = line.partition("=")
            if k.strip() == "CURSOR_API_KEY":
                v = v.strip().strip('"').strip("'")
                if v:
                    return v
    return None


def _sdk_importable() -> bool:
    try:
        __import__("cursor_sdk")
    except ImportError:
        return False
    return True


def cursor_available() -> bool:
    """True only when both a real API key and the SDK package are present.

    Never returns True based on assumption — this is what every caller
    should check before attempting ``cursor_prompt()`` so the honest
    "not configured yet" state stays visible in each loop's own audit log.
    """
    return bool(_load_api_key()) and _sdk_importable()


def _append_log(obj: dict) -> None:
    BRIDGE_LOG.parent.mkdir(parents=True, exist_ok=True)
    with BRIDGE_LOG.open("a", encoding="utf-8") as f:
        f.write(json.dumps({"ts": _utc_now(), **obj}, ensure_ascii=False) + "\n")


def cursor_prompt(
    prompt: str,
    *,
    caller: str,
    model: str | None = None,
    tier: str = DEFAULT_TIER,
    cwd: str | None = None,
    timeout_s: float = DEFAULT_TIMEOUT_S,
    mcp_servers: dict | None = None,
) -> CursorBridgeResult:
    """One-shot ``Agent.prompt()`` call (SDK skill Pattern 1). Raises ``CursorBridgeError``
    if the call can't be attempted or executed with a result, never silently degrades to a
    guessed/fabricated response — callers decide their own fallback (Ollama, stub, etc.).

    Model resolution (added 2026-07-25): an explicit ``model=`` always wins (100%
    backward compatible with every existing call site, none of which pass it
    today — verified by inspection, not assumed). Otherwise ``tier`` resolves via
    ``MODEL_TIERS`` (default ``"strategic"`` == ``DEFAULT_MODEL`` == ``"auto"``,
    so every caller that passes neither keeps its exact prior behavior).

    ``mcp_servers``: optional passthrough to ``AgentOptions(mcp_servers=...)``,
    the native MCP wiring ``cursor-sdk`` supports (``StdioMcpServerConfig`` /
    ``HttpMcpServerConfig`` / ``SseMcpServerConfig``). Defaults to ``None`` —
    zero regression for callers that don't pass it. Giving a Cursor call live
    tool-execution access via MCP is a materially larger capability grant
    than this bridge's default text-only scope, so treat it as opt-in
    plumbing: only pass real, proven-safe server configs (e.g. read-only
    filesystem) and be honest about what still needs real credentials before
    a given MCP bridge is more than a stub.
    """
    api_key = _load_api_key()
    if not api_key:
        raise CursorBridgeError("CURSOR_API_KEY not provisioned (checked env + ~/.config/phantom-box/cursor.env)")

    resolved_tier = tier if model is None else None
    effective_model = model if model is not None else MODEL_TIERS.get(tier, MODEL_TIERS[DEFAULT_TIER])

    try:
        from cursor_sdk import Agent, AgentOptions, CursorAgentError, LocalAgentOptions
    except ImportError as exc:
        raise CursorBridgeError(f"cursor_sdk not installed: {exc}") from exc

    agent_options_kwargs: dict = {
        "api_key": api_key,
        "model": effective_model,
        "local": LocalAgentOptions(cwd=cwd or str(ROOT)),
    }
    if mcp_servers:
        agent_options_kwargs["mcp_servers"] = mcp_servers

    start = time.monotonic()
    try:
        result = Agent.prompt(
            prompt,
            AgentOptions(**agent_options_kwargs),
        )
    except CursorAgentError as exc:
        duration = time.monotonic() - start
        _append_log(
            {
                "caller": caller,
                "event": "startup_error",
                "model": effective_model,
                "tier": resolved_tier,
                "duration_s": round(duration, 2),
                "error": str(exc)[:300],
                "retryable": bool(getattr(exc, "is_retryable", False)),
            }
        )
        raise CursorBridgeError(f"Cursor agent failed to start: {exc}") from exc
    duration = time.monotonic() - start

    status = str(getattr(result, "status", "unknown"))
    text = getattr(result, "result", None) or ""
    usage = getattr(result, "usage", None)
    total_tokens = getattr(usage, "total_tokens", None) if usage is not None else None
    input_tokens = getattr(usage, "input_tokens", None) if usage is not None else None
    output_tokens = getattr(usage, "output_tokens", None) if usage is not None else None
    # Real server-echoed model id (RunResult.model.id) — never fabricated;
    # None if the SDK response didn't carry one (e.g. an older SDK build).
    server_model = getattr(result, "model", None)
    resolved_model = getattr(server_model, "id", None) if server_model is not None else None

    _append_log(
        {
            "caller": caller,
            "event": "prompt",
            "model": effective_model,
            "tier": resolved_tier,
            "resolved_model": resolved_model,
            "status": status,
            "duration_s": round(duration, 2),
            "prompt_chars": len(prompt),
            "result_chars": len(text),
            "total_tokens": total_tokens,
            "input_tokens": input_tokens,
            "output_tokens": output_tokens,
            "mcp_server_names": sorted(mcp_servers) if mcp_servers else [],
        }
    )

    if duration > timeout_s:
        # SDK already returned by the time we'd notice this, but flag it —
        # a run that's slow every time is a signal to investigate, not a
        # hard-stop trigger (that's the SDK's own timeout's job).
        _append_log({"caller": caller, "event": "slow_call", "duration_s": round(duration, 2)})

    if isinstance(total_tokens, int) and total_tokens > MAX_TOKENS_PER_CALL:
        _append_log(
            {
                "caller": caller,
                "event": "token_ceiling_exceeded",
                "total_tokens": total_tokens,
                "ceiling_tokens": MAX_TOKENS_PER_CALL,
            }
        )
        raise CursorBridgeError(
            f"Cursor call used {total_tokens} tokens, exceeds per-call ceiling {MAX_TOKENS_PER_CALL} "
            "(real $ enforcement lives on Cursor's team dashboard spend limits, not this bridge)"
        )

    if status == "error":
        raise CursorBridgeError("Cursor run finished with status=error (see logs/dogfood/cursor_bridge.jsonl)")

    return CursorBridgeResult(
        text=text,
        model=effective_model,
        status=status,
        duration_s=duration,
        total_tokens=total_tokens,
        input_tokens=input_tokens,
        output_tokens=output_tokens,
        agent_id=getattr(result, "agent_id", None) or getattr(result, "agentId", None),
        run_id=getattr(result, "id", None) or getattr(result, "run_id", None),
        tier=resolved_tier,
        resolved_model=resolved_model,
    )


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Cursor SDK bridge diagnostic (shape 4a)")
    ap.add_argument("--selfcheck", action="store_true", help="Report configured/not-configured, no call made")
    ap.add_argument("--prompt", default=None, help="Send a one-shot prompt (only if configured)")
    ap.add_argument("--caller", default="manual_cli")
    ap.add_argument("--model", default=None, help="Explicit model id (overrides --tier)")
    ap.add_argument("--tier", default=DEFAULT_TIER, choices=sorted(MODEL_TIERS), help="Model tier (routine|strategic)")
    args = ap.parse_args(argv)

    if args.selfcheck or not args.prompt:
        key_present = bool(_load_api_key())
        sdk_present = _sdk_importable()
        print(
            json.dumps(
                {
                    "cursor_available": key_present and sdk_present,
                    "api_key_present": key_present,
                    "sdk_importable": sdk_present,
                    "max_tokens_per_call": MAX_TOKENS_PER_CALL,
                    "default_model": DEFAULT_MODEL,
                    "model_tiers": MODEL_TIERS,
                    "note": (
                        "Ready — callers can use cursor_prompt()."
                        if key_present and sdk_present
                        else                         "Not configured — callers should fall back to their own "
                        "local model/stub. Provision CURSOR_API_KEY (env or "
                        "~/.config/phantom-box/cursor.env) and `pip install cursor-sdk` "
                        "to enable."
                    ),
                },
                indent=2,
            )
        )
        return 0

    try:
        result = cursor_prompt(args.prompt, caller=args.caller, model=args.model, tier=args.tier)
    except CursorBridgeError as exc:
        print(f"error: {exc}", file=sys.stderr)
        return 1
    print(result.text)
    print(
        json.dumps(
            {
                "requested_model": result.model,
                "tier": result.tier,
                "resolved_model": result.resolved_model,
                "total_tokens": result.total_tokens,
                "duration_s": result.duration_s,
            }
        ),
        file=sys.stderr,
    )
    return 0


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