"""Stop and reconcile factory runs via the global trace database.

Process identity uses the kernel start ticks recorded at spawn. Rows created
before start-tick recording use the command identity contract below.

Command identity contract (names are referenced in tests):

``recorded_command_argv`` splits the stored ``processes.command`` string into
tokens (``shlex``). Path-like tokens are resolved to absolute paths for
comparison; other tokens are preserved verbatim. ``proc_argv_for_match`` applies
the same normalization to live ``/proc/<pid>/cmdline`` argv.

``argv_matches_recorded_subsequence`` accepts a live process only when the
recorded argv equals the normalized proc argv, OR the recorded argv appears as
a *contiguous subsequence* of the proc argv. A known Python interpreter prefix
(``[interpreter, script_path]`` as exact contiguous tokens) may be stripped
from the live argv before comparison — path basenames are never used.
"""

from __future__ import annotations

import argparse
import json
import os
import select
import shlex
import signal
import sqlite3
import sys
import time
from dataclasses import dataclass, field
from pathlib import Path

from .agents import load_config
from .tracer import Tracer
from .utils import ensure_dir, now_iso

STOP_TERM_WAIT_S = 2.0
POLL_INTERVAL_MS = 50


@dataclass
class ProcessRow:
    id: int
    kind: str
    name: str
    pid: int
    start_ticks: int | None
    command: str


@dataclass
class StopResult:
    adw_id: str
    ok: bool
    exit_code: int
    message: str
    mismatches: list[str] = field(default_factory=list)
    signaled_pids: list[int] = field(default_factory=list)


def _is_path_like(token: str) -> bool:
    return "/" in token or token.startswith(".")


def _resolve_token(token: str) -> str:
    if not _is_path_like(token):
        return token
    try:
        return str(Path(token).resolve())
    except OSError:
        return token


def _is_python_interpreter(token: str) -> bool:
    name = Path(token).name
    return name == "python" or name.startswith("python3")


def recorded_command_argv(recorded_command: str) -> list[str]:
    """Tokenize a stored ``processes.command`` value; malformed input is safe."""
    text = recorded_command.strip()
    if not text:
        return []
    try:
        return shlex.split(text)
    except ValueError:
        return []


def command_tokens_for_match(tokens: list[str]) -> list[str]:
    """Normalize argv/command tokens for identity comparison."""
    return [_resolve_token(token) for token in tokens]


def proc_argv_for_match(proc_argv: list[str]) -> list[str]:
    """Normalize live ``/proc/<pid>/cmdline`` argv for identity comparison."""
    return command_tokens_for_match(proc_argv)


def _proc_argv_match_candidates(proc_argv: list[str]) -> list[list[str]]:
    """Live argv forms to compare against a recorded identity."""
    resolved = proc_argv_for_match(proc_argv)
    candidates = [resolved]
    if len(proc_argv) >= 2 and _is_python_interpreter(proc_argv[0]) and _is_path_like(proc_argv[1]):
        stripped = command_tokens_for_match(proc_argv[2:])
        if stripped != resolved:
            candidates.append(stripped)
    return candidates


def _recorded_live_pairs(recorded: list[str], live: list[str]) -> list[tuple[list[str], list[str]]]:
    """Recorded/live token lists to compare, including interpreter-tail alignment."""
    pairs = [(recorded, live)]
    if (
        recorded
        and live
        and _is_python_interpreter(recorded[0])
        and _is_python_interpreter(live[0])
    ):
        pairs.append((recorded[1:], live[1:]))
    return pairs


def _subsequence_match(recorded: list[str], live: list[str]) -> bool:
    if live == recorded:
        return True
    m = len(recorded)
    if m > len(live):
        return False
    for i in range(len(live) - m + 1):
        if live[i:i + m] == recorded:
            return True
    return False


