ops-warden/src/warden/worker.py

"""ops-warden coordination worker (WARDEN-WP-0020).

Pulls ops-warden's unread State Hub coordination requests and turns each into a
**plan** of ops-warden actions. This module is the llm-connect-independent foundation
(T1): the inbox client, the plan model, the deterministic ``RuleBrain`` default, the
guardrail allowlist, and the dry-run renderer. The llm-connect brain (T2) and the
executing dispatcher (T3) plug into the same ``Brain`` protocol and ``WorkerPlan``.

Guardrails live here, not in the brain — the allowlist and no-secret invariant are
enforced on every action *regardless* of what the brain proposes, so an LLM (or a
prompt-injected message) cannot widen ops-warden's authority. Dry-run is the default;
nothing executes in T1.
"""
from __future__ import annotations

import os
import re
from dataclasses import dataclass, field
from typing import List, Optional, Protocol

import httpx

DEFAULT_HUB_URL = "http://127.0.0.1:8000"
WORKER_AGENT = "ops-warden"

# Actions the worker may take autonomously. Anything else escalates to a human.
ALLOWED_ACTION_KINDS = frozenset(
    {"route_answer", "reply", "mark_read", "propose_catalog_diff", "progress_note"}
)

# Signals that a task would breach the conduit-not-broker boundary (handle a secret
# value) or touch production config / irreversible state — always escalate, never auto.
_SECRET_SIGNS = re.compile(
    r"\b(token value|secret value|raw token|api[_ ]?key|password|private key|"
    r"vault[_ ]?token|npm_auth_token|client[_ ]?secret|credential value)\b",
    re.IGNORECASE,
)
_PROD_SIGNS = re.compile(
    r"\b(policy\.enabled|prod flip|production config|enable the gate|"
    r"~/\.config/warden/warden\.yaml|deploy to prod)\b",
    re.IGNORECASE,
)
# A routing/credential question the worker can answer read-only.
_ROUTING_SIGNS = re.compile(
    r"\b(where|which subsystem|how do i (get|obtain)|route|who owns|"
    r"credential|warden route|warden access)\b",
    re.IGNORECASE,
)


@dataclass
class PlannedAction:
    kind: str
    summary: str
    payload: dict = field(default_factory=dict)
    # filled by the guardrail pass: "safe" or "escalate" (+ reason when escalated)
    risk: str = "safe"
    reason: str = ""


@dataclass
class WorkerPlan:
    message_id: str
    from_agent: str
    subject: str
    actions: List[PlannedAction] = field(default_factory=list)

    @property
    def escalated(self) -> bool:
        return any(a.risk == "escalate" for a in self.actions) or not self.actions


class Brain(Protocol):
    """Turns one inbox message into a proposed WorkerPlan. Pure: no side effects."""

    def plan(self, message: dict) -> WorkerPlan: ...


def validate_action(action: PlannedAction, message: dict) -> Optional[str]:
    """Return a rejection reason if the action must escalate, else None.

    Defense-in-depth: enforced on every action regardless of what the brain proposed.
    """
    if action.kind not in ALLOWED_ACTION_KINDS:
        return f"action kind {action.kind!r} is not on the allowlist"
    blob = f"{message.get('subject', '')} {message.get('body', '')} {action.summary}"
    if action.kind in ("reply", "route_answer", "progress_note", "propose_catalog_diff"):
        # These are fine in general, but never when the task is about a secret *value*
        # or a production-config change — those need a human.
        if _SECRET_SIGNS.search(blob):
            return "task involves a secret value (conduit-not-broker — never auto-handled)"
        if _PROD_SIGNS.search(blob):
            return "task touches production config (requires explicit human approval)"
    return None


def _guardrail(plan: WorkerPlan, message: dict) -> WorkerPlan:
    """Downgrade any action that fails validation to an escalation. Brain-agnostic."""
    for a in plan.actions:
        reason = validate_action(a, message)
        if reason:
            a.risk = "escalate"
            a.reason = reason
    return plan


class RuleBrain:
    """Deterministic, no-LLM brain for the scaffold + tests.

    Conservative by design: it only proposes a read-only routing answer for clear
    routing questions, and escalates everything else to a human. The llm-connect brain
    (T2) replaces this with real reasoning over the same WorkerPlan contract.
    """

    def plan(self, message: dict) -> WorkerPlan:
        wp = WorkerPlan(
            message_id=str(message.get("id", "")),
            from_agent=str(message.get("from_agent", "")),
            subject=str(message.get("subject", "")),
        )
        blob = f"{message.get('subject', '')} {message.get('body', '')}"
        if _SECRET_SIGNS.search(blob) or _PROD_SIGNS.search(blob):
            return wp  # no actions → escalates
        if _ROUTING_SIGNS.search(blob):
            wp.actions.append(
                PlannedAction(
                    kind="route_answer",
                    summary="Answer the routing/credential question via `warden route`/`access`.",
                    payload={"query": message.get("subject", "")},
                )
            )
        return wp  # otherwise no actions → escalates to a human


class HubClient:
    """Minimal read client for the State Hub inbox (honors WARDEN_HUB_URL)."""

    def __init__(self, base_url: Optional[str] = None, timeout: float = 10.0):
        self.base_url = (base_url or os.environ.get("WARDEN_HUB_URL", DEFAULT_HUB_URL)).rstrip("/")
        self.timeout = timeout

    def unread(self, to_agent: str = WORKER_AGENT) -> List[dict]:
        url = f"{self.base_url}/messages/"
        resp = httpx.get(
            url, params={"to_agent": to_agent, "unread_only": "true"}, timeout=self.timeout
        )
        resp.raise_for_status()
        data = resp.json()
        return data if isinstance(data, list) else []


def build_plans(messages: List[dict], brain: Brain) -> List[WorkerPlan]:
    """Plan every message and apply the guardrail pass. Pure — no execution."""
    return [_guardrail(brain.plan(m), m) for m in messages]


def render_plans(plans: List[WorkerPlan]) -> str:
    """Human-readable dry-run rendering."""
    if not plans:
        return "inbox empty — no coordination requests for ops-warden."
    lines: List[str] = []
    for p in plans:
        tag = "ESCALATE" if p.escalated else "AUTO"
        lines.append(f"[{tag}] {p.from_agent}: {p.subject}  ({p.message_id})")
        if not p.actions:
            lines.append("    · no in-scope action — hand to a human")
        for a in p.actions:
            mark = "→" if a.risk == "safe" else "⚠"
            lines.append(f"    {mark} {a.kind}: {a.summary}")
            if a.risk == "escalate":
                lines.append(f"       escalated: {a.reason}")
    return "\n".join(lines)