feat(kb): apply-worker — auto-land approved proposals (stage 2 automation) (#36)

* feat(kb): apply-worker to auto-land approved proposals (stage 2 automation)

Event-driven worker that turns a HUMAN-approved kb_stage proposal into canonical
state, so an approval in Telegram surfaces in Leo's identity without Leo applying
its own work.

- Fires only on status='approved' (never pending_review) -> proposer != applier
  holds; the human approval stays the trigger. No auto-approve anywhere.
- Reuses scripts/apply_proposal.py verbatim as the sole apply path (same txn,
  rowcount=1 guard, FK stamp). Connects as the narrow kb_apply role, never
  superuser, never inside the hermes harness.
- Render hook (--render-cmd / KB_APPLY_RENDER_CMD) is inert until the SOUL
  renderer (PR2) is deployed; applying still works, rendered SOUL just lags.
- Ships INERT: report-only unless --enable / KB_APPLY_WORKER_ENABLED=1. systemd
  oneshot service + 5min timer, both shipped disabled.
- 10 unit tests; candidate query validated read-only vs prod (0 applyable today).

* fix(kb): apply-worker --max-per-tick cap + poison-pill retry ceiling

Fixer draft-exit items:
- --max-per-tick=1 (default): an enabled worker lands applies one-at-a-time
  and observably instead of draining the whole approved queue in one tick.
- --max-attempts=3 ceiling with a persisted failure-count state file: a
  deterministically-failing approved proposal is treated as a poison pill and
  skipped after N consecutive failures, instead of retrying every tick forever.
  State persists on disk because the worker runs oneshot per timer tick.

Both are inert until the worker is enabled; it still ships disabled.

scripts/apply_worker.py

@@ -0,0 +1,303 @@
+#!/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
+-> renderer runs -> Leo reads the new self.
+
+This is the "natural evolution" engine: it makes an approval in Telegram surface
+in Leo's identity 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 render step is a configurable hook (``--render-cmd`` / ``KB_APPLY_RENDER_CMD``).
+  Until the SOUL renderer (PR2) is deployed the hook is simply absent and the worker
+  logs that render was skipped -- applying still works, Leo's canonical is correct,
+  and the rendered SOUL.md just lags until the renderer exists.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import subprocess
+import sys
+from pathlib import Path
+from typing import Any, Dict, List, Optional
+
+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 = ("revise_strategy", "add_edge", "attach_evidence")
+
+# Only a revise_strategy changes an agent's identity spine, so only it triggers a
+# SOUL re-render. Evidence/edge applies change the graph but not the rendered self.
+RENDER_TYPES = ("revise_strategy",)
+
+
+# --------------------------------------------------------------------------- #
+# Pure helpers (unit-tested without a DB)                                      #
+# --------------------------------------------------------------------------- #
+def build_candidate_query(types: tuple = WORKER_TYPES, limit: int = 20) -> 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.
+    """
+    type_list = ", ".join(ap.sql_literal(t) for t in types)
+    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'
+ 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: Optional[str], agent_id: Optional[str]) -> Optional[List[str]]:
+    """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: Optional[str]) -> 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:  # noqa: BLE001  -- a bad state file must never wedge the worker
+        return {}
+
+
+def save_failure_state(path: Optional[str], 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``: already at/over the failure ceiling -> never re-attempted,
+      surfaced loudly for the operator (needs a human fix, not another retry).
+    - ``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) -> List[Dict[str, Any]]:
+    sql = build_candidate_query(limit=args.limit)
+    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: Optional[str]) -> 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)
+    candidates = fetch_candidates(args, password)
+
+    if not candidates:
+        print("no approved+applyable proposals; nothing to do")
+        return 0
+
+    failure_counts = load_failure_state(args.failure_state_file)
+    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:  # noqa: BLE001  -- one bad proposal must not stop the batch
+            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: Optional[List[str]] = None) -> int:
+    return run(parse_args(sys.argv[1:] if argv is None else argv))
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

systemd/teleo-kb-apply-worker.service

