"""Shared job-queue plumbing for worker.py and builder.py.

Deliberately excludes most role-specific model launchers: Worker always
invokes Codex read-only, while Builder dispatches to Codex's workspace-write
sandbox, the same harness with an OpenRouter provider, or Claude's
settings-based permission scope. The shared module provides validated
options, parsers, secret-file loading and streaming loops.
"""
import json
import fcntl
import os
import queue
import re
import subprocess
import sys
import threading
import time
from pathlib import Path

import db

SUPPORTED_RUNTIMES = ("codex", "claude", "openrouter")
OPENROUTER_BASE_URL = "https://openrouter.ai/api/v1"


def validate_runtime(runtime):
    """Return the only runtime values accepted by board jobs."""
    value = str(runtime or "").strip().lower()
    if value not in SUPPORTED_RUNTIMES:
        raise ValueError("Runtime muss exakt codex, claude oder openrouter sein.")
    return value


def load_secret_file(variable):
    """Read a secret from the file named by ``variable`` without logging it."""
    path = os.environ.get(variable, "").strip()
    if not path:
        raise RuntimeError(f"{variable} ist nicht gesetzt.")
    try:
        secret = Path(path).read_text(encoding="utf-8").strip()
    except OSError as error:
        raise RuntimeError(f"Secret-Datei aus {variable} ist nicht lesbar.") from error
    if not secret:
        raise RuntimeError(f"Secret-Datei aus {variable} ist leer.")
    return secret


def append_openrouter_options(
    command,
    model_env,
    model=None,
    reasoning_env=None,
    reasoning=None,
):
    """Configure Codex's Responses harness for the OpenRouter provider."""
    command.extend(
        [
            "-c",
            'model_provider="openrouter"',
            "-c",
            'model_providers.openrouter.name="OpenRouter"',
            "-c",
            f"model_providers.openrouter.base_url={json.dumps(OPENROUTER_BASE_URL)}",
            "-c",
            'model_providers.openrouter.env_key="OPENROUTER_API_KEY"',
            "-c",
            'model_providers.openrouter.wire_api="responses"',
            "-c",
            "model_providers.openrouter.requires_openai_auth=false",
        ]
    )
    selected_model = str(model or "").strip() or os.environ.get(model_env, "").strip()
    if not selected_model:
        raise ValueError(f"{model_env} ist nicht gesetzt.")
    if len(selected_model) > 128 or not re.fullmatch(
        r"[A-Za-z0-9][A-Za-z0-9._:+/-]*", selected_model
    ):
        raise ValueError("OpenRouter-Modell enthält ungültige Zeichen.")
    command.extend(["--model", selected_model])

    selected_reasoning = (
        os.environ.get(reasoning_env, "")
        if reasoning is None and reasoning_env
        else reasoning or ""
    ).strip()
    if selected_reasoning:
        if not re.fullmatch(r"[A-Za-z0-9_-]{1,32}", selected_reasoning):
            raise ValueError("OpenRouter-Reasoning-Effort enthält ungültige Zeichen.")
        command.extend(
            ["-c", f"model_reasoning_effort={json.dumps(selected_reasoning)}"]
        )


def write_payload(path, payload):
    temporary = path.with_suffix(".tmp")
    temporary.write_text(
        json.dumps(payload, ensure_ascii=False) + "\n",
        encoding="utf-8",
    )
    temporary.chmod(0o600)
    temporary.replace(path)


def claim_jobs(inbox, processing):
    processing.mkdir(parents=True, exist_ok=True, mode=0o700)
    for source in sorted(inbox.glob("*.json")):
        target = processing / source.name
        with (inbox / '.delivery.lock').open('a+') as handle:
            fcntl.flock(handle, fcntl.LOCK_EX)
            if not source.exists():
                continue
            claims = inbox / '.claimed'
            claims.mkdir(exist_ok=True, mode=0o700)
            with (claims / source.name).open('w') as marker:
                marker.write('{}\n')
                marker.flush()
                os.fsync(marker.fileno())
            fd = os.open(claims, os.O_DIRECTORY)
            try:
                os.fsync(fd)
            finally:
                os.close(fd)
            source.replace(target)
        yield target


