#!/usr/bin/env python3
"""Event-driven worker that lands human-approved proposals into canonical state.

Stage 2-of-the-loop automation:  Leo proposes -> HUMAN approves -> WORKER applies
-> a supported render hook may run -> Leo reads canonical state.

This is the "natural evolution" engine: it makes an approval in Telegram canonical
without Leo ever applying its own work. It does NOT think, does NOT approve, and
does NOT create proposals -- it only acts on proposals a human has already moved
to ``status='approved'``.

Governance boundary (why this is safe)
--------------------------------------
* Proposer != applier. The worker fires ONLY on ``status='approved'``; it never
  touches ``pending_review`` and never auto-approves. The human approval stays the
  trigger, so "Leo proposes but does not self-apply" holds.
* It connects as the narrow ``kb_apply`` role (never superuser, never Leo's creds)
  and reuses ``apply_proposal.py`` verbatim as the apply path -- same transaction,
  same ``rowcount=1`` concurrency guard, same FK stamp. No new write logic here.
* It runs as an operator-side systemd unit, NOT inside the hermes harness.

Safety posture
--------------
* Gated OFF by default. Without ``--enable`` (or ``KB_APPLY_WORKER_ENABLED=1``) the
  worker only REPORTS what it would apply -- it performs no writes. Flip it on only
  after the first apply has been proven by hand.
* The optional render step is a configurable hook (``--render-cmd`` /
  ``KB_APPLY_RENDER_CMD``). Only ``revise_strategy`` currently has a proven
  agent-owned render contract. Other apply types update canonical state without
  invoking a generic agent render.
"""

from __future__ import annotations

import argparse
import json
import os
import stat
import subprocess
import sys
from pathlib import Path
from typing import Any

HERE = Path(__file__).resolve().parent
sys.path.insert(0, str(HERE))

import apply_proposal as ap  # noqa: E402  (sibling module, reused verbatim)

# Types the worker is willing to auto-apply. Deliberately the same set the engine
# supports; a proposal type outside this list is ignored (never applied).
WORKER_TYPES = ap.APPLYABLE_TYPES

# Only strategy revisions have a proven agent-owned render hook. Canonical claim
# bundles and evidence/edge applies do not trigger a generic identity render.
RENDER_TYPES = ("revise_strategy",)

POSTCOMMIT_RECEIPT_FAILURE_MARKER = "committed, but private replay receipt capture failed"
POSTCOMMIT_RECEIPT_RECOVERY_MARKER = ". Recover without reapplying via: "


class PostCommitReceiptError(RuntimeError):
    """The proposal committed, but its private row-level receipt is unavailable."""


# --------------------------------------------------------------------------- #
# Pure helpers (unit-tested without a DB)                                      #
# --------------------------------------------------------------------------- #
def build_candidate_query(
    types: tuple = WORKER_TYPES,
    limit: int = 20,
    excluded_ids: tuple = (),
) -> str:
    """SELECT approved, applyable proposals that carry an apply_payload.

    Filters on status='approved' at the DB level -- this is the structural
    guarantee that the worker cannot act on anything a human has not approved.
    IDs at the failure ceiling are excluded before LIMIT so exhausted rows cannot
    occupy every fetched slot and starve later eligible proposals.
    """
    type_list = ", ".join(ap.sql_literal(t) for t in types)
    excluded = tuple(str(proposal_id) for proposal_id in excluded_ids)
    exclusion_clause = ""
    if excluded:
        excluded_list = ", ".join(ap.sql_literal(proposal_id) for proposal_id in excluded)
        exclusion_clause = f"\n   and id::text not in ({excluded_list})"
    return f"""select jsonb_build_object(
        'id', id::text,
        'proposal_type', proposal_type,
        'agent_id', payload->'apply_payload'->>'agent_id')::text
  from kb_stage.kb_proposals
 where status = 'approved'
   and proposal_type in ({type_list})
   and payload ? 'apply_payload'{exclusion_clause}
 order by created_at asc
 limit {int(limit)};"""


def parse_candidates(psql_output: str) -> list[dict[str, Any]]:
    """Parse newline-delimited JSON rows from psql -At output."""
    rows: list[dict[str, Any]] = []
    for line in psql_output.splitlines():
        line = line.strip()
        if line:
            rows.append(json.loads(line))
    return rows


def build_render_command(render_cmd: str | None, agent_id: str | None) -> list[str] | None:
    """Expand the render-hook template for one agent, or None if no hook configured.

    ``render_cmd`` is a shell-word template, e.g. ``python3 render_soul.py --agent-id {agent_id}``.
    Returns a token list ready for subprocess, or None when unset (render skipped).
    """
    if not render_cmd:
        return None
    if not agent_id:
        return None
    import shlex

    return [tok.format(agent_id=agent_id) for tok in shlex.split(render_cmd)]


