inboxos-v1/pipelines/common.py

"""Shared helpers for the InboxOS pipelines.

Design invariants:
- Idempotency: every NocoDB write uses Message-ID as the natural key and
  performs an UPSERT (POST with the xc-token / PATCH when a record exists).
  No code path is allowed to do a plain INSERT without conflict handling.
- Observability: every step logs structured JSON to stdout and a per-run
  file under pipelines/_artifacts/ so a failed run can be triaged from
  the workflow artifact without re-executing.
- Air-gapped: the only outbound network calls are to the self-hosted
  services in the inboxos-net Docker network (NocoDB, Papra, SeaweedFS,
  Ollama, OpenCode). No external APIs are called.
"""

from __future__ import annotations

import json
import logging
import os
import re
import subprocess
import sys
import time
import uuid
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Any, Iterable, Optional

import boto3
import requests
from botocore.client import Config as BotoConfig
from tenacity import (
    retry,
    retry_if_exception_type,
    stop_after_attempt,
    wait_exponential,
)


ARTIFACT_DIR = Path(__file__).parent / "_artifacts"
ARTIFACT_DIR.mkdir(parents=True, exist_ok=True)


# ── Logging ─────────────────────────────────────────────

def _run_id() -> str:
    return os.environ.get("INBOXOS_RUN_ID") or f"{int(time.time())}-{uuid.uuid4().hex[:8]}"


def get_logger(name: str, run_kind: str) -> logging.Logger:
    run_id = _run_id()
    log_path = ARTIFACT_DIR / f"{run_kind}-run-{run_id}.log"

    logger = logging.getLogger(f"inboxos.{name}.{run_id}")
    logger.setLevel(os.environ.get("LOG_LEVEL", "INFO"))
    logger.propagate = False

    if not logger.handlers:
        fmt = logging.Formatter(
            fmt='{"ts":"%(asctime)s","lvl":"%(levelname)s","logger":"%(name)s","msg":"%(message)s"}',
            datefmt="%Y-%m-%dT%H:%M:%S%z",
        )

        sh = logging.StreamHandler(sys.stdout)
        sh.setFormatter(fmt)
        logger.addHandler(sh)

        fh = logging.FileHandler(log_path, encoding="utf-8")
        fh.setFormatter(fmt)
        logger.addHandler(fh)

    return logger


def log_event(logger: logging.Logger, event: str, **fields: Any) -> None:
    """Emit a structured event line."""
    payload = {"event": event, **fields}
    logger.info(json.dumps(payload, default=str))


# ── SeaweedFS (S3-compatible) ──────────────────────────

@dataclass
class S3Config:
    endpoint: str
    access_key: str
    secret_key: str
    bucket: str

    @classmethod
    def from_env(cls, bucket_default: str | None = None) -> "S3Config":
        endpoint = _require("SEAWEEDFS_S3_ENDPOINT")
        access_key = _require("SEAWEEDFS_S3_ACCESS_KEY")
        secret_key = _require("SEAWEEDFS_S3_SECRET_KEY")
        bucket = os.environ.get("SEAWEEDFS_S3_BUCKET", bucket_default or "")
        if not bucket:
            raise RuntimeError("SEAWEEDFS_S3_BUCKET is required")
        return cls(endpoint=endpoint, access_key=access_key, secret_key=secret_key, bucket=bucket)


def make_s3_client(cfg: S3Config) -> Any:
    """Build a boto3 S3 client pointed at SeaweedFS.

    Path-style addressing is required because SeaweedFS does not
    support virtual-hosted buckets out of the box.
    """
    return boto3.client(
        "s3",
        endpoint_url=cfg.endpoint,
        aws_access_key_id=cfg.access_key,
        aws_secret_access_key=cfg.secret_key,
        config=BotoConfig(
            signature_version="s3v4",
            s3={"addressing_style": "path"},
            retries={"max_attempts": 5, "mode": "standard"},
        ),
        region_name="us-east-1",  # required by boto3; ignored by SeaweedFS
    )


@retry(
    reraise=True,
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
    retry=retry_if_exception_type(Exception),
)
def ensure_bucket(s3: Any, bucket: str) -> None:
    try:
        s3.head_bucket(Bucket=bucket)
    except Exception as exc:  # noqa: BLE001
        code = getattr(exc, "response", {}).get("Error", {}).get("Code", "")
        if code in ("404", "NoSuchBucket", "NotFound"):
            s3.create_bucket(Bucket=bucket)
        else:
            raise


@retry(
    reraise=True,
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
)
def s3_put_bytes(s3: Any, bucket: str, key: str, body: bytes, content_type: str = "application/octet-stream") -> str:
    s3.put_object(Bucket=bucket, Key=key, Body=body, ContentType=content_type)
    return f"s3://{bucket}/{key}"


@retry(
    reraise=True,
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
)
def s3_get_bytes(s3: Any, bucket: str, key: str) -> bytes:
    obj = s3.get_object(Bucket=bucket, Key=key)
    return obj["Body"].read()


@retry(
    reraise=True,
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
)
def s3_get_text_optional(s3: Any, bucket: str, key: str) -> str | None:
    try:
        return s3_get_bytes(s3, bucket, key).decode("utf-8")
    except Exception as exc:  # noqa: BLE001
        code = getattr(exc, "response", {}).get("Error", {}).get("Code", "")
        if code in ("404", "NoSuchKey", "NoSuchBucket", "NotFound"):
            return None
        raise


# ── NocoDB ─────────────────────────────────────────────