def add_first_reply_notice(role, sender, response, notice_text):
    if db.is_introduced(role, sender):
        return response, None
    return f"{notice_text.strip()}\n\n{response}", sender


def mark_introduced(role, sender):
    db.mark_introduced(role, sender)


def load_session(role, sender, now=time.time):
    state = db.load_session(role, sender)
    if state is None:
        return None
    maximum_idle = int(os.environ.get("AGENT_SESSION_IDLE_SECONDS", "1800"))
    session_id, updated_at = state
    if (
        not isinstance(session_id, str)
        or not session_id
        or not isinstance(updated_at, (int, float))
        or now() - updated_at > maximum_idle
    ):
        db.reset_session(role, sender)
        return None
    return session_id


def save_session(role, sender, session_id):
    db.save_session(role, sender, session_id)


def reset_session(role, sender):
    db.reset_session(role, sender)


def load_context(role, sender, now=time.time, scope=""):
    """Counterpart to load_session() for roles that carry a previous
    reply forward as plain-text context instead of resuming a Codex
    session (see docs/intent/builder-iterationslimit.md) -- same
    idle-expiry rule, separate table (db.context, not db.sessions).

    `scope` isolates this from an unrelated concurrent job for the same
    (role, sender) -- see db.py's context table comment. Board jobs pass
    their board_slug; plain XMPP chat leaves it at the default "" for
    ordinary per-sender conversational continuity."""
    state = db.load_context(role, sender, scope)
    if state is None:
        return None
    maximum_idle = int(os.environ.get("AGENT_SESSION_IDLE_SECONDS", "1800"))
    context_text, updated_at = state
    if (
        not isinstance(context_text, str)
        or not context_text
        or not isinstance(updated_at, (int, float))
        or now() - updated_at > maximum_idle
    ):
        db.reset_context(role, sender, scope)
        return None
    return context_text


def save_context(role, sender, context_text, scope=""):
    db.save_context(role, sender, context_text, scope=scope)


def reset_context(role, sender, scope=""):
    db.reset_context(role, sender, scope)


def parse_codex_events(output):
    session_id = None
    # Codex can emit more than one item.completed/agent_message per turn
    # (e.g. an intermediate update followed by the final reply) -- keeping
    # only the last one silently dropped everything before it, which
    # looked from the outside like a long reply getting cut off, with no
    # error and well under AGENT_MAX_RESPONSE_CHARS (found 2026-08-01 after
    # a user report of truncated replies). Concatenate all of them instead.
    response_parts = []
    json_errors = []
    for line in output.splitlines():
        if not line.strip():
            continue
        try:
            event = json.loads(line)
        except ValueError as error:
            json_errors.append(error)
            continue
        if event.get("type") == "thread.started":
            session_id = event.get("thread_id")
        if event.get("type") == "item.completed":
            item = event.get("item", {})
            if item.get("type") == "agent_message" and item.get("text"):
                response_parts.append(item["text"])

    if not session_id:
        raise RuntimeError("Codex lieferte keine Session-ID.")
    response = "\n\n".join(response_parts).strip()
    if not response:
        if json_errors:
            raise RuntimeError(
                f"Ungültige Codex-JSONL-Ausgabe: {json_errors[-1]}"
            ) from json_errors[-1]
        raise RuntimeError("Codex lieferte keine Antwort.")
    return response, session_id


def _integer(value):
    """Return a non-negative integer for provider usage fields."""
    return value if isinstance(value, int) and not isinstance(value, bool) and value >= 0 else 0