def load_failure_state(path: str | None) -> dict[str, int]:
    """Load the persisted {proposal_id: consecutive_failure_count} map.

    The worker runs oneshot per timer tick, so in-memory failure counts do not
    survive between ticks. A poison-pill proposal (deterministically failing but
    stuck at 'approved') would otherwise be re-selected and re-attempted every
    tick forever. Persisting the count on disk lets the ceiling actually bite.
    Missing/corrupt file -> empty map (fail open to "no known failures").
    """
    if not path:
        return {}
    p = Path(path)
    if not p.is_file():
        return {}
    try:
        data = json.loads(p.read_text(encoding="utf-8"))
        return {str(k): int(v) for k, v in data.items()} if isinstance(data, dict) else {}
    except Exception:
        return {}


def save_failure_state(path: str | None, state: dict[str, int]) -> None:
    if not path:
        return
    p = Path(path)
    p.parent.mkdir(parents=True, exist_ok=True)
    p.write_text(json.dumps(state, sort_keys=True), encoding="utf-8")


def partition_candidates(
    candidates: list[dict[str, Any]],
    failure_counts: dict[str, int],
    max_attempts: int,
    max_per_tick: int,
) -> dict[str, list[dict[str, Any]]]:
    """Split fetched candidates into what to apply now vs. skip.

    - ``poisoned``: defensive post-fetch check for a candidate already at/over
      the failure ceiling. Normal DB fetches exclude these IDs before LIMIT.
    - ``to_apply``: the first ``max_per_tick`` non-poisoned candidates -> capped
      so an enabled worker lands applies one-at-a-time and observably, rather
      than draining the whole approved queue in a single unobserved tick.
    """
    poisoned = [c for c in candidates if failure_counts.get(c["id"], 0) >= max_attempts]
    eligible = [c for c in candidates if failure_counts.get(c["id"], 0) < max_attempts]
    return {"poisoned": poisoned, "to_apply": eligible[: max(0, max_per_tick)]}


def has_exact_postcommit_receipt_failure(stderr: str | None, proposal_id: str) -> bool:
    """Recognize only the proposal-bound post-COMMIT message from apply_proposal.

    The old marker can occur inside payload-validation tracebacks (for example,
    as an attacker-controlled unknown field name).  Requiring the complete
    engine message shape prevents those pre-COMMIT failures from bypassing the
    normal failure and poison-pill accounting.
    """
    prefix = f"proposal {proposal_id} {POSTCOMMIT_RECEIPT_FAILURE_MARKER}: "
    recovery_suffix = f" {proposal_id} --receipt-only"
    for line in (stderr or "").splitlines():
        candidate = line.rstrip("\r")
        if (
            candidate.startswith(prefix)
            and POSTCOMMIT_RECEIPT_RECOVERY_MARKER in candidate[len(prefix) :]
            and candidate.endswith(recovery_suffix)
        ):
            return True
    return False


def receipt_path_for_proposal(args: argparse.Namespace, proposal_id: str) -> Path:
    """Return the explicit, deterministic private receipt path for one proposal."""
    receipt_dir = (
        getattr(args, "receipt_dir", None)
        or os.environ.get("KB_APPLY_WORKER_RECEIPT_DIR")
        or os.environ.get("KB_APPLY_RECEIPT_DIR")
        or ap.replay_receipt.DEFAULT_RECEIPT_DIR
    )
    return Path(receipt_dir).expanduser().resolve() / f"{proposal_id}.json"


def assert_private_replay_receipt(path: Path, proposal_id: str) -> None:
    """Verify the worker can rely on the exact private row-level receipt."""
    try:
        path_stat = path.lstat()
    except OSError as exc:
        raise RuntimeError("private replay receipt was not written") from exc
    if not stat.S_ISREG(path_stat.st_mode) or path.is_symlink():
        raise RuntimeError("private replay receipt is not a regular file")
    if stat.S_IMODE(path_stat.st_mode) & 0o077:
        raise RuntimeError("private replay receipt permissions are broader than 0600")
    try:
        receipt = json.loads(path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError) as exc:
        raise RuntimeError("private replay receipt is unreadable or invalid") from exc
    if not isinstance(receipt, dict):
        raise RuntimeError("private replay receipt is not a JSON object")
    try:
        ap.replay_receipt.validate_replay_receipt(
            receipt,
            expected_proposal_id=str(proposal_id),
        )
    except ValueError as exc:
        raise RuntimeError("private replay receipt row-level contract is invalid") from exc


