"""Writer core for runs/<task-type>/logs/errors-<task-type>-<seq>.jsonl.

Every error record in a run log is appended through this module. It lives here
rather than inside `scripts/okstra-error-log.py` because the CLI is no longer the
only caller: the deterministic dispatcher records a worker wrapper's non-zero
exit as a `cli-failure` in-process (`dispatch_core`), and the contract sentence
that says it does is only true while both paths share one writer. The script
keeps the argparse surface and re-exports these names.
"""
from __future__ import annotations

import datetime as dt
import json
import os
from pathlib import Path

from .models import provider_ids
from .json_boundary import external_error_sidecar_source, load_external_json

STDERR_EXCERPT_MAX_BYTES = 2048
TRUNCATION_SUFFIX = "...[truncated]"
PIPE_BUF_BYTES = 4096

ALLOWED_ERROR_TYPES = {"tool-failure", "cli-failure", "contract-violation"}
# Derived, not listed. The hand-written set had drifted two providers behind the
# registry: `grok` and `kimi` ship worker definitions and can be dispatched, but
# their agent names were absent, so every error they reported was rejected at
# the argument parser — silently, for anyone who did not read the exit code.
# The registry is where a provider is added, so it is where this follows from.
ALLOWED_AGENTS = (
    {f"{provider}-worker" for provider in provider_ids("analyser")}
    # The report writer is not an analyser and is named without the suffix,
    # matching `REPORT_WRITER_WORKER_ID`.
    | {"report-writer"}
    # Lead identities come from the selected host adapter rather than the
    # provider registry; `adapters/hosts/*/relay.md` names the value to pass.
    | {"claude-lead"}
)
ALLOWED_AGENT_ROLES = {"lead", "worker", "report-writer"}
SUPPORTED_SIDECAR_SCHEMA_VERSIONS = {1}

ALLOWED_CAUSES = {
    "sandbox-denied", "service-unavailable", "auth-failed", "unknown",
}
# A `sandbox-denied` claim is only admissible with both probes attached.
CAUSE_EVIDENCE_FIELDS = ("targetProbe", "controlProbe")
# Both probes share a record's PIPE_BUF_BYTES budget with stderrExcerpt, so
# they cannot reuse the 2048 cap that assumes stderrExcerpt owns it alone.
CAUSE_PROBE_MAX_BYTES = 256
# Backstop vocabulary: scanned in `message` only — never in stderrExcerpt,
# where a kernel's real "Operation not permitted" is legitimate content.
# Deliberately excludes bare "blocked"/"blocks": everyday English that would
# reject honest records like "test blocked on upstream dependency".
_BLOCKING_CLAIM_TERMS = (
    "sandbox", "not permitted", "permission denied", "eperm",
)
# The backstop targets *unclassified* blocking claims. A worker that declared
# a specific cause has already done the honest work — `auth-failed` legitimately
# reads "permission denied" (MySQL 1045).
_UNCLASSIFIED_CAUSES = (None, "unknown")


def _now_utc():
    return dt.datetime.now(dt.timezone.utc)


def _iso(t):
    return t.isoformat()


def _truncate_utf8(s, limit):
    """Truncate to `limit` bytes without splitting a multibyte character."""
    if s is None:
        return None
    encoded = s.encode("utf-8")
    if len(encoded) <= limit:
        return s
    cut = encoded[:limit]
    while cut:
        try:
            return cut.decode("utf-8") + TRUNCATION_SUFFIX
        except UnicodeDecodeError:
            cut = cut[:-1]
    return TRUNCATION_SUFFIX


def truncate_stderr(s):
    """Truncate stderr text to STDERR_EXCERPT_MAX_BYTES, multibyte-safe."""
    return _truncate_utf8(s, STDERR_EXCERPT_MAX_BYTES)