def parse_codex_telemetry(output):
    """Extract prompt-free telemetry from Codex/OpenRouter ``--json`` output.

    ``turn.completed.usage`` is the authoritative aggregate emitted by the
    harness.  There is normally one completed turn per ``codex exec`` call;
    summing also does the right thing if a future harness emits several.
    """
    session_id = None
    usage = {
        "input_tokens": 0,
        "cached_input_tokens": 0,
        "output_tokens": 0,
        "reasoning_tokens": 0,
        "total_tokens": 0,
    }
    iterations = 0
    for line in output.splitlines():
        try:
            event = json.loads(line)
        except (TypeError, ValueError):
            continue
        if event.get("type") == "thread.started" and event.get("thread_id"):
            session_id = event["thread_id"]
        if event.get("type") == "item.completed":
            iterations += 1
        if event.get("type") != "turn.completed" or not isinstance(event.get("usage"), dict):
            continue
        source = event["usage"]
        input_tokens = _integer(source.get("input_tokens"))
        output_tokens = _integer(source.get("output_tokens"))
        details = source.get("output_tokens_details") or {}
        reasoning_tokens = _integer(
            source.get("reasoning_tokens", details.get("reasoning_tokens"))
        )
        usage["input_tokens"] += input_tokens
        usage["cached_input_tokens"] += _integer(source.get("cached_input_tokens"))
        usage["output_tokens"] += output_tokens
        usage["reasoning_tokens"] += reasoning_tokens
        usage["total_tokens"] += _integer(source.get("total_tokens")) or input_tokens + output_tokens
    return {
        "session_id": session_id,
        "actual_profile": {},
        "usage": usage,
        "timing": {},
        "iterations": iterations,
    }


def build_partial_response(output):
    """Best-effort counterpart to parse_codex_events() for an
    iteration-limit cutoff (see run_streaming() below): the JSONL stream
    may end mid-turn with no thread.started/complete turn, so this never
    raises -- it just concatenates whatever agent_message text arrived
    before the process was terminated, or "" if none did."""
    parts = []
    for line in output.splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            event = json.loads(line)
        except ValueError:
            continue
        if event.get("type") == "item.completed":
            item = event.get("item", {})
            if item.get("type") == "agent_message" and item.get("text"):
                parts.append(item["text"])
    return "\n\n".join(parts).strip()


def _drain_lines(pipe, sink):
    try:
        for line in iter(pipe.readline, ""):
            sink.put(line)
    finally:
        sink.put(None)
        pipe.close()


def run_streaming(command, body, environment, timeout, iteration_limit):
    """Run `command` with `body` on stdin, counting Codex `item.completed`
    JSONL events (the finest step granularity Codex's own --json output
    exposes -- verified live: even a multi-tool-call turn is one single
    turn.started/turn.completed pair) as iterations. Once the count
    exceeds `iteration_limit`, the process is terminated (SIGTERM, then
    SIGKILL if it doesn't exit promptly) instead of being allowed to run
    unbounded -- see docs/intent/builder-iterationslimit.md. The overall
    wall-clock `timeout` is enforced the same way subprocess.run's
    `timeout` was (raises subprocess.TimeoutExpired) when the iteration
    limit itself is never hit.

    Returns (stdout_text, limit_reached, returncode).
    """
    process = subprocess.Popen(
        command,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        env=environment,
    )
    process.stdin.write(body)
    process.stdin.close()

    output_queue = queue.Queue()
    reader = threading.Thread(
        target=_drain_lines, args=(process.stdout, output_queue), daemon=True
    )
    reader.start()

    lines = []
    iterations = 0
    limit_reached = False
    deadline = time.monotonic() + timeout
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            process.kill()
            process.wait()
            raise subprocess.TimeoutExpired(command, timeout)
        try:
            line = output_queue.get(timeout=min(remaining, 1.0))
        except queue.Empty:
            continue
        if line is None:
            break
        lines.append(line)
        stripped = line.strip()
        if not stripped:
            continue
        try:
            event = json.loads(stripped)
        except ValueError:
            continue
        if event.get("type") == "item.completed":
            iterations += 1
            if iterations > iteration_limit:
                limit_reached = True
                process.terminate()
                try:
                    process.wait(timeout=5)
                except subprocess.TimeoutExpired:
                    process.kill()
                    process.wait()
                break

    returncode = process.wait()
    stderr = process.stderr.read()
    if stderr:
        print(stderr.rstrip(), file=sys.stderr)
    return "".join(lines), limit_reached, returncode