@dataclass
class NocoConfig:
    base_url: str
    api_token: str
    project_id: str

    @classmethod
    def from_env(cls) -> "NocoConfig":
        return cls(
            base_url=_require("NOCODB_BASE_URL").rstrip("/"),
            api_token=_require("NOCODB_API_TOKEN"),
            project_id=_require("NOCODB_PROJECT_ID"),
        )


def papra_index_upsert(
    cfg: NocoConfig,
    *,
    message_id: str,
    papra_document_id: str | None,
    module: str | None,
    kind: str,
    outcome: str,
    error: str | None = None,
    logger: Any = None,
) -> dict[str, Any]:
    """Record one Papra write intent in the `papra_index` audit table (REQ-INB-19).

    Thin wrapper around `nocodb_upsert` keyed on `Message_ID`. Reads
    the target table id from the env var `NOCODB_PAPRA_INDEX_TABLE`
    (set by `scripts/apply_schemas.py` output, identical shape to
    `NOCODB_MODULE_TABLES` but keyed on the audit-table name). If
    the env var is unset the helper is a no-op and emits a
    `papra_index_upsert_skipped` log event so the missing-config
    state is observable in the run log.

    The helper never raises. Papra write failures must not stop the
    pipeline; the audit-write failure is logged with the same
    `papra_index_upsert_failed` event and the call returns an empty
    dict.
    """
    table_id = os.environ.get("NOCODB_PAPRA_INDEX_TABLE", "").strip()
    if not table_id:
        if logger is not None:
            log_event(
                logger,
                "papra_index_upsert_skipped",
                reason="NOCODB_PAPRA_INDEX_TABLE unset",
                message_id=message_id,
                kind=kind,
                outcome=outcome,
            )
        return {}

    record = {
        "Message_ID": message_id,
        "papra_document_id": papra_document_id,
        "module": module,
        "kind": kind,
        "decision_at": now_iso(),
        "outcome": outcome,
        "error": error,
    }
    try:
        result = nocodb_upsert(cfg, table_id, record, message_id)
    except Exception as exc:  # noqa: BLE001
        if logger is not None:
            log_event(
                logger,
                "papra_index_upsert_failed",
                message_id=message_id,
                kind=kind,
                outcome=outcome,
                error=str(exc),
            )
        return {}

    if logger is not None:
        log_event(
            logger,
            "papra_index_event",
            message_id=message_id,
            papra_document_id=papra_document_id,
            module=module,
            kind=kind,
            decision_at=record["decision_at"],
            outcome=outcome,
            error=error,
            noco_action=result.get("action"),
        )
    return result


@retry(
    reraise=True,
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
)
def nocodb_upsert(
    cfg: NocoConfig,
    table_id: str,
    record: dict[str, Any],
    message_id: str,
) -> dict[str, Any]:
    """UPSERT one record into a NocoDB table.

    NocoDB does not provide a server-side upsert on the free
    / self-hosted plan via the REST API, so we look the record
    up by Message-ID and either PATCH (if found) or POST (if not).
    Either way the result is idempotent: re-running with the same
    Message-ID overwrites the same row.

    The Message-ID filter is exposed by NocoDB as
    `?where=(Message_ID,eq,<value>)`. If your column name differs,
    pass it in via the `message_id_field` parameter.
    """
    message_id_field = "Message_ID"
    headers = {"xc-token": cfg.api_token, "Content-Type": "application/json"}

    list_url = f"{cfg.base_url}/api/v2/tables/{table_id}/records"
    search = requests.get(
        list_url,
        params={"where": f"({message_id_field},eq,{message_id})", "limit": 1},
        headers=headers,
        timeout=30,
    )
    search.raise_for_status()
    rows = search.json().get("list", [])

    if rows:
        record_id = rows[0]["Id"]
        # PATCH existing record (full overwrite of provided fields).
        # NocoDB v2 dropped the per-record `/records/{id}` endpoint; the
        # supported path is a bulk PATCH to `/records` with the body
        # shaped as `[{"Id": id, ...overrides...}]`. PATCHing the
        # single-record URL returns 404 even when the row exists (live
        # run #682 failure on 2026-06-07).
        patch_url = list_url
        resp = requests.patch(
            patch_url, json=[{"Id": record_id, **record}], headers=headers, timeout=30
        )
        if not resp.ok:
            raise RuntimeError(
                f"NocoDB PATCH failed: status={resp.status_code} body={resp.text[:500]!r} "
                f"record_id={record_id} table_id={table_id}"
            )
        resp.raise_for_status()
        return {"action": "patch", "id": record_id, "record": resp.json()}

    # POST new record
    resp = requests.post(list_url, json=record, headers=headers, timeout=30)
    if not resp.ok:
        raise RuntimeError(
            f"NocoDB POST failed: status={resp.status_code} body={resp.text[:500]!r} "
            f"table_id={table_id} record_keys={sorted(record.keys())}"
        )
    resp.raise_for_status()
    return {"action": "post", "record": resp.json()}


def nocodb_upsert_with_modules(
    cfg: NocoConfig,
    modules: Iterable[tuple[str, dict[str, Any]]],
    message_id: str,
) -> list[dict[str, Any]]:
    """Upsert a single Message-ID into multiple module tables.

    `modules` is an iterable of (table_id, record) pairs. The record
    is expected to already contain the Message_ID field; the function
    is responsible for ensuring idempotency on that key.
    """
    out = []
    for table_id, record in modules:
        result = nocodb_upsert(cfg, table_id, record, message_id)
        out.append({"table_id": table_id, **result})
    return out


# ── Gitea REST API (Basic Auth) ──────────────────────────