def argv_matches_recorded_subsequence(proc_argv: list[str], recorded_command: str) -> bool:
    """True when ``recorded_command`` matches live argv under the subsequence contract."""
    recorded = command_tokens_for_match(recorded_command_argv(recorded_command))
    if not recorded:
        return False
    for live in _proc_argv_match_candidates(proc_argv):
        for recorded_cmp, live_cmp in _recorded_live_pairs(recorded, live):
            if _subsequence_match(recorded_cmp, live_cmp):
                return True
    return False


def read_proc_argv(pid: int) -> list[str] | None:
    """Read ``/proc/<pid>/cmdline`` as argv; ``None`` when the pid is absent."""
    path = Path(f"/proc/{pid}/cmdline")
    if not path.exists():
        return None
    raw = path.read_bytes()
    if not raw:
        return None
    parts = [part.decode(errors="replace") for part in raw.split(b"\0")]
    while parts and not parts[-1]:
        parts.pop()
    return parts or None


def read_proc_start_ticks(pid: int) -> int | None:
    """Read kernel process start ticks from field 22 of ``/proc/<pid>/stat``."""
    try:
        line = Path(f"/proc/{pid}/stat").read_text()
    except (OSError, UnicodeError):
        return None
    close = line.rfind(")")
    if close < 0:
        return None
    fields = line[close + 1:].split()
    if len(fields) <= 19:
        return None
    try:
        return int(fields[19])
    except ValueError:
        return None


def pid_is_alive(pid: int) -> bool:
    status_path = Path(f"/proc/{pid}/status")
    if not status_path.exists():
        return False
    try:
        for line in status_path.read_text().splitlines():
            if line.startswith("State:"):
                return line.split()[1] != "Z"
    except OSError:
        return False
    try:
        os.kill(pid, 0)
    except ProcessLookupError:
        return False
    except PermissionError:
        return True
    return True


def _poll_until(deadline: float) -> None:
    remaining_ms = max(0, int((deadline - time.monotonic()) * 1000))
    if remaining_ms:
        select.poll().poll(min(remaining_ms, POLL_INTERVAL_MS))


def wait_pids_exit(pids: list[int], deadline: float) -> set[int]:
    """Return pids still alive at ``deadline``, using ``poll`` rather than bare sleep."""
    surviving = {pid for pid in pids if pid_is_alive(pid)}
    while surviving and time.monotonic() < deadline:
        for pid in list(surviving):
            if not pid_is_alive(pid):
                surviving.discard(pid)
        if surviving:
            _poll_until(deadline)
    return surviving


def signal_if_verified(pid: int, recorded_command: str, sig: signal.Signals,
                       *, start_ticks: int | None = None) -> bool:
    """Re-read process identity and signal only when it still matches."""
    if start_ticks is not None:
        if read_proc_start_ticks(pid) != start_ticks:
            return False
        os.kill(pid, sig)
        return True
    proc_argv = read_proc_argv(pid)
    if proc_argv is None:
        return False
    if not argv_matches_recorded_subsequence(proc_argv, recorded_command):
        return False
    os.kill(pid, sig)
    return True


def open_trace_db(db_path: str | Path) -> sqlite3.Connection:
    """Open the trace db with WAL pragmas and ensure schema exists."""
    return _open_tracer(db_path).conn


def _open_tracer(db_path: str | Path) -> Tracer:
    path = Path(db_path)
    ensure_dir(path.parent)
    return Tracer(path, path.parent / ".factory-control-events.jsonl")


def _cancel_decisions_for_inactive_sessions(tracer: Tracer) -> int:
    """Cancel pending decisions whose session is no longer live."""
    rows = tracer.conn.execute(
        "SELECT DISTINCT d.adw_id FROM decisions d"
        " JOIN sessions s ON s.adw_id = d.adw_id"
        " WHERE d.status='pending'"
        " AND (s.status != 'running' OR s.ended_at IS NOT NULL)",
    ).fetchall()
    total = 0
    for (adw_id,) in rows:
        total += tracer.decisions_cancel(adw_id)
    return total


