teleo-infrastructure/scripts/apply_worker.py

#!/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 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",)


# --------------------------------------------------------------------------- #
# 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)]}


# --------------------------------------------------------------------------- #
# 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 via the audited apply_proposal.py CLI -- same txn + rowcount guard."""
    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,
    ]
    result = subprocess.run(cmd, text=True, capture_output=True, check=False)
    if result.returncode != 0:
        raise RuntimeError(
            f"apply failed for {proposal_id}: {result.stdout.strip()} {result.stderr.strip()}"
        )


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 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]}/"
                  f"{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(
        "--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())