#!/usr/bin/env python3
"""Tier-1 host drive loop — ADR-070 (v0: single-turn) + ADR-072 (codex / gemini).

A Tier-1 host (Claude Code / Codex / Gemini — ADR-023, host-agent-protocol) is
CLI-drivable: spawn `claude -p "<prompt>" --output-format json`, parse the JSON
envelope, and record the turn. This module is the executor `detectHostTier`
(ADR-068) reported as `tier1-drive-pending`. The rendered prompt comes from
`workspace_render.py` (ADR-069); the caller appends the returned turn to the
session store via `workspace_sessions.py append --kind host.turn`.

Design (AI-council 2026-06-08, claude-sonnet-4-5 + gpt-4o, design mode):

- **Single-turn v0.** One prompt → one host call → one turn record. Tool calls
  in the envelope are **recorded as opaque JSON, never executed** — agentic
  multi-turn (feeding results back) is v1.
- **Explicit envelope contract per host.** Each host config names its
  `required` / `optional` keys; a missing required key (or `is_error: true`)
  **fails closed** — the caller degrades to the Tier-3 inbox hand-off rather
  than fabricating a turn.
- **Unified adapter.** One `drive(host, prompt, …)` with a per-host config
  (arg builder + envelope parser), mirroring the proven `ai-council`
  subprocess shape. All three Tier-1 hosts (`claude-code` / `codex` /
  `gemini`) slot in without touching the control flow.
- **Sync, bounded.** Default 90 s timeout. CLI-missing / non-zero-exit /
  timeout / unrecognised-envelope all return an `ok=False` error turn (with an
  `error_kind`) so the caller records `host.error` and degrades — the user is
  never stuck.
- **`runner` injectable** so tests never spawn a real host CLI.

Deferred to v1 (debt, recorded in ADR-070 / ADR-072): multi-turn agentic loops,
tool-call execution, drive success/timeout **metrics** + kill-switch
observability, error-taxonomy refinement.

CLI::

    workspace_drive.py drive --host <id> --prompt-file <f|-> [--cwd <d>] \
        [--timeout <s>] [--json]
"""

from __future__ import annotations

import argparse
import json
import subprocess
import sys
from pathlib import Path

DEFAULT_TIMEOUT = 90  # seconds (AI-council 2026-06-08: realistic for LLM+tools)


class EnvelopeError(ValueError):
    """The host CLI returned output that does not match its declared contract."""


# --- per-host envelope parsers (stdout → uniform turn) ----------------------
# Each parser owns its host's full stdout shape: a single JSON object
# (claude / gemini) or a newline-delimited event stream (codex). Required keys
# are validated; a missing one raises EnvelopeError → fail-closed. Envelope
# shapes are reused verbatim from the proven `ai_council` CLI clients
# (claude / codex) and a live `gemini --output-format json` probe.

def _turn(*, text, model=None, usage=None, session_id=None, cost_usd=None,
          num_turns=None, tool_calls=None) -> dict:
    return {
        "text": text, "model": model, "usage": usage, "session_id": session_id,
        "cost_usd": cost_usd, "num_turns": num_turns,
        "tool_calls": tool_calls if isinstance(tool_calls, list) else [],
    }


def _parse_claude(stdout: str) -> dict:
    """`claude -p --output-format json` → single JSON envelope.

    Required: ``result`` (assistant text). Truthy ``is_error`` fails closed.
    """
    env = json.loads(stdout)
    if not isinstance(env, dict):
        raise EnvelopeError("envelope is not a JSON object")
    if env.get("is_error") is True:
        raise EnvelopeError(f"host reported is_error: {str(env.get('result', ''))[:200]}")
    if not isinstance(env.get("result"), str):
        raise EnvelopeError("missing required key: result")
    usage = env.get("usage") if isinstance(env.get("usage"), dict) else None
    return _turn(
        text=env["result"], model=env.get("model"), usage=usage,
        session_id=env.get("session_id"), cost_usd=env.get("total_cost_usd"),
        num_turns=env.get("num_turns"),
        tool_calls=env.get("tool_calls"),
    )