@dataclass
class GiteaConfig:
    """Credentials for talking to the Gitea REST API.

    InboxOS uses HTTP Basic Auth (GITEA_USER / GITEA_PASS) rather than a
    personal access token. Rationale: Gitea's API does not return the
    plaintext of a freshly created access token (only its sha1 hash and
    last-eight characters), so a headless agent cannot produce a
    GITEA_TOKEN env var from the API alone. The user/password pair is
    sufficient for every endpoint the InboxOS automation needs
    (repo create, secrets, runner registration lookups, etc.) and
    matches the air-gapped, self-hosted posture of the project.
    """

    base_url: str
    username: str
    password: str

    @classmethod
    def from_env(cls) -> "GiteaConfig":
        return cls(
            base_url=_require("GITEA_INSTANCE_URL").rstrip("/"),
            username=_require("GITEA_USER"),
            password=_require("GITEA_PASS"),
        )


@retry(
    reraise=True,
    stop=stop_after_attempt(5),
    wait=wait_exponential(multiplier=1, min=1, max=30),
    retry=retry_if_exception_type((requests.RequestException, RuntimeError)),
)
def gitea_request(
    cfg: GiteaConfig,
    method: str,
    path: str,
    *,
    json_body: dict[str, Any] | None = None,
    params: dict[str, Any] | None = None,
    timeout: int = 30,
) -> requests.Response:
    """HTTP request to the Gitea REST API with HTTP Basic Auth.

    `path` should start with `/` (e.g. `/api/v1/user`). 5xx responses
    are retried; 4xx responses are returned to the caller as-is so the
    caller can decide whether to treat them as a hard failure.
    """
    url = f"{cfg.base_url}{path}"
    resp = requests.request(
        method.upper(),
        url,
        auth=(cfg.username, cfg.password),
        json=json_body,
        params=params,
        timeout=timeout,
    )
    if resp.status_code >= 500:
        raise RuntimeError(
            f"Gitea {method.upper()} {path} returned {resp.status_code}: {resp.text[:200]!r}"
        )
    return resp


# ── OpenCode invocation ────────────────────────────────

@dataclass
class OpenCodeConfig:
    bin: str
    prompt_dir: str

    @classmethod
    def from_env(cls) -> "OpenCodeConfig":
        return cls(
            bin=os.environ.get("OPENCODE_BIN", "opencode"),
            prompt_dir=os.environ.get("OPENCODE_PROMPT_DIR", "./.opencode/commands"),
        )


@retry(
    reraise=True,
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=2, min=2, max=30),
)
def opencode_run_command(
    cfg: OpenCodeConfig,
    command_name: str,
    payload: dict[str, Any],
    timeout: int = 120,
) -> dict[str, Any]:
    """Invoke the OpenCode CLI with a named command + JSON payload.

    `command_name` is the filename (without .md) under
    `OPENCODE_PROMPT_DIR` — e.g. `project:extract-finance` for
    `project:extract-finance.md`.

    OpenCode is expected to print exactly one JSON object on stdout.
    The wrapper passes the email as JSON on stdin and parses the
    response. Non-zero exit, empty stdout, or invalid JSON are
    treated as failures and retried.

    NOTE: opencode 1.16.2 server returns `UnknownError` on every
    `--command` invocation (ref: err_xxx). This path is currently
    broken end-to-end; use `llm_run_command` with the `ollama`
    backend instead. This function is kept for offline tests and
    future opencode versions.
    """
    cmd = [
        cfg.bin,
        "run",
        "--command", command_name,
    ]
    proc = subprocess.run(
        cmd,
        input=json.dumps(payload).encode("utf-8"),
        capture_output=True,
        timeout=timeout,
        check=False,
    )
    if proc.returncode != 0:
        raise RuntimeError(
            f"opencode exited {proc.returncode}: stderr={proc.stderr.decode('utf-8', 'replace')}"
        )
    stdout = proc.stdout.decode("utf-8", "replace").strip()
    if not stdout:
        raise RuntimeError("opencode produced no stdout")
    try:
        return json.loads(stdout)
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"opencode did not emit valid JSON: {exc}; raw={stdout[:500]!r}") from exc


# ── LLM invocation (backend selector) ──────────────────