def append_model_options(
    command,
    model_env="AGENT_CODEX_MODEL",
    reasoning_env="AGENT_CODEX_REASONING_EFFORT",
    model=None,
    reasoning=None,
):
    """Add validated model controls before the ``exec`` command.

    The conversational Agent uses the defaults; the Builder passes its
    dedicated environment names so both roles can be tuned independently.
    """
    model = (os.environ.get(model_env, "") if model is None else model).strip()
    if model:
        if len(model) > 128 or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:+-]*", model):
            raise ValueError("AGENT_CODEX_MODEL enthält ungültige Zeichen.")
        command.extend(["--model", model])

    reasoning = (os.environ.get(reasoning_env, "") if reasoning is None else reasoning).strip()
    if reasoning:
        if not re.fullmatch(r"[A-Za-z0-9_-]{1,32}", reasoning):
            raise ValueError("AGENT_CODEX_REASONING_EFFORT enthält ungültige Zeichen.")
        # The CLI has no dedicated reasoning flag; -c is the supported
        # config override and must be supplied before the `exec` subcommand.
        command.extend(["-c", f"model_reasoning_effort={json.dumps(reasoning)}"])


def parse_claude_events(output):
    """Claude Code's --output-format stream-json counterpart to
    parse_codex_events(): each line is one JSONL event; the run ends with
    exactly one type=="result" event carrying the final plain-text answer
    (result["result"]), the session id, and an is_error flag -- unlike
    Codex's thread.started/item.completed shape, there's no separate
    session-start event to watch for.

    The is_error branch below is verified only against the *shape* of a
    successful run (is_error/subtype/api_error_status all present and
    populated in a success case) -- it has not yet been exercised against
    a genuine Claude-side failure (e.g. a rate limit) in production.
    Revisit this once one is observed live.
    """
    result_event = None
    for line in output.splitlines():
        if not line.strip():
            continue
        try:
            event = json.loads(line)
        except ValueError as error:
            raise RuntimeError(f"Ungültige Claude-JSONL-Ausgabe: {error}") from error
        if event.get("type") == "result":
            result_event = event

    if result_event is None or not result_event.get("session_id"):
        raise RuntimeError("Claude lieferte keine Session-ID.")
    if result_event.get("is_error"):
        detail = result_event.get("result") or result_event.get("subtype") or "unbekannter Fehler"
        raise RuntimeError(f"Claude endete mit einem Fehler: {detail}")
    response = (result_event.get("result") or "").strip()
    if not response:
        raise RuntimeError("Claude lieferte keine Antwort.")
    return response, result_event["session_id"]


def parse_claude_telemetry(output):
    """Extract Claude's final aggregate usage/result metadata.

    Assistant events repeat per-request usage while streaming.  The final
    ``result`` event is therefore the sole source for totals, avoiding double
    counting and keeping message/prompt contents out of the completion record.
    """
    result = None
    streamed_tool_calls = 0
    for line in output.splitlines():
        try:
            event = json.loads(line)
        except (TypeError, ValueError):
            continue
        if event.get("type") == "assistant":
            streamed_tool_calls += sum(
                1
                for block in event.get("message", {}).get("content", [])
                if block.get("type") == "tool_use"
            )
        elif event.get("type") == "result":
            result = event

    result = result or {}
    source = result.get("usage") if isinstance(result.get("usage"), dict) else {}
    direct_input = _integer(source.get("input_tokens"))
    cache_creation = _integer(source.get("cache_creation_input_tokens"))
    cache_read = _integer(source.get("cache_read_input_tokens"))
    input_tokens = direct_input + cache_creation + cache_read
    output_tokens = _integer(source.get("output_tokens"))
    details = source.get("output_tokens_details") or {}
    reasoning_tokens = _integer(
        source.get("reasoning_tokens", details.get("thinking_tokens"))
    )

    model_usage = result.get("modelUsage") or {}
    actual_model = None
    if isinstance(model_usage, dict) and model_usage:
        def model_weight(item):
            values = item[1] if isinstance(item[1], dict) else {}
            return (
                float(values.get("costUSD") or 0),
                _integer(values.get("inputTokens"))
                + _integer(values.get("cacheReadInputTokens"))
                + _integer(values.get("cacheCreationInputTokens"))
                + _integer(values.get("outputTokens")),
            )

        actual_model = max(model_usage.items(), key=model_weight)[0]

    timing = {}
    for source_key, target_key in (
        ("duration_ms", "duration_seconds"),
        ("duration_api_ms", "model_duration_seconds"),
        ("ttft_ms", "ttft_seconds"),
    ):
        value = result.get(source_key)
        if isinstance(value, (int, float)) and not isinstance(value, bool) and value >= 0:
            timing[target_key] = value / 1000

    iterations = _integer(result.get("num_turns")) or streamed_tool_calls
    return {
        "session_id": result.get("session_id"),
        "actual_profile": {"model": actual_model} if actual_model else {},
        "usage": {
            "input_tokens": input_tokens,
            "cached_input_tokens": cache_read,
            "cache_creation_input_tokens": cache_creation,
            "output_tokens": output_tokens,
            "reasoning_tokens": reasoning_tokens,
            "total_tokens": input_tokens + output_tokens,
        },
        "timing": timing,
        "iterations": iterations,
    }