def _load_session(conn: sqlite3.Connection, adw_id: str) -> tuple[str, str | None] | None:
    row = conn.execute(
        "SELECT status, ended_at FROM sessions WHERE adw_id=?",
        (adw_id,),
    ).fetchone()
    if row is None:
        return None
    return row[0], row[1]


def _load_live_processes(conn: sqlite3.Connection, adw_id: str) -> list[ProcessRow]:
    rows = conn.execute(
        "SELECT id, kind, name, pid, start_ticks, command FROM processes"
        " WHERE adw_id=? AND ended_at IS NULL"
        " ORDER BY CASE kind WHEN 'agent' THEN 0 WHEN 'adw' THEN 1 ELSE 2 END, id",
        (adw_id,),
    ).fetchall()
    return [ProcessRow(*row) for row in rows]


def _close_process_ids(conn: sqlite3.Connection, process_ids: list[int]) -> None:
    if not process_ids:
        return
    ts = now_iso()
    placeholders = ",".join("?" * len(process_ids))
    conn.execute(
        f"UPDATE processes SET ended_at=? WHERE id IN ({placeholders}) AND ended_at IS NULL",
        (ts, *process_ids),
    )


def finalize_session_failed(conn: sqlite3.Connection, adw_id: str) -> None:
    """Atomically close every open process row and mark the session failed."""
    ts = now_iso()
    conn.execute("BEGIN IMMEDIATE")
    try:
        conn.execute(
            "UPDATE processes SET ended_at=? WHERE adw_id=? AND ended_at IS NULL",
            (ts, adw_id),
        )
        conn.execute(
            "UPDATE sessions SET status='fail', ended_at=? WHERE adw_id=?",
            (ts, adw_id),
        )
        conn.execute("COMMIT")
    except sqlite3.Error:
        conn.execute("ROLLBACK")
        raise


def stop_run(db_path: str | Path, adw_id: str) -> StopResult:
    tracer = _open_tracer(db_path)
    conn = tracer.conn
    try:
        session = _load_session(conn, adw_id)
        if session is None:
            return StopResult(
                adw_id=adw_id,
                ok=False,
                exit_code=2,
                message=f"unknown adw_id {adw_id} — no session row in trace db",
            )

        status, _ended_at = session
        live_rows = _load_live_processes(conn, adw_id)

        if not live_rows:
            if status != "running":
                tracer.decisions_cancel(adw_id)
                return StopResult(
                    adw_id=adw_id,
                    ok=True,
                    exit_code=0,
                    message=f"adw_id {adw_id} already stopped (status={status})",
                )
            finalize_session_failed(conn, adw_id)
            tracer.decisions_cancel(adw_id)
            return StopResult(
                adw_id=adw_id,
                ok=True,
                exit_code=0,
                message=f"adw_id {adw_id} had no live processes; session marked fail",
            )

        verified: list[ProcessRow] = []
        absent_ids: list[int] = []
        mismatches: list[str] = []

        for row in live_rows:
            if row.start_ticks is not None:
                live_start_ticks = read_proc_start_ticks(row.pid)
                if live_start_ticks == row.start_ticks:
                    verified.append(row)
                else:
                    mismatches.append(
                        f"pid {row.pid} ({row.kind}/{row.name}): "
                        f"start_ticks recorded={row.start_ticks} live={live_start_ticks}"
                    )
                continue
            proc_argv = read_proc_argv(row.pid)
            if proc_argv is None:
                absent_ids.append(row.id)
                continue
            if argv_matches_recorded_subsequence(proc_argv, row.command):
                verified.append(row)
            else:
                mismatches.append(
                    f"pid {row.pid} ({row.kind}/{row.name}): recorded={row.command!r} "
                    f"live_argv={proc_argv_for_match(proc_argv)!r}"
                )

        if absent_ids:
            _close_process_ids(conn, absent_ids)

        if mismatches and not verified:
            return StopResult(
                adw_id=adw_id,
                ok=False,
                exit_code=1,
                message="refusing to signal: every live process failed identity check",
                mismatches=mismatches,
            )

        child_rows = [row for row in verified if row.kind != "adw"]
        parent_rows = [row for row in verified if row.kind == "adw"]

        signaled: list[int] = []
        surviving: set[int] = set()
        if child_rows:
            child_signaled, child_surviving = _terminate_verified_rows(
                child_rows, STOP_TERM_WAIT_S,
            )
            signaled.extend(child_signaled)
            surviving.update(child_surviving)
        if parent_rows:
            parent_signaled, parent_surviving = _terminate_verified_rows(
                parent_rows, STOP_TERM_WAIT_S,
            )
            signaled.extend(parent_signaled)
            surviving.update(parent_surviving)

        closed_ids: list[int] = []
        for row in verified:
            if not pid_is_alive(row.pid):
                closed_ids.append(row.id)
        if closed_ids:
            _close_process_ids(conn, closed_ids)

        still_open = _load_live_processes(conn, adw_id)
        if mismatches or surviving or still_open:
            parts = ["stop incomplete"]
            if mismatches:
                parts.append(f"{len(mismatches)} identity mismatch(es)")
            if surviving:
                parts.append(f"{len(surviving)} verified pid(s) still alive")
            if still_open:
                parts.append(f"{len(still_open)} open process row(s)")
            return StopResult(
                adw_id=adw_id,
                ok=False,
                exit_code=1,
                message="; ".join(parts),
                mismatches=mismatches,
                signaled_pids=signaled,
            )

        finalize_session_failed(conn, adw_id)
        tracer.decisions_cancel(adw_id)
        return StopResult(
            adw_id=adw_id,
            ok=True,
            exit_code=0,
            message=f"stopped adw_id {adw_id} ({len(verified)} process(es))",
            signaled_pids=signaled,
        )
    finally:
        conn.close()