@dataclass
class LLMConfig:
    """Backend selector for the language-model step.

    Two backends are supported:

    - `ollama` (default, recommended): direct call to an
      OpenAI-compatible /v1/chat/completions endpoint. Bypasses the
      opencode CLI entirely. Reads the prompt from
      `prompt_dir/<command_name>.md` and sends the body as the
      system message, the JSON payload as the user message.
    - `opencode`: legacy path that shells out to the opencode CLI.
      Currently broken on opencode 1.16.2 (`UnknownError` on every
      `--command` invocation); kept for offline tests only.

    Per-step model override (the one human-facing knob):
        `INBOXOS_LLM_MODEL` is the model for classify and any command
        that does not start with `INBOXOS_EXTRACT_COMMAND_PREFIX`
        (default `inboxos:extract-`). Default `gemma4:e4b` — cheap
        enough to run on every new email.
        `INBOXOS_LLM_MODEL_EXTRACT` overrides the model used for
        `inboxos:extract-*` commands only. Default `gemma4:e4b`.
        Set both to the same value if you want a single model across
        the pipeline.

    Operator note (INB-21 board decision, 2026-06-07):
        There is no model escalation chain. The pipeline makes a
        single LLM call per step. Transient HTTP / JSON failures are
        retried by tenacity (3 attempts with exponential backoff).
        A permanent failure dead-letters the message and the next
        scheduled run retries from the same checkpoint.
    """

    backend: str
    prompt_dir: str
    opencode_bin: str
    ollama_base_url: str
    ollama_api_key: str | None
    default_model: str
    extract_model: str
    extract_command_prefix: str
    timeout: int
    max_tokens: int

    @classmethod
    def from_env(cls) -> "LLMConfig":
        return cls(
            backend=os.environ.get("INBOXOS_LLM_BACKEND", "ollama").strip().lower(),
            prompt_dir=os.environ.get("OPENCODE_PROMPT_DIR", "./.opencode/commands"),
            opencode_bin=os.environ.get("OPENCODE_BIN", "opencode"),
            ollama_base_url=os.environ.get(
                "OLLAMA_BASE_URL", "https://ollama-frp.cloudinit.dev/v1"
            ).rstrip("/"),
            ollama_api_key=os.environ.get("OLLAMA_API_KEY") or None,
            default_model=os.environ.get("INBOXOS_LLM_MODEL", "gemma4:e4b"),
            extract_model=os.environ.get("INBOXOS_LLM_MODEL_EXTRACT", "gemma4:e4b"),
            extract_command_prefix=os.environ.get(
                "INBOXOS_EXTRACT_COMMAND_PREFIX", "inboxos:extract-"
            ),
            timeout=int(os.environ.get("INBOXOS_LLM_TIMEOUT", "120")),
            # 4096 leaves enough headroom for a structured 11-field
            # JSON extract on a 2k-char body. 1024 (the previous
            # default) ran out of room on the very first 86KB
            # Substack email because the small model spent its
            # entire budget on chain-of-thought reasoning.
            max_tokens=int(os.environ.get("INBOXOS_LLM_MAX_TOKENS", "4096")),
        )

    def model_for_command(self, command_name: str) -> str:
        """Pick the configured model for a given prompt command name.

        Extract commands route to the larger `extract_model`; every
        other command (classify, summarise, ad-hoc) uses the cheap
        `default_model`. The prefix match is `startswith` so any
        future `inboxos:extract-*-v2` still routes to the big model.
        """
        if (
            self.extract_model
            and self.extract_command_prefix
            and command_name.startswith(self.extract_command_prefix)
        ):
            return self.extract_model
        return self.default_model


def _load_prompt(prompt_dir: str, command_name: str) -> tuple[str, str | None]:
    """Load a prompt .md file, returning (system_body, declared_model_or_none).

    The file is the standard opencode command format: YAML
    frontmatter delimited by `---` lines, followed by markdown body.
    The `model:` field, if present, is the model's `provider/name`
    string; we extract the `name` half as the default Ollama model.
    """
    path = Path(prompt_dir) / f"{command_name}.md"
    raw = path.read_text(encoding="utf-8")
    declared_model: str | None = None
    body = raw
    if raw.startswith("---"):
        # Split on the second `---` line.
        parts = raw.split("\n---\n", 2)
        if len(parts) >= 2:
            front = parts[0].lstrip("-").strip()
            body = parts[1] if len(parts) == 2 else "\n---\n".join(parts[1:])
            for line in front.splitlines():
                line = line.strip()
                if line.startswith("model:"):
                    value = line.split(":", 1)[1].strip()
                    declared_model = value.split("/", 1)[-1] or None
                    break
    return body.strip(), declared_model


@retry(
    reraise=True,
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=2, min=2, max=30),
    retry=retry_if_exception_type((requests.RequestException, ValueError, RuntimeError)),
)
def ollama_run_command(
    cfg: LLMConfig,
    command_name: str,
    payload: dict[str, Any],
    logger: logging.Logger | None = None,
) -> dict[str, Any]:
    """Call an Ollama-served model directly with the named prompt.

    The system message is the prompt's markdown body (frontmatter
    stripped). The user message is the JSON-serialised payload.
    The model's reply is parsed as JSON. Retryable on transient
    HTTP errors, JSON parse errors, and explicit RuntimeErrors
    raised by the parser.

    Model selection: the per-step model from `LLMConfig` (driven
    by `INBOXOS_LLM_MODEL` / `INBOXOS_LLM_MODEL_EXTRACT`) wins;
    otherwise the prompt's `model:` frontmatter is used as a
    fallback. This lets the operator pin a pipeline-level model
    (e.g. `gemma4:31b-cloud` for live) without editing the
    prompt files, while still letting a prompt declare its own
    default when no env override is set.
    """
    system_prompt, declared_model = _load_prompt(cfg.prompt_dir, command_name)
    model = cfg.model_for_command(command_name) or declared_model
    return _ollama_call_one_model(cfg, command_name, payload, model, logger=logger)