@@ -0,0 +1,23 @@
+[Unit]
+Description=Teleo KB apply worker — lands human-approved proposals into canonical
+After=network.target docker.service
+Wants=network.target
+
+[Service]
+Type=oneshot
+User=teleo
+Group=teleo
+# Runs from the git-backed deploy-infra worktree (scripts/ is NOT rsynced into the
+# live pipeline tree; the reviewed copy is invoked in place). Operator tool, not a
+# hermes runtime component.
+ExecStart=/opt/teleo-eval/pipeline/.venv/bin/python3 \
+  /opt/teleo-eval/workspaces/deploy-infra/scripts/apply_worker.py
+# Ships INERT: no --enable here, so each tick is report-only until an operator opts
+# in. Flip on (after the first manual apply is proven) by dropping an EnvironmentFile
+# with KB_APPLY_WORKER_ENABLED=1 and KB_APPLY_RENDER_CMD=... — no unit edit needed.
+EnvironmentFile=-/opt/teleo-eval/secrets/kb-apply-worker.env
+StandardOutput=journal
+StandardError=journal
+
+[Install]
+WantedBy=multi-user.target

systemd/teleo-kb-apply-worker.timer

@@ -0,0 +1,13 @@
+[Unit]
+Description=Poll for approved KB proposals and apply them (kb-apply worker)
+
+[Timer]
+# Applies are rare and human-gated, so a slow poll is plenty. Matches the
+# teleo-timer pattern (oneshot service + timer) rather than adding LISTEN/NOTIFY.
+OnBootSec=3min
+OnUnitActiveSec=5min
+AccuracySec=30s
+Persistent=true
+
+[Install]
+WantedBy=timers.target

tests/test_apply_worker.py