# --------------------------------------------------------------------------- #
# Side-effecting steps                                                         #
# --------------------------------------------------------------------------- #
def _psql_args(args: argparse.Namespace) -> argparse.Namespace:
    """Namespace shaped for ap.run_psql (reuses the kb_apply connection path)."""
    return argparse.Namespace(container=args.container, db=args.db, host=args.host, role=args.role)


def fetch_candidates(
    args: argparse.Namespace,
    password: str,
    excluded_ids: tuple,
) -> list[dict[str, Any]]:
    sql = build_candidate_query(limit=args.limit, excluded_ids=excluded_ids)
    out = ap.run_psql(_psql_args(args), sql, password)
    return parse_candidates(out)


def apply_one(args: argparse.Namespace, proposal_id: str) -> None:
    """Apply once and retain or safely recover the exact private replay receipt."""
    receipt_path = receipt_path_for_proposal(args, proposal_id)
    cmd = [
        sys.executable,
        str(args.apply_script),
        proposal_id,
        "--applied-by",
        args.applied_by,
        "--secrets-file",
        args.secrets_file,
        "--container",
        args.container,
        "--db",
        args.db,
        "--host",
        args.host,
        "--role",
        args.role,
        "--receipt-dir",
        str(receipt_path.parent),
        "--receipt-out",
        str(receipt_path),
    ]
    result = subprocess.run(cmd, text=True, capture_output=True, check=False)
    committed = result.returncode == 0 or has_exact_postcommit_receipt_failure(
        result.stderr,
        proposal_id,
    )
    if not committed:
        raise RuntimeError(
            f"apply failed for {proposal_id}: {(result.stdout or '').strip()} {(result.stderr or '').strip()}"
        )

    if result.returncode == 0:
        try:
            assert_private_replay_receipt(receipt_path, proposal_id)
            return
        except RuntimeError:
            # The engine returned only after COMMIT. A missing/invalid receipt is
            # recoverable without replaying the apply transaction.
            pass

    try:
        recovery = subprocess.run(
            [*cmd, "--receipt-only"],
            text=True,
            capture_output=True,
            check=False,
        )
    except OSError as exc:
        raise PostCommitReceiptError(
            f"proposal {proposal_id} committed, but the read-only receipt recovery "
            "command could not start; do not retry the apply transaction"
        ) from exc
    if recovery.returncode != 0:
        raise PostCommitReceiptError(
            f"proposal {proposal_id} committed, but read-only receipt recovery "
            f"failed (exit {recovery.returncode}); do not retry the apply transaction"
        )
    try:
        assert_private_replay_receipt(receipt_path, proposal_id)
    except RuntimeError as exc:
        raise PostCommitReceiptError(
            f"proposal {proposal_id} committed, but receipt recovery did not produce "
            "a valid private row-level receipt; do not retry the apply transaction"
        ) from exc


def render_one(args: argparse.Namespace, agent_id: str | None) -> str:
    cmd = build_render_command(args.render_cmd, agent_id)
    if cmd is None:
        return "render skipped (no render-cmd configured; PR2 renderer not deployed)"
    result = subprocess.run(cmd, text=True, capture_output=True, check=False)
    if result.returncode != 0:
        raise RuntimeError(f"render failed for agent {agent_id}: {result.stdout.strip()} {result.stderr.strip()}")
    return f"rendered agent {agent_id}"