def _ollama_call_one_model(
    cfg: LLMConfig,
    command_name: str,
    payload: dict[str, Any],
    model: str,
    logger: logging.Logger | None = None,
) -> dict[str, Any]:
    """Single-shot Ollama call. No retries, no escalation.

    The retry policy lives in the `ollama_run_command` decorator
    (tenacity, 3 attempts with exponential backoff). One model per
    call; no chain (see LLMConfig docstring for the INB-21 board
    decision).
    """
    system_prompt, _declared = _load_prompt(cfg.prompt_dir, command_name)

    url = f"{cfg.ollama_base_url}/chat/completions"
    headers = {"Content-Type": "application/json"}
    if cfg.ollama_api_key:
        headers["Authorization"] = f"Bearer {cfg.ollama_api_key}"

    body = {
        "model": model,
        "messages": [
            {"role": "system", "content": system_prompt},
            {"role": "user", "content": json.dumps(payload, default=str)},
        ],
        "stream": False,
        "temperature": 0,
        "max_tokens": cfg.max_tokens,
    }

    if logger is not None:
        log_event(
            logger,
            "llm_call",
            service="llm",
            backend="ollama",
            modelID=model,
            command=command_name,
            url=url,
        )

    # LLM metrics (REQ-INB-23). When LLM_METRICS_ENABLED is truthy
    # (default), wrap the HTTP POST in time.monotonic() and increment
    # the per-process `llm_call_seconds_sum` and `llm_call_count`
    # accumulators. The metrics line emitted at the end of main()
    # reads these accumulators via os.environ (set by the wrapper).
    # The gate exists so operators can opt out of timing overhead
    # during high-throughput canary runs.
    import time as _time
    _t0 = _time.monotonic()
    try:
        resp = requests.post(url, headers=headers, json=body, timeout=cfg.timeout)
    finally:
        if os.environ.get("LLM_METRICS_ENABLED", "true").strip().lower() not in (
            "false", "0", "no", "off"
        ):
            _dt = _time.monotonic() - _t0
            _prev_sum = float(os.environ.get("INBOXOS_LLM_CALL_SECONDS_SUM", "0") or 0)
            _prev_count = int(os.environ.get("INBOXOS_LLM_CALL_COUNT", "0") or 0)
            os.environ["INBOXOS_LLM_CALL_SECONDS_SUM"] = str(_prev_sum + _dt)
            os.environ["INBOXOS_LLM_CALL_COUNT"] = str(_prev_count + 1)
    resp.raise_for_status()
    data = resp.json()

    try:
        content = data["choices"][0]["message"]["content"]
    except (KeyError, IndexError, TypeError) as exc:
        raise RuntimeError(f"Ollama response missing message content: {data!r}") from exc

    if not content or not content.strip():
        raise RuntimeError(f"Ollama returned empty content: {data!r}")

    # Find the first balanced JSON object in the response.
    text = content.strip()
    decoder = json.JSONDecoder()
    for i, ch in enumerate(text):
        if ch != "{":
            continue
        try:
            obj, _end = decoder.raw_decode(text[i:])
        except json.JSONDecodeError:
            continue
        return obj
    raise RuntimeError(f"Ollama response did not contain a JSON object: {text[:500]!r}")


@retry(
    reraise=True,
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=2, min=2, max=30),
)
def llm_run_command(
    cfg: LLMConfig,
    command_name: str,
    payload: dict[str, Any],
    logger: logging.Logger | None = None,
) -> dict[str, Any]:
    """Backend-dispatching LLM call.

    Default backend is `ollama` (see `LLMConfig`). The legacy
    `opencode` backend shells out to the opencode CLI and is kept
    for offline tests.

    v1.4 REQ-INB-57: If the payload contains a "body" field, the body
    is PII-scrubbed before being sent to the LLM. The audit trail is
    written to _artifacts/pii_scrub.jsonl (when configured).
    """
    scrubbed_payload = payload
    if "body" in payload and isinstance(payload["body"], str) and payload["body"]:
        try:
            from pii_scrub import scrub_pii
            scrubbed_body, mask_summary = scrub_pii(payload["body"])
            scrubbed_payload = {**payload, "body": scrubbed_body}
            # Audit trail
            artifacts_dir = os.environ.get("INBOXOS_ARTIFACTS_DIR", "pipelines/_artifacts")
            os.makedirs(artifacts_dir, exist_ok=True)
            audit_path = os.path.join(artifacts_dir, "pii_scrub.jsonl")
            audit_entry = {
                "ts": now_iso(),
                "command": command_name,
                "mid": payload.get("Message-ID", payload.get("mid", "?")),
                "mask_summary": mask_summary.to_dict(),
            }
            with open(audit_path, "a") as f:
                f.write(json.dumps(audit_entry) + "\n")
        except Exception as exc:
            if logger:
                logger.warning("PII scrub failed (passing through raw body): %s", exc)
            scrubbed_payload = payload

    if cfg.backend == "ollama":
        return ollama_run_command(cfg, command_name, scrubbed_payload, logger=logger)
    if cfg.backend == "opencode":
        return opencode_run_command(
            OpenCodeConfig(bin=cfg.opencode_bin, prompt_dir=cfg.prompt_dir),
            command_name,
            scrubbed_payload,
            timeout=cfg.timeout,
        )
    raise RuntimeError(f"Unknown INBOXOS_LLM_BACKEND: {cfg.backend!r}")


# ── Papra (selective write-through) ────────────────────
#
# The InboxOS pipelines write a *curated subset* of processed emails to
# Papra for long-term storage and searchability. The policy is described
# in docs/papra-write-policy.md and the INB-27 plan r2. Briefly:
#
#   1. If the email has a MIME attachment, the attachment is the document
#      of record. Each qualifying attachment is uploaded to Papra.
#   2. Otherwise, if the classifier routed the email to a module in
#      PAPRA_BODY_WRITE_MODULES, the email body is uploaded to Papra.
#   3. Otherwise, nothing is written. The raw .eml stays in SeaweedFS
#      and the structured fields stay in NocoDB.
#
# All writes go through `papra_write_attachment` and `papra_write_body`.
# Both helpers honor `PapraConfig.dry_run` -- when True (the default for
# the first ship), no HTTP request is made; the helper just logs the
# decision and returns a synthetic id. A pure `papra_decide` function
# returns the decision without making any side effects, so unit tests
# can cover the policy without monkey-patching HTTP.
#
# Failure handling: real Papra failures are non-fatal. The helper logs
# `papra_write_failed` with the error, returns None, and the pipeline
# continues. The NocoDB row can be marked Papra_Status=pending by the
# caller once a state index exists (INB-27.b).


def _csv_env(name: str, default: list[str]) -> list[str]:
    raw = os.environ.get(name, "").strip()
    if not raw:
        return list(default)
    return [part.strip().lower() for part in raw.split(",") if part.strip()]


