diff --git a/scripts/apply_worker.py b/scripts/apply_worker.py new file mode 100644 index 0000000..f49a504 --- /dev/null +++ b/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()) diff --git a/systemd/teleo-kb-apply-worker.service b/systemd/teleo-kb-apply-worker.service new file mode 100644 index 0000000..d9cabde --- /dev/null +++ b/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 diff --git a/systemd/teleo-kb-apply-worker.timer b/systemd/teleo-kb-apply-worker.timer new file mode 100644 index 0000000..a875b01 --- /dev/null +++ b/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 diff --git a/tests/test_apply_worker.py b/tests/test_apply_worker.py new file mode 100644 index 0000000..2c855e0 --- /dev/null +++ b/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)