def _terminate_verified_rows(
    rows: list[ProcessRow],
    term_wait_s: float,
) -> tuple[list[int], set[int]]:
    """SIGTERM then SIGKILL each row, re-verifying before every signal."""
    signaled: list[int] = []
    for row in rows:
        if signal_if_verified(
            row.pid, row.command, signal.SIGTERM, start_ticks=row.start_ticks,
        ):
            signaled.append(row.pid)

    deadline = time.monotonic() + term_wait_s
    surviving = wait_pids_exit(signaled, deadline)
    for row in rows:
        if row.pid in surviving:
            if signal_if_verified(
                row.pid, row.command, signal.SIGKILL, start_ticks=row.start_ticks,
            ):
                signaled.append(row.pid)

    final_deadline = time.monotonic() + term_wait_s
    surviving = wait_pids_exit([row.pid for row in rows], final_deadline)
    return signaled, surviving


def reconcile_runs(db_path: str | Path) -> list[str]:
    """Mark ``running`` sessions failed only when every open process is definitely absent.

    A pid that still exists with a mismatched cmdline is ambiguous — those sessions
  are left untouched.
    """
    tracer = _open_tracer(db_path)
    conn = tracer.conn
    messages: list[str] = []
    try:
        running = conn.execute(
            "SELECT adw_id FROM sessions WHERE status='running'",
        ).fetchall()
        for (adw_id,) in running:
            live_rows = _load_live_processes(conn, adw_id)
            if not live_rows:
                finalize_session_failed(conn, adw_id)
                tracer.decisions_cancel(adw_id)
                messages.append(f"reconciled {adw_id}: no live process rows")
                continue

            closable: list[int] = []
            ambiguous = False
            for row in live_rows:
                proc_argv = read_proc_argv(row.pid)
                if proc_argv is None:
                    closable.append(row.id)
                    continue
                ambiguous = True
                break

            if ambiguous:
                messages.append(f"skipped {adw_id}: live or ambiguous process identity")
                continue

            _close_process_ids(conn, closable)
            finalize_session_failed(conn, adw_id)
            tracer.decisions_cancel(adw_id)
            messages.append(f"reconciled {adw_id}: all recorded pids absent")

        canceled = _cancel_decisions_for_inactive_sessions(tracer)
        if canceled:
            messages.append(f"canceled {canceled} pending decision(s) for inactive sessions")
    finally:
        conn.close()
    return messages