@dataclass
class PapraConfig:
    """Configuration for the Papra write-through.

    All fields are env-driven. The historical pipeline workflow already
    exports PAPRA_BASE_URL / PAPRA_API_TOKEN / PAPRA_ORGANIZATION_ID as
    optional (falls back silently when missing); the live pipeline
    workflow is being upgraded to export them as well.

    `dry_run` defaults to True. The first ship goes out dry-run so we
    can audit the policy's decisions on real traffic before flipping
    the real-write switch.
    """

    base_url: str
    api_token: str
    organization_id: str
    body_write_modules: list[str]
    metadata_only_modules: list[str]
    dry_run: bool
    dry_run_override: bool | None
    timeout: int

    @classmethod
    def from_env(
        cls,
        *,
        body_modules_default: list[str] | None = None,
        metadata_only_default: list[str] | None = None,
    ) -> "PapraConfig":
        base = os.environ.get("PAPRA_BASE_URL", "").rstrip("/")
        token = os.environ.get("PAPRA_API_TOKEN", "")
        org = os.environ.get("PAPRA_ORGANIZATION_ID", "")
        body_default = body_modules_default if body_modules_default is not None else ["crm", "finance", "legal"]
        meta_default = metadata_only_default if metadata_only_default is not None else ["intel", "promotions"]
        # Override precedence (REQ-INB-20):
        #   PAPRA_DRY_RUN_OVERRIDE (per-message kill switch) >
        #   PAPRA_DRY_RUN (repo var) > default true
        override_raw = os.environ.get("PAPRA_DRY_RUN_OVERRIDE", "").strip().lower()
        if override_raw in ("true", "1", "yes", "on"):
            dry_run_override: bool | None = True
        elif override_raw in ("false", "0", "no", "off"):
            dry_run_override = False
        else:
            dry_run_override = None  # empty string = no override

        repo_dry_run = os.environ.get("PAPRA_DRY_RUN", "true").strip().lower() not in ("false", "0", "no", "off")
        effective_dry_run = dry_run_override if dry_run_override is not None else repo_dry_run

        return cls(
            base_url=base,
            api_token=token,
            organization_id=org,
            body_write_modules=_csv_env("PAPRA_BODY_WRITE_MODULES", body_default),
            metadata_only_modules=_csv_env("PAPRA_METADATA_ONLY_MODULES", meta_default),
            dry_run=effective_dry_run,
            dry_run_override=dry_run_override,
            timeout=int(os.environ.get("PAPRA_HTTP_TIMEOUT", "60")),
        )

    def is_configured(self) -> bool:
        """True iff base_url, api_token, and organization_id are all set."""
        return bool(self.base_url and self.api_token and self.organization_id)

    def upload_url(self) -> str:
        if not self.base_url:
            raise RuntimeError("PapraConfig.base_url is empty")
        return f"{self.base_url}/api/organizations/{self.organization_id}/documents"


def emit_papra_write_gate_event(cfg: "PapraConfig", logger: Any) -> None:
    """Emit the `papra_write_gate` structured log event (REQ-INB-20).

    Called once at the start of the Papra branch in each pipeline
    (T2.2 wires the call-sites). Carries the three fields an
    operator needs to audit the live-write gate:

    - `dry_run`: the effective dry-run flag (after
      PAPRA_DRY_RUN_OVERRIDE > PAPRA_DRY_RUN > default).
    - `dry_run_override`: the parsed override (True / False / None).
    - `papra_dry_run_env`: the raw repo-var value, so the canary
      reviewer can verify the precedence.

    Returns None; no side effects beyond the log event.
    """
    raw_repo_var = os.environ.get("PAPRA_DRY_RUN", "")
    log_event(
        logger,
        "papra_write_gate",
        dry_run=cfg.dry_run,
        dry_run_override=cfg.dry_run_override,
        papra_dry_run_env=raw_repo_var,
    )


@dataclass
class PapraTarget:
    """One concrete document to upload to Papra."""

    kind: str  # "attachment" | "body"
    filename: str  # logical name; the original filename for attachments, derived for body
    content_type: str
    payload: bytes
    content_hash: str  # sha256 hex

    def short(self) -> dict[str, Any]:
        return {
            "kind": self.kind,
            "filename": self.filename,
            "content_type": self.content_type,
            "content_hash": self.content_hash,
            "bytes": len(self.payload),
        }


@dataclass
class PapraDecision:
    """The result of `papra_decide` -- what would be written, and why."""

    would_write: bool
    reason: str
    module: str | None  # the module that triggered a body write, if any
    targets: list[PapraTarget]  # the actual upload units (may be empty)

    def to_event(self, message_id: str) -> dict[str, Any]:
        return {
            "message_id": message_id,
            "would_write": self.would_write,
            "reason": self.reason,
            "module": self.module,
            "targets": [t.short() for t in self.targets],
            "target_count": len(self.targets),
        }


def _content_hash(payload: bytes) -> str:
    import hashlib
    return hashlib.sha256(payload).hexdigest()