def build_claude_partial_response(output):
    """Best-effort counterpart to build_partial_response() for Claude's
    event shape -- never raises. Concatenates every text content block
    from every assistant event seen before an iteration-limit kill, or ""
    if none arrived."""
    parts = []
    for line in output.splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            event = json.loads(line)
        except ValueError:
            continue
        if event.get("type") != "assistant":
            continue
        for block in event.get("message", {}).get("content", []):
            if block.get("type") == "text" and block.get("text"):
                parts.append(block["text"])
    return "\n\n".join(parts).strip()


def run_claude_streaming(command, body, environment, cwd, timeout, iteration_limit):
    """Claude Code counterpart to run_streaming(): same Popen/drain-thread/
    deadline/SIGTERM-then-SIGKILL structure, two differences forced by
    Claude's own --output-format stream-json shape rather than Codex's:

    - `cwd` is an explicit parameter and passed to subprocess.Popen
      directly -- Claude has no -C-equivalent CLI flag to set it via argv.
    - Iterations are counted as tool_use *content blocks*, not JSONL
      lines: a single type=="assistant" event's message.content list can
      carry more than one tool_use block (parallel tool calls), each one
      a separate unit of work, so each is counted individually. This is
      not a 1:1 semantic match for Codex's item.completed counting (which
      also counts pure-text turns with no tool call at all) -- see
      docs/intent/builder-iterationslimit.md's note on this being a known
      per-runtime difference.

    Returns (stdout_text, limit_reached, returncode).
    """
    process = subprocess.Popen(
        command,
        stdin=subprocess.PIPE,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        text=True,
        env=environment,
        cwd=cwd,
    )
    process.stdin.write(body)
    process.stdin.close()

    output_queue = queue.Queue()
    reader = threading.Thread(
        target=_drain_lines, args=(process.stdout, output_queue), daemon=True
    )
    reader.start()

    lines = []
    iterations = 0
    limit_reached = False
    deadline = time.monotonic() + timeout
    while True:
        remaining = deadline - time.monotonic()
        if remaining <= 0:
            process.kill()
            process.wait()
            raise subprocess.TimeoutExpired(command, timeout)
        try:
            line = output_queue.get(timeout=min(remaining, 1.0))
        except queue.Empty:
            continue
        if line is None:
            break
        lines.append(line)
        stripped = line.strip()
        if not stripped:
            continue
        try:
            event = json.loads(stripped)
        except ValueError:
            continue
        if event.get("type") == "assistant":
            tool_calls = sum(
                1
                for block in event.get("message", {}).get("content", [])
                if block.get("type") == "tool_use"
            )
            if tool_calls:
                iterations += tool_calls
                if iterations > iteration_limit:
                    limit_reached = True
                    process.terminate()
                    try:
                        process.wait(timeout=5)
                    except subprocess.TimeoutExpired:
                        process.kill()
                        process.wait()
                    break

    returncode = process.wait()
    stderr = process.stderr.read()
    if stderr:
        print(stderr.rstrip(), file=sys.stderr)
    return "".join(lines), limit_reached, returncode