def _resolve_db_from_config(config_path: str, preset: str | None = None) -> str:
    cfg = load_config(config_path, preset)
    return cfg.observability.db


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(prog="factory", description="Factory control plane")
    sub = parser.add_subparsers(dest="command", required=True)

    stop = sub.add_parser("stop", help="Stop a stuck run by adw_id")
    stop.add_argument("adw_id", help="Session id to stop")
    stop.add_argument("--config", help="Path to sssf.config.yaml")
    stop.add_argument("--preset")

    reconcile = sub.add_parser(
        "reconcile",
        help="Close crashed runs with no live matching processes",
    )
    reconcile.add_argument("--config", help="Path to sssf.config.yaml")
    reconcile.add_argument("--preset")

    decisions = sub.add_parser("decisions", help="List pending human decisions")
    decisions.add_argument("--adw", dest="adw_id", help="Filter by session id")
    decisions.add_argument("--json", action="store_true", help="Emit JSON")
    decisions.add_argument("--config", help="Path to sssf.config.yaml")
    decisions.add_argument("--preset")

    answer = sub.add_parser("answer", help="Answer a pending human decision")
    answer.add_argument("decision_id", help="Decision id to answer")
    answer.add_argument("--choice", help="Selected option value")
    answer.add_argument("--text", help="Free-text answer")
    answer.add_argument("--config", help="Path to sssf.config.yaml")
    answer.add_argument("--preset")
    return parser


def main(argv: list[str] | None = None) -> int:
    parser = build_parser()
    args = parser.parse_args(argv)

    config_path = args.config
    if not config_path:
        factory_root = Path(__file__).resolve().parents[1]
        config_path = str(factory_root / "sssf.config.yaml")

    db_path = _resolve_db_from_config(config_path, args.preset)

    if args.command == "stop":
        result = stop_run(db_path, args.adw_id)
        print(result.message)
        for mismatch in result.mismatches:
            print(f"factory stop: mismatch: {mismatch}", file=sys.stderr)
        return result.exit_code

    if args.command == "reconcile":
        for line in reconcile_runs(db_path):
            print(line)
        return 0

    if args.command == "decisions":
        tracer = _open_tracer(db_path)
        try:
            pending = tracer.decisions_pending(args.adw_id)
            if args.json:
                payload = [
                    {
                        "id": row["decision_id"],
                        "adw_id": row["adw_id"],
                        "question": row["question"],
                        "options": row["options"],
                    }
                    for row in pending
                ]
                print(json.dumps(payload, indent=2))
            else:
                for row in pending:
                    print(
                        f"{row['decision_id']}\t{row['adw_id']}\t{row['question']}"
                    )
        finally:
            tracer.conn.close()
        return 0

    if args.command == "answer":
        if not args.choice and not args.text:
            print("factory answer: requires --choice or --text", file=sys.stderr)
            return 1
        tracer = _open_tracer(db_path)
        try:
            current = tracer.decision_get(args.decision_id)
            if current is None:
                print(f"factory answer: unknown decision_id {args.decision_id}", file=sys.stderr)
                return 1
            if current["status"] != "pending":
                print(
                    f"factory answer: decision {args.decision_id} is not pending"
                    f" (status={current['status']})",
                    file=sys.stderr,
                )
                return 1
            try:
                ok = tracer.decision_answer(
                    args.decision_id,
                    value=args.choice,
                    text=args.text,
                    answered_by="cli",
                )
            except ValueError as exc:
                print(f"factory answer: {exc}", file=sys.stderr)
                return 1
            if not ok:
                print(
                    f"factory answer: decision {args.decision_id} is not pending",
                    file=sys.stderr,
                )
                return 1
        finally:
            tracer.conn.close()
        return 0

    parser.error(f"unknown command: {args.command}")
    return 2


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