# --------------------------------------------------------------------------- #
# Main                                                                         #
# --------------------------------------------------------------------------- #
def run(args: argparse.Namespace) -> int:
    password = ap.load_password(args.secrets_file)
    failure_counts = load_failure_state(args.failure_state_file)
    exhausted_ids = tuple(proposal_id for proposal_id, count in failure_counts.items() if count >= args.max_attempts)
    candidates = fetch_candidates(args, password, exhausted_ids)

    for proposal_id in exhausted_ids:
        print(
            f"EXCLUDE failure-ceiling ID {proposal_id}: failed "
            f"{failure_counts[proposal_id]}x >= max-attempts {args.max_attempts}; "
            "omitted from candidate SQL before LIMIT",
            file=sys.stderr,
        )

    if not candidates:
        print("no approved+applyable proposals; nothing to do")
        return 0

    split = partition_candidates(candidates, failure_counts, args.max_attempts, args.max_per_tick)

    for c in split["poisoned"]:
        print(
            f"SKIP poison-pill {c['id']} ({c['proposal_type']}): failed "
            f"{failure_counts.get(c['id'], 0)}x >= max-attempts {args.max_attempts}; "
            f"needs a human fix, not another retry",
            file=sys.stderr,
        )

    enabled = args.enable or os.environ.get("KB_APPLY_WORKER_ENABLED") == "1"
    if not enabled:
        print(
            f"[report-only] worker disabled; {len(split['to_apply'])} proposal(s) would apply "
            f"this tick (cap {args.max_per_tick}):"
        )
        for c in split["to_apply"]:
            print(f"  would apply {c['id']} ({c['proposal_type']}) agent={c.get('agent_id')}")
        print("enable with --enable or KB_APPLY_WORKER_ENABLED=1 after the first manual apply is proven")
        return 0

    failures = 0
    for c in split["to_apply"]:
        pid, ptype, agent_id = c["id"], c["proposal_type"], c.get("agent_id")
        try:
            apply_one(args, pid)
            failure_counts.pop(pid, None)  # success clears any prior failure count
            print(f"applied {pid} ({ptype})")
            if ptype in RENDER_TYPES:
                print("  " + render_one(args, agent_id))
        except PostCommitReceiptError as exc:
            failures += 1
            # The canonical transaction already committed. Do not classify this
            # as an apply failure or poison/retry the proposal; receipt-only is
            # the sole safe recovery path.
            print(
                f"POST-COMMIT RECEIPT ERROR for {pid} ({ptype}): {exc}",
                file=sys.stderr,
            )
        except Exception as exc:
            failures += 1
            # Leave the proposal at 'approved' so a fixed one reapplies next tick
            # (the rowcount=1 guard makes reapply safe), but bump its failure count
            # so a deterministically-failing proposal hits the ceiling instead of
            # retrying forever. Surface loudly for the operator.
            failure_counts[pid] = failure_counts.get(pid, 0) + 1
            print(
                f"ERROR applying {pid} ({ptype}) [attempt {failure_counts[pid]}/{args.max_attempts}]: {exc}",
                file=sys.stderr,
            )

    save_failure_state(args.failure_state_file, failure_counts)
    return 1 if failures else 0


def parse_args(argv: list[str]) -> argparse.Namespace:
    p = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter)
    p.add_argument(
        "--enable",
        action="store_true",
        help="actually apply (default: report-only). Also honored via KB_APPLY_WORKER_ENABLED=1.",
    )
    p.add_argument("--limit", type=int, default=20, help="max candidates fetched per tick")
    p.add_argument(
        "--max-per-tick",
        type=int,
        default=1,
        help="max proposals actually APPLIED per tick (default 1: applies land "
        "one-at-a-time and observably, not a whole-queue drain)",
    )
    p.add_argument(
        "--max-attempts",
        type=int,
        default=3,
        help="consecutive apply failures before a proposal is treated as a "
        "poison pill and skipped (needs a human fix, not endless retries)",
    )
    p.add_argument(
        "--failure-state-file",
        default=os.environ.get("KB_APPLY_WORKER_STATE", "/opt/teleo-eval/logs/kb-apply-worker-failures.json"),
        help="persisted per-proposal failure counts (survives oneshot ticks so the poison-pill ceiling actually bites)",
    )
    p.add_argument(
        "--applied-by",
        default=ap.SERVICE_AGENT_HANDLE,
        help="handle recorded as applied_by (default: the kb-apply service agent)",
    )
    p.add_argument(
        "--apply-script",
        default=str(HERE / "apply_proposal.py"),
        help="path to the apply_proposal.py engine (the sole apply path)",
    )
    p.add_argument(
        "--receipt-dir",
        default=(
            os.environ.get("KB_APPLY_WORKER_RECEIPT_DIR")
            or os.environ.get("KB_APPLY_RECEIPT_DIR")
            or ap.replay_receipt.DEFAULT_RECEIPT_DIR
        ),
        help="private replay receipt directory; each apply writes the deterministic "
        "<proposal-id>.json path and verifies 0600 permissions",
    )
    p.add_argument(
        "--render-cmd",
        default=os.environ.get("KB_APPLY_RENDER_CMD", ""),
        help="render hook template, e.g. 'python3 render_soul.py --agent-id {agent_id}'. "
        "Empty (default) skips render until the SOUL renderer is deployed.",
    )
    p.add_argument("--secrets-file", default=ap.DEFAULT_SECRETS_FILE)
    p.add_argument("--container", default=ap.DEFAULT_CONTAINER)
    p.add_argument("--db", default=ap.DEFAULT_DB)
    p.add_argument("--host", default=ap.DEFAULT_HOST)
    p.add_argument("--role", default=ap.DEFAULT_ROLE)
    return p.parse_args(argv)


def main(argv: list[str] | None = None) -> int:
    return run(parse_args(sys.argv[1:] if argv is None else argv))


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