def normalize_cause_context(context, *, message):
    """Validate a record's cause claim and return the normalized context.

    Raises ValueError on three conditions:
    - `cause` is set to a value outside ALLOWED_CAUSES;
    - `cause` is 'sandbox-denied' but the two probes are missing or blank —
      those probes are what distinguish a real denial from an unreachable
      or auth-gated target, the misdiagnosis this gate exists to stop;
    - `message` asserts a block in prose while the record left its cause
      unclassified, which would smuggle the same claim past the gate.
    """
    cause = context.get("cause") if isinstance(context, dict) else None

    if cause is not None and cause not in ALLOWED_CAUSES:
        raise ValueError(
            f"invalid cause: {cause!r} (allowed: {sorted(ALLOWED_CAUSES)})"
        )

    if cause == "sandbox-denied":
        evidence = context.get("causeEvidence")
        if not isinstance(evidence, dict):
            raise ValueError(
                "cause 'sandbox-denied' requires context.causeEvidence with "
                f"{list(CAUSE_EVIDENCE_FIELDS)}"
            )
        normalized_evidence = {}
        for field in CAUSE_EVIDENCE_FIELDS:
            value = evidence.get(field)
            if not isinstance(value, str) or not value.strip():
                raise ValueError(
                    f"cause 'sandbox-denied' requires a non-empty "
                    f"context.causeEvidence.{field}: record the command and "
                    f"its raw output that proves the claim"
                )
            normalized_evidence[field] = _truncate_utf8(
                value, CAUSE_PROBE_MAX_BYTES
            )
        return {**context, "causeEvidence": normalized_evidence}

    if message and cause in _UNCLASSIFIED_CAUSES:
        lowered = message.lower()
        hit = next((t for t in _BLOCKING_CLAIM_TERMS if t in lowered), None)
        if hit:
            raise ValueError(
                f"message asserts a blocking claim ({hit!r}) without "
                "context.cause='sandbox-denied' + context.causeEvidence. "
                "Either attach the two probes, or state the cause you "
                "actually verified."
            )

    return context


def append_jsonl_line(path, record):
    """Append a single JSON record as one line to ``path``.

    Atomicity guarantee (POSIX only):
        With ``O_APPEND`` and a single ``write()`` syscall, the kernel
        appends the entire payload as one indivisible operation as long as
        the payload size is at most ``PIPE_BUF`` (4096 bytes on Linux and
        macOS). Larger payloads may be split across syscalls and interleave
        with concurrent writers, so this helper rejects them with
        ``ValueError`` rather than silently losing atomicity.

    The atomicity contract holds only on POSIX filesystems with O_APPEND
    semantics. Concurrent writers using ``O_TRUNC``, ``unlink``, or
    non-append modes against the same path break the contract and are
    out of scope for this helper.

    Caller responsibilities:
    - Keep records small (this module's stderr excerpt cap of
      ``STDERR_EXCERPT_MAX_BYTES`` exists to keep records well under
      ``PIPE_BUF_BYTES``).
    - Handle ``TypeError`` from ``json.dumps`` for non-serializable values.

    Creates parent directories as needed.
    """
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    # ensure_ascii=False keeps UTF-8 compact (no \uXXXX escapes).
    # json.dumps escapes literal newlines inside string values, so the
    # only unescaped newline is the record separator we append below.
    line = json.dumps(record, ensure_ascii=False, separators=(",", ":")) + "\n"
    data = line.encode("utf-8")
    if len(data) > PIPE_BUF_BYTES:
        raise ValueError(
            f"record too large for atomic append: {len(data)} bytes > "
            f"PIPE_BUF ({PIPE_BUF_BYTES})"
        )
    # mode 0o644: owner read/write, group/world read-only.
    fd = os.open(str(p), os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o644)
    try:
        os.write(fd, data)
    finally:
        os.close(fd)