def _strip_html_to_text(raw: str) -> str:
    """Best-effort HTML-to-text for body uploads.

    Same algorithm as live_pipeline._strip_html_to_text; duplicated here
    so common.py does not have to import live_pipeline. Behaviour
    matches the live pipeline exactly.
    """
    if not raw:
        return ""
    no_blocks = re.sub(
        r"<(script|style|head)[^>]*>.*?</\1>",
        "",
        raw,
        flags=re.DOTALL | re.IGNORECASE,
    )
    no_blocks = re.sub(r"<!--.*?-->", "", no_blocks, flags=re.DOTALL)
    no_blocks = re.sub(
        r"</(p|div|li|h[1-6]|br|tr|td|th)\s*>",
        "\n",
        no_blocks,
        flags=re.IGNORECASE,
    )
    no_blocks = re.sub(r"<br\s*/?>", "\n", no_blocks, flags=re.IGNORECASE)
    no_blocks = re.sub(r"<[^>]+>", "", no_blocks)
    import html as _html
    no_blocks = _html.unescape(no_blocks)
    no_blocks = re.sub(r"[ \t]+", " ", no_blocks)
    no_blocks = re.sub(r"\n{3,}", "\n\n", no_blocks)
    return no_blocks.strip()


def collect_attachment_targets(msg: Any) -> list[PapraTarget]:
    """Walk the MIME tree and return one PapraTarget per real attachment.

    A "real attachment" is a part with a Content-Disposition containing
    the word "attachment" AND a non-empty decoded payload. Inline images
    and signature blobs are skipped.
    """
    targets: list[PapraTarget] = []
    for part in msg.walk():
        disp = str(part.get("Content-Disposition", ""))
        if "attachment" not in disp:
            continue
        payload = part.get_payload(decode=True) or b""
        if not payload:
            continue
        filename = part.get_filename() or f"attachment-{len(targets)}.bin"
        ct = part.get_content_type() or "application/octet-stream"
        targets.append(
            PapraTarget(
                kind="attachment",
                filename=filename,
                content_type=ct,
                payload=payload,
                content_hash=_content_hash(payload),
            )
        )
    return targets


def build_body_target(msg: Any, message_id: str, max_bytes: int = 64 * 1024) -> PapraTarget:
    """Build a PapraTarget for the email body, with attachments removed.

    The body is decoded from the best text/plain or text/html part,
    normalised to plain text if HTML, and capped at `max_bytes`.
    """
    body_text = ""
    try:
        candidate = msg.get_body(preferencelist=("plain", "html"))
        if candidate is not None:
            content = candidate.get_content()
            if isinstance(content, bytes):
                try:
                    content = content.decode(candidate.get_content_charset() or "utf-8", "replace")
                except Exception:
                    content = content.decode("utf-8", "replace")
            if candidate.get_content_type() == "text/html":
                body_text = _strip_html_to_text(content)
            else:
                body_text = content or ""
    except Exception:
        body_text = ""

    if len(body_text) > max_bytes:
        body_text = body_text[:max_bytes] + f"\n\n[truncated {len(body_text) - max_bytes} chars]"

    payload = body_text.encode("utf-8")
    safe_mid = (message_id or "unknown").replace("/", "_").replace("\\", "_")
    return PapraTarget(
        kind="body",
        filename=f"{safe_mid}.txt",
        content_type="text/plain; charset=utf-8",
        payload=payload,
        content_hash=_content_hash(payload),
    )


def papra_decide(
    *,
    message_id: str,
    modules: list[str],
    msg: Any,
    cfg: PapraConfig,
) -> PapraDecision:
    """Pure function: decide what to write to Papra for one message.

    The decision is a *plan*, not an action. Callers must pass it to
    `papra_write_attachment` / `papra_write_body` (which honor dry-run)
    to materialise.

    Policy (mirrors docs/papra-write-policy.md):

    1. If the message has >= 1 attachment -> write the attachment(s).
       The body is *not* written alongside an attachment.
    2. Else, if any module in `modules` is in `cfg.body_write_modules`,
       pick the first match (canonical ordering = list order) and write
       the body once.
    3. Else -> skip.
    """
    attachments = collect_attachment_targets(msg)
    if attachments:
        return PapraDecision(
            would_write=True,
            reason="attachment_present",
            module=None,
            targets=attachments,
        )

    for module in modules:
        if module in cfg.body_write_modules:
            body_target = build_body_target(msg, message_id)
            if not body_target.payload:
                return PapraDecision(
                    would_write=False,
                    reason="body_module_but_empty_body",
                    module=module,
                    targets=[],
                )
            return PapraDecision(
                would_write=True,
                reason="body_module",
                module=module,
                targets=[body_target],
            )
        if module in cfg.metadata_only_modules:
            return PapraDecision(
                would_write=False,
                reason="metadata_only_module",
                module=module,
                targets=[],
            )

    return PapraDecision(
        would_write=False,
        reason="no_matching_module",
        module=None,
        targets=[],
    )


@retry(
    reraise=False,
    stop=stop_after_attempt(3),
    wait=wait_exponential(multiplier=2, min=2, max=15),
    retry=retry_if_exception_type((requests.RequestException, OSError, IOError)),
)
def _papra_post_document(
    cfg: PapraConfig,
    target: PapraTarget,
    extra_tags: list[str],
) -> str | None:
    """POST one target to Papra. Returns the document id, or None on failure.

    Retries on transient network errors. Non-retryable HTTP errors raise
    immediately and are caught by the caller.
    """
    headers = {"Authorization": f"Bearer {cfg.api_token}"}
    tags = list(extra_tags) + [f"message_id:{target.content_hash[:12]}"]
    resp = requests.post(
        cfg.upload_url(),
        headers=headers,
        files={"file": (target.filename, target.payload, target.content_type)},
        data={"tags": ",".join(tags)},
        timeout=cfg.timeout,
    )
    if not resp.ok:
        raise RuntimeError(
            f"Papra upload failed: status={resp.status_code} body={resp.text[:500]!r} "
            f"filename={target.filename} kind={target.kind}"
        )
    doc_id = resp.json().get("id")
    return doc_id