def _parse_codex(stdout: str) -> dict:
    """`codex exec --json` → newline-delimited JSON event stream.

    Mirrors `ai_council.clients.OpenAICliClient`: the final assistant text is
    the last ``item.completed`` event's ``item.content[].text``; token usage
    comes from ``turn.completed``; ``session.created`` carries the session id.
    Unknown events are skipped. Required: at least one item with text.
    """
    text = ""
    usage: dict | None = None
    session_id = None
    tool_calls: list = []
    for line in stdout.splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            event = json.loads(line)
        except json.JSONDecodeError:
            continue
        if not isinstance(event, dict):
            continue
        etype = event.get("type")
        if etype == "item.completed":
            item = event.get("item") or {}
            if isinstance(item, dict):
                content = item.get("content") or []
                if isinstance(content, list):
                    chunks = [str(e["text"]) for e in content
                              if isinstance(e, dict) and e.get("text")]
                    if chunks:
                        text = "\n".join(chunks).strip()
                # Tool items are recorded opaquely (never executed).
                if item.get("type") in {"tool_call", "function_call", "command"}:
                    tool_calls.append(item)
        elif etype == "turn.completed":
            u = event.get("usage") or {}
            if isinstance(u, dict):
                usage = {"input_tokens": int(u.get("input_tokens", 0) or 0),
                         "output_tokens": int(u.get("output_tokens", 0) or 0)}
        elif etype == "session.created" and event.get("session_id"):
            session_id = str(event["session_id"])
    if text == "":
        raise EnvelopeError("no item.completed text in codex event stream")
    return _turn(text=text, usage=usage, session_id=session_id, tool_calls=tool_calls)


def _parse_gemini(stdout: str) -> dict:
    """`gemini -p … --output-format json` → single JSON envelope.

    Required: ``response`` (assistant text). Token usage is nested under
    ``stats.models.<model>.tokens`` (model name is dynamic); take the first
    model entry best-effort. ``session_id`` is top-level.
    """
    env = json.loads(stdout)
    if not isinstance(env, dict):
        raise EnvelopeError("envelope is not a JSON object")
    if not isinstance(env.get("response"), str):
        raise EnvelopeError("missing required key: response")
    model = None
    usage = None
    stats = env.get("stats")
    if isinstance(stats, dict) and isinstance(stats.get("models"), dict) and stats["models"]:
        model = next(iter(stats["models"]))
        tokens = stats["models"][model].get("tokens") if isinstance(stats["models"][model], dict) else None
        if isinstance(tokens, dict):
            in_tok = int(tokens.get("prompt", tokens.get("input", 0)) or 0)
            total = int(tokens.get("total", 0) or 0)
            usage = {"input_tokens": in_tok, "output_tokens": max(total - in_tok, 0)}
    return _turn(text=env["response"], model=model, usage=usage,
                 session_id=env.get("session_id"))


HOST_CONFIGS: dict[str, dict] = {
    # All three Tier-1 hosts (ADR-023). Each owns its CLI flags + envelope
    # parser; the unified drive() never changes. Envelope shapes verified
    # against the ai_council CLI clients (claude / codex) + a gemini probe.
    # `build_resume_args` continues a prior session by its host session id
    # (ADR-076, conversation multi-turn) — all three expose a documented
    # non-interactive resume, so `supports_resume` is True for each. Resume
    # invocations verified: claude `--resume <id> -p`, codex `exec resume <id>
    # --json <prompt>`, gemini `--resume <id> -p`.
    "claude-code": {
        "build_args": lambda prompt, cwd: ["claude", "-p", prompt, "--output-format", "json"],
        "build_resume_args": lambda sid, prompt, cwd: ["claude", "--resume", sid, "-p", prompt, "--output-format", "json"],
        "supports_resume": True,
        "parse": _parse_claude,
    },
    "codex": {
        "build_args": lambda prompt, cwd: ["codex", "exec", "--json", prompt],
        "build_resume_args": lambda sid, prompt, cwd: ["codex", "exec", "resume", sid, "--json", prompt],
        "supports_resume": True,
        "parse": _parse_codex,
    },
    "gemini": {
        "build_args": lambda prompt, cwd: ["gemini", "-p", prompt, "--output-format", "json"],
        "build_resume_args": lambda sid, prompt, cwd: ["gemini", "--resume", sid, "-p", prompt, "--output-format", "json"],
        "supports_resume": True,
        "parse": _parse_gemini,
    },
}


def _subprocess_runner(args: list[str], cwd: str | None, timeout: int) -> tuple[int, str, str]:
    """Default runner: spawn the host CLI. Injectable so tests stay hermetic."""
    proc = subprocess.run(
        args, cwd=cwd, capture_output=True, text=True, timeout=timeout, check=False,
    )
    return proc.returncode, proc.stdout, proc.stderr


def _error_turn(host: str, message: str, kind: str) -> dict:
    return {"ok": False, "host": host, "error": message, "error_kind": kind}