@@ -0,0 +1,199 @@
+"""Unit tests for scripts/apply_worker.py.
+
+Pure tests: query building, candidate parsing, render-hook expansion, and the
+report-only enablement gate -- no live database. Runnable under pytest or
+standalone (`python3 tests/test_apply_worker.py`).
+"""
+
+import argparse
+from pathlib import Path
+import sys
+
+try:
+    import pytest
+except ImportError:  # standalone fallback so the file runs without pytest installed
+    import contextlib
+    import types
+
+    pytest = types.SimpleNamespace()
+
+    @contextlib.contextmanager
+    def _raises(exc):
+        try:
+            yield
+        except exc:
+            return
+        raise AssertionError(f"expected {exc.__name__} to be raised")
+
+    pytest.raises = _raises
+
+REPO_ROOT = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO_ROOT / "scripts"))
+
+import apply_worker as w  # noqa: E402
+
+
+# --- candidate query ------------------------------------------------------ #
+def test_candidate_query_filters_on_approved():
+    sql = w.build_candidate_query()
+    # The structural guarantee: worker can only ever act on approved proposals.
+    assert "status = 'approved'" in sql
+    # Never selects pending_review (no auto-approve possible).
+    assert "pending_review" not in sql
+
+
+def test_candidate_query_requires_apply_payload():
+    sql = w.build_candidate_query()
+    assert "payload ? 'apply_payload'" in sql
+
+
+def test_candidate_query_lists_worker_types_only():
+    sql = w.build_candidate_query()
+    for t in w.WORKER_TYPES:
+        assert f"'{t}'" in sql
+    # a non-applyable type must not appear
+    assert "reject_claim" not in sql
+
+
+# --- candidate parsing ---------------------------------------------------- #
+def test_parse_candidates_skips_blank_lines():
+    out = '{"id": "a", "proposal_type": "add_edge", "agent_id": null}\n\n'
+    rows = w.parse_candidates(out)
+    assert len(rows) == 1
+    assert rows[0]["id"] == "a"
+
+
+# --- render hook ---------------------------------------------------------- #
+def test_render_command_none_when_unset():
+    assert w.build_render_command("", "agent-1") is None
+
+
+def test_render_command_none_without_agent():
+    assert w.build_render_command("python3 render_soul.py --agent-id {agent_id}", None) is None
+
+
+def test_render_command_expands_agent_id():
+    cmd = w.build_render_command("python3 render_soul.py --agent-id {agent_id}", "abc")
+    assert cmd == ["python3", "render_soul.py", "--agent-id", "abc"]
+
+
+# --- enablement gate ------------------------------------------------------ #
+def _args(**over):
+    base = dict(
+        enable=False, limit=20, max_per_tick=1, max_attempts=3,
+        failure_state_file=None, applied_by="kb-apply",
+        apply_script="apply_proposal.py", render_cmd="",
+        secrets_file="x", container="teleo-pg", db="teleo",
+        host="127.0.0.1", role="kb_apply",
+    )
+    base.update(over)
+    return argparse.Namespace(**base)
+
+
+def test_report_only_gate_does_not_apply(monkeypatch=None):
+    calls = []
+    w.ap.load_password = lambda _f: "pw"
+    w.fetch_candidates = lambda a, p: [{"id": "p1", "proposal_type": "revise_strategy", "agent_id": "ag"}]
+    w.apply_one = lambda a, pid: calls.append(pid)
+
+    rc = w.run(_args(enable=False))
+    assert rc == 0
+    assert calls == []  # disabled worker performs NO writes
+
+
+def test_enabled_worker_applies_and_renders():
+    applied, rendered = [], []
+    w.ap.load_password = lambda _f: "pw"
+    w.fetch_candidates = lambda a, p: [{"id": "p1", "proposal_type": "revise_strategy", "agent_id": "ag"}]
+    w.apply_one = lambda a, pid: applied.append(pid)
+    w.render_one = lambda a, agent_id: rendered.append(agent_id) or "ok"
+
+    rc = w.run(_args(enable=True))
+    assert rc == 0
+    assert applied == ["p1"]
+    assert rendered == ["ag"]  # revise_strategy triggers a render
+
+
+def test_enabled_worker_skips_render_for_add_edge():
+    applied, rendered = [], []
+    w.ap.load_password = lambda _f: "pw"
+    w.fetch_candidates = lambda a, p: [{"id": "e1", "proposal_type": "add_edge", "agent_id": None}]
+    w.apply_one = lambda a, pid: applied.append(pid)
+    w.render_one = lambda a, agent_id: rendered.append(agent_id) or "ok"
+
+    rc = w.run(_args(enable=True))
+    assert rc == 0
+    assert applied == ["e1"]
+    assert rendered == []  # add_edge changes the graph, not the rendered self
+
+
+# --- per-tick cap + poison-pill ceiling ----------------------------------- #
+def _cands(*ids):
+    return [{"id": i, "proposal_type": "add_edge", "agent_id": None} for i in ids]
+
+
+def test_partition_caps_to_max_per_tick():
+    split = w.partition_candidates(_cands("a", "b", "c"), {}, max_attempts=3, max_per_tick=1)
+    assert [c["id"] for c in split["to_apply"]] == ["a"]  # only 1 applied per tick
+    assert split["poisoned"] == []
+
+
+def test_partition_skips_poison_pill_over_ceiling():
+    split = w.partition_candidates(_cands("a", "b"), {"a": 3}, max_attempts=3, max_per_tick=5)
+    assert [c["id"] for c in split["poisoned"]] == ["a"]  # a is poisoned, never retried
+    assert [c["id"] for c in split["to_apply"]] == ["b"]  # b still eligible
+
+
+def test_enabled_worker_caps_applies_at_one_per_tick():
+    applied = []
+    w.ap.load_password = lambda _f: "pw"
+    w.fetch_candidates = lambda a, p: _cands("e1", "e2", "e3")
+    w.apply_one = lambda a, pid: applied.append(pid)
+
+    rc = w.run(_args(enable=True, max_per_tick=1))
+    assert rc == 0
+    assert applied == ["e1"]  # capped to 1 even with 3 approved candidates
+
+
+def test_failure_state_roundtrip_and_ceiling(tmp_path=None):
+    import tempfile
+
+    d = tempfile.mkdtemp()
+    path = str(Path(d) / "failures.json")
+    w.save_failure_state(path, {"p1": 2})
+    assert w.load_failure_state(path) == {"p1": 2}
+    # p1 at 2 with max_attempts 3 is still eligible; at 3 it poisons
+    split = w.partition_candidates(_cands("p1"), {"p1": 3}, max_attempts=3, max_per_tick=1)
+    assert [c["id"] for c in split["poisoned"]] == ["p1"]
+
+
+def test_apply_failure_increments_persisted_count():
+    import tempfile
+
+    path = str(Path(tempfile.mkdtemp()) / "failures.json")
+    w.ap.load_password = lambda _f: "pw"
+    w.fetch_candidates = lambda a, p: _cands("bad")
+
+    def _boom(a, pid):
+        raise RuntimeError("deterministic failure")
+
+    w.apply_one = _boom
+    rc = w.run(_args(enable=True, failure_state_file=path, max_per_tick=1))
+    assert rc == 1
+    assert w.load_failure_state(path).get("bad") == 1  # count persisted for next tick
+
+
+if __name__ == "__main__":
+    import traceback
+
+    failures = 0
+    for name, fn in sorted(globals().items()):
+        if name.startswith("test_") and callable(fn):
+            try:
+                fn()
+                print(f"PASS {name}")
+            except Exception:  # noqa: BLE001
+                failures += 1
+                print(f"FAIL {name}")
+                traceback.print_exc()
+    sys.exit(1 if failures else 0)