def papra_write_attachment(
    cfg: PapraConfig,
    decision: PapraDecision,
    *,
    message_id: str,
    module: str | None,
    logger: Any,
) -> list[str]:
    """Materialize the decision for the attachment kind.

    Returns the list of Papra document ids actually written (dry-run
    returns synthetic ids starting with `dryrun-`). Always emits a
    `papra_decision` event plus, per target, a `papra_write_ok` or
    `papra_write_failed` event. Never raises.
    """
    log_event(
        logger,
        "papra_decision",
        message_id=message_id,
        would_write=decision.would_write,
        reason=decision.reason,
        module=decision.module,
        target_count=len(decision.targets),
        dry_run=cfg.dry_run,
    )
    if not decision.would_write:
        return []
    if not decision.targets:
        return []

    written: list[str] = []
    extra_tags = ["inboxos"]
    if module:
        extra_tags.append(f"module:{module}")

    for target in decision.targets:
        if cfg.dry_run:
            synthetic = f"dryrun-{target.content_hash[:16]}"
            log_event(
                logger,
                "papra_write_dry_run",
                message_id=message_id,
                kind=target.kind,
                filename=target.filename,
                content_hash=target.content_hash,
                synthetic_id=synthetic,
            )
            written.append(synthetic)
            continue

        if not cfg.is_configured():
            log_event(
                logger,
                "papra_skipped_no_config",
                message_id=message_id,
                kind=target.kind,
                filename=target.filename,
            )
            continue

        try:
            doc_id = _papra_post_document(cfg, target, extra_tags)
        except Exception as exc:  # noqa: BLE001
            log_event(
                logger,
                "papra_write_failed",
                message_id=message_id,
                kind=target.kind,
                filename=target.filename,
                content_hash=target.content_hash,
                error=str(exc),
            )
            continue
        log_event(
            logger,
            "papra_write_ok",
            message_id=message_id,
            kind=target.kind,
            filename=target.filename,
            content_hash=target.content_hash,
            papra_document_id=doc_id,
        )
        written.append(doc_id or "")
    return [w for w in written if w]


def papra_write_body(
    cfg: PapraConfig,
    decision: PapraDecision,
    *,
    message_id: str,
    logger: Any,
) -> str | None:
    """Materialize the decision for the body kind. Returns the first written id, or None.

    Thin wrapper over `papra_write_attachment` -- kept as a separate
    symbol so call sites read clearly.
    """
    if decision.reason != "body_module":
        return None
    ids = papra_write_attachment(
        cfg,
        decision,
        message_id=message_id,
        module=decision.module,
        logger=logger,
    )
    return ids[0] if ids else None


# ── Misc helpers ───────────────────────────────────────

def _require(name: str) -> str:
    val = os.environ.get(name, "").strip()
    if not val:
        raise RuntimeError(f"Required env var {name} is not set")
    return val


def now_iso() -> str:
    return datetime.now(timezone.utc).isoformat()


def safe_message_id(raw: str | None) -> str | None:
    """Return a stable Message-ID stripped of angle brackets, or None."""
    if not raw:
        return None
    mid = raw.strip()
    if mid.startswith("<") and mid.endswith(">"):
        mid = mid[1:-1]
    return mid or None


# ── Operator audit artifacts (v1.4 REQ-INB-49 + REQ-INB-52) ────

ARTIFACTS_DIR = os.environ.get("INBOXOS_ARTIFACTS_DIR", "pipelines/_artifacts")
METRICS_DIR = os.environ.get("INBOXOS_METRICS_DIR", "pipelines/_metrics")
LAST_RUN_PATH = os.path.join(ARTIFACTS_DIR, "last_run.json")
HEALTHCHECK_PATH = os.path.join(METRICS_DIR, "healthcheck.json")


def write_last_run_artifact(
    run_id: str,
    run_type: str,
    started_at: str,
    finished_at: str,
    status: str,
    mids_total: int = 0,
    mids_upserted: int = 0,
    mids_skipped: int = 0,
    mids_errored: int = 0,
    modules: Optional[dict[str, int]] = None,
    error_summary: Optional[list[dict]] = None,
) -> None:
    """Write the last-run audit cursor. Idempotent: overwrites on each call."""
    os.makedirs(ARTIFACTS_DIR, exist_ok=True)
    payload = {
        "run_id": run_id,
        "run_type": run_type,
        "started_at": started_at,
        "finished_at": finished_at,
        "status": status,
        "mids_total": mids_total,
        "mids_upserted": mids_upserted,
        "mids_skipped": mids_skipped,
        "mids_errored": mids_errored,
        "modules": modules or {"finance": 0, "crm": 0, "intel": 0, "promotions": 0},
        "error_summary": error_summary or [],
    }
    with open(LAST_RUN_PATH, "w") as f:
        json.dump(payload, f, indent=2)


def write_healthcheck_artifact(
    last_run_ts: str,
    last_run_status: str,
    pending_count: int = 0,
    error_count_24h: int = 0,
    archive_size_gb: float = 0.0,
    modules: Optional[dict[str, int]] = None,
) -> None:
    """Write the healthcheck JSON. Idempotent: overwrites on each call."""
    os.makedirs(METRICS_DIR, exist_ok=True)
    payload = {
        "last_run_ts": last_run_ts,
        "last_run_status": last_run_status,
        "pending_count": pending_count,
        "error_count_24h": error_count_24h,
        "archive_size_gb": archive_size_gb,
        "modules": modules or {"finance": 0, "crm": 0, "intel": 0, "promotions": 0},
    }
    with open(HEALTHCHECK_PATH, "w") as f:
        json.dump(payload, f, indent=2)