# Verified host "resume session not found / expired" stderr signatures
# (ADR-080, probed 2026-06-09): claude `No conversation found with session ID`,
# gemini `Invalid session identifier`, codex `no rollout found for thread id`.
# Substring, case-insensitive — best-effort + extensible as hosts evolve.
SESSION_EXPIRED_SIGNATURES = (
    "no conversation found with session",
    "invalid session identifier",
    "no rollout found for thread",
    "thread/resume failed",
    "session not found",
)


def _is_session_expired(stderr: str | None) -> bool:
    s = (stderr or "").lower()
    return any(sig in s for sig in SESSION_EXPIRED_SIGNATURES)


def drive(
    host: str,
    prompt: str,
    *,
    cwd: str | None = None,
    timeout: int = DEFAULT_TIMEOUT,
    resume_session_id: str | None = None,
    runner=None,
) -> dict:
    """Drive one Tier-1 host turn → a uniform turn record.

    Returns ``{ok: True, host, text, model, usage, session_id, cost_usd,
    num_turns, tool_calls}`` on success, or ``{ok: False, host, error,
    error_kind}`` on any failure (caller records ``host.error`` + degrades to
    the Tier-3 inbox). Never raises for an operational failure.

    When ``resume_session_id`` is given (ADR-076 multi-turn continuation), the
    host's resume invocation is used instead of a fresh launch — the host
    continues its own session. A host without a documented resume → error
    ``resume-unsupported``.
    """
    cfg = HOST_CONFIGS.get(host)
    if cfg is None:
        return _error_turn(host, f"host is not a drivable Tier-1 host in v0: {host}", "unsupported-host")
    if not isinstance(prompt, str) or prompt.strip() == "":
        return _error_turn(host, "prompt is empty", "empty-prompt")

    if resume_session_id is not None:
        if not cfg.get("supports_resume"):
            return _error_turn(host, f"host {host} does not support resume", "resume-unsupported")
        args = cfg["build_resume_args"](resume_session_id, prompt, cwd)
    else:
        args = cfg["build_args"](prompt, cwd)
    run = runner or _subprocess_runner
    try:
        rc, stdout, stderr = run(args, cwd, timeout)
    except subprocess.TimeoutExpired:
        return _error_turn(host, f"host CLI timed out after {timeout}s", "timeout")
    except FileNotFoundError:
        return _error_turn(host, f"host CLI not found: {args[0]}", "cli-missing")
    except OSError as err:
        return _error_turn(host, f"host CLI spawn failed: {err}", "spawn-failed")

    # On a resume, a host whose session has expired / is unknown reports it on
    # stderr (verified messages, ADR-080). Map that to a distinct
    # `session-expired` so the caller can answer 410 (start a new conversation)
    # rather than a generic failure. Only on the resume path.
    if resume_session_id is not None and _is_session_expired(stderr):
        return _error_turn(host, f"host session expired: {(stderr or '').strip()[:200]}", "session-expired")

    if rc != 0:
        return _error_turn(host, f"host CLI exited {rc}: {(stderr or '').strip()[:200]}", "nonzero-exit")

    try:
        turn = cfg["parse"](stdout)
    except (json.JSONDecodeError, EnvelopeError) as err:
        return _error_turn(host, f"unrecognised envelope: {err}", "bad-envelope")

    turn["ok"] = True
    turn["host"] = host
    return turn


def main(argv: list[str] | None = None) -> int:
    p = argparse.ArgumentParser(prog="workspace_drive")
    sub = p.add_subparsers(dest="cmd", required=True)
    d = sub.add_parser("drive")
    d.add_argument("--host", required=True)
    d.add_argument("--prompt-file", required=True, help="path to the rendered prompt, or '-' for stdin")
    d.add_argument("--cwd")
    d.add_argument("--timeout", type=int, default=DEFAULT_TIMEOUT)
    d.add_argument("--resume-session-id", help="continue a prior host session (ADR-076)")
    d.add_argument("--json", action="store_true", help="emit the turn record as JSON")
    args = p.parse_args(argv)

    if args.cmd == "drive":
        prompt = sys.stdin.read() if args.prompt_file == "-" else Path(args.prompt_file).read_text(encoding="utf-8")
        turn = drive(args.host, prompt, cwd=args.cwd, timeout=args.timeout,
                     resume_session_id=args.resume_session_id)
        if args.json:
            print(json.dumps(turn, sort_keys=True))
        elif turn["ok"]:
            sys.stdout.write(turn["text"])
        else:
            print(f"{turn['error_kind']}: {turn['error']}", file=sys.stderr)
        # Exit 1 on a failed drive so the Node caller degrades to the inbox.
        return 0 if turn["ok"] else 1
    return 2


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