def append_observed(
    *,
    out_path,
    task_key,
    phase,
    agent,
    agent_role,
    model,
    error_type,
    command,
    command_kind,
    exit_code,
    duration_ms,
    message,
    stderr_excerpt,
    context,
    now=None,
    identity=None,
):
    """Append a lead-observed error event to errors.jsonl."""
    if error_type not in ALLOWED_ERROR_TYPES:
        raise ValueError(f"invalid errorType: {error_type!r}")
    if agent not in ALLOWED_AGENTS:
        raise ValueError(f"invalid agent: {agent!r}")
    if agent_role not in ALLOWED_AGENT_ROLES:
        raise ValueError(f"invalid agentRole: {agent_role!r}")
    # Runs before append_jsonl_line so a rejected claim leaves no trace in the
    # log: a written-then-flagged record is still a record someone can cite.
    context = normalize_cause_context(context, message=message)
    ts = _iso(now or _now_utc())
    rec = {
        "ts": ts,
        "recordedAt": ts,
        "taskKey": task_key,
        "phase": str(phase),
        "agent": agent,
        "agentRole": agent_role,
        "model": model,
        "source": "lead-observed",
        "errorType": error_type,
        "command": command,
        "commandKind": command_kind,
        "exitCode": exit_code,
        "durationMs": duration_ms,
        "message": message,
        "stderrExcerpt": truncate_stderr(stderr_excerpt),
        "context": context,
    }
    from okstra_ctl.execution_identity import stored_identity

    rec.update(stored_identity(identity))
    append_jsonl_line(out_path, rec)
    return rec


def _worker_sidecar_record(entry, *, recorded_at, task_key, agent, agent_role, model, identity):
    error_type = entry.get("errorType")
    if error_type not in ALLOWED_ERROR_TYPES:
        raise ValueError(f"invalid errorType in sidecar: {error_type!r}")
    rec = {
        "ts": entry.get("ts"), "recordedAt": recorded_at,
        "taskKey": task_key,
        "phase": str(entry.get("phase")) if entry.get("phase") is not None else None,
        "agent": agent, "agentRole": agent_role, "model": model,
        "source": "worker-reported", "errorType": error_type,
        "command": entry.get("command"), "commandKind": entry.get("commandKind"),
        "exitCode": entry.get("exitCode"), "durationMs": entry.get("durationMs"),
        "message": entry.get("message"),
        "stderrExcerpt": truncate_stderr(entry.get("stderrExcerpt")),
        "context": normalize_cause_context(
            entry.get("context"), message=entry.get("message")
        ),
    }
    from okstra_ctl.execution_identity import stored_identity

    rec.update(stored_identity(identity))
    rec.update(stored_identity(entry))
    return rec


def dump_from_worker_sidecar(
    *,
    sidecar_path,
    out_path,
    task_key,
    agent,
    agent_role,
    model,
    now=None,
    identity=None,
):
    """Read worker sidecar errors[] and append each to errors.jsonl with
    Lead-side metadata filled in. Returns number of records appended.

    Raises ValueError if:
      - ``agent`` or ``agent_role`` is not in the allow-lists
      - sidecar ``schemaVersion`` is not in ``SUPPORTED_SIDECAR_SCHEMA_VERSIONS``
      - any entry's ``errorType`` is not in ``ALLOWED_ERROR_TYPES``
      - any entry asserts a blocking cause without its required evidence

    Returns 0 (no-op) if the sidecar file does not exist or its
    ``errors`` list is empty.

    Partial-failure semantics: entries are validated and appended in
    order. If entry N fails validation, entries 0..N-1 have already
    been written to ``out_path`` and are NOT rolled back. Callers that
    require atomicity must validate the sidecar payload before invoking
    this function.
    """
    if agent not in ALLOWED_AGENTS:
        raise ValueError(f"invalid agent: {agent!r}")
    if agent_role not in ALLOWED_AGENT_ROLES:
        raise ValueError(f"invalid agentRole: {agent_role!r}")
    p = Path(sidecar_path)
    if not p.exists():
        return 0
    # 외부 입력: 레거시 자동화가 제공하는 오류 sidecar 호환 경로다.
    payload = load_external_json(
        external_error_sidecar_source(p),
        artifact="legacy error sidecar",
    )
    schema = payload.get("schemaVersion")
    if schema not in SUPPORTED_SIDECAR_SCHEMA_VERSIONS:
        raise ValueError(f"unsupported sidecar schemaVersion: {schema!r}")
    recorded_at = _iso(now or _now_utc())
    count = 0
    for entry in payload.get("errors") or []:
        rec = _worker_sidecar_record(
            entry, recorded_at=recorded_at, task_key=task_key, agent=agent,
            agent_role=agent_role, model=model, identity=identity,
        )
        append_jsonl_line(out_path, rec)
        count += 1
    return count