def append_claude_model_options(command, model=None, effort=None):
    """Claude counterpart to append_model_options(): real --model/--effort
    flags rather than Codex's -c model_reasoning_effort= config override,
    so both are plain trailing flags with no exec-subcommand-boundary
    concern."""
    model = (os.environ.get("AGENT_BUILDER_CLAUDE_MODEL", "") if model is None else model).strip()
    if model:
        if len(model) > 128 or not re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:+-]*", model):
            raise ValueError("AGENT_BUILDER_CLAUDE_MODEL enthält ungültige Zeichen.")
        command.extend(["--model", model])

    effort = (os.environ.get("AGENT_BUILDER_CLAUDE_REASONING_EFFORT", "") if effort is None else effort).strip()
    if effort:
        if not re.fullmatch(r"[A-Za-z0-9_-]{1,32}", effort):
            raise ValueError("AGENT_BUILDER_CLAUDE_REASONING_EFFORT enthält ungültige Zeichen.")
        command.extend(["--effort", effort])


def extract_commit_message(response, start_marker, end_marker):
    start = response.find(start_marker)
    end = response.find(end_marker)
    if start == -1 or end == -1 or end < start:
        return response, None

    message = response[start + len(start_marker) : end].strip()
    before = response[:start].strip()
    after = response[end + len(end_marker) :].strip()
    visible = "\n\n".join(part for part in (before, after) if part)
    if not message:
        return visible or response, None
    return visible or response[:start].strip() or "Änderung committed.", message


def repo_has_changes(repo_dir):
    result = subprocess.run(
        ["git", "-C", str(repo_dir), "status", "--porcelain"],
        capture_output=True,
        text=True,
        timeout=30,
        check=False,
    )
    return bool(result.stdout.strip())


def commit_and_push(repo_dir, message, branch=None):
    """Push to `branch`, or -- if not given -- whatever branch is
    currently checked out (nixos-config and website use dev, but
    dashboard and launcher only have main, and that's what their deploy
    webhooks actually pull). An explicit `branch` is for isolated builder
    worktrees (see docs/intent/board-builder-worktree-parallelism.md):
    their own checked-out branch is a disposable per-job branch, never
    the real target, so the caller must name the target explicitly
    instead of relying on the worktree's own HEAD."""
    if branch is None:
        branch_result = subprocess.run(
            ["git", "-C", str(repo_dir), "rev-parse", "--abbrev-ref", "HEAD"],
            capture_output=True,
            text=True,
            timeout=15,
            check=False,
        )
        branch = branch_result.stdout.strip()
        if branch_result.returncode != 0 or not branch:
            return False, (branch_result.stdout + branch_result.stderr).strip(), None

    for command in (
        ["git", "-C", str(repo_dir), "add", "-A"],
        ["git", "-C", str(repo_dir), "commit", "-m", message],
        ["git", "-C", str(repo_dir), "push", "origin", f"HEAD:{branch}"],
    ):
        result = subprocess.run(
            command,
            capture_output=True,
            text=True,
            timeout=60,
            check=False,
        )
        if result.returncode != 0:
            return False, (result.stdout + result.stderr).strip(), None
    log = subprocess.run(
        ["git", "-C", str(repo_dir), "log", "-1", "--format=%h"],
        capture_output=True,
        text=True,
        timeout=15,
        check=False,
    )
    return True, log.stdout.strip(), branch


def send_reply(config_env_var, sender, response):
    command = [
        os.environ["AGENT_SENDXMPP_BIN"],
        "--file",
        os.environ[config_env_var],
        sender,
    ]
    result = subprocess.run(
        command,
        input=response,
        text=True,
        capture_output=True,
        timeout=30,
        check=False,
    )
    if result.stderr:
        print(result.stderr.rstrip(), file=sys.stderr)
    if result.returncode != 0:
        raise RuntimeError(f"XMPP-Antwort endete mit Status {result.returncode}.")


def fail_job(path, failed, payload, error):
    payload["failed_at"] = time.time()
    payload["error"] = str(error)
    write_payload(path, payload)
    failed.mkdir(parents=True, exist_ok=True, mode=0o700)
    path.replace(failed / path.name)