feat(WARDEN-WP-0020): conservative triage tier as the --execute default (Option A)

Per Bernd's call: the guardrails prevent security harm but not LLM content errors, so the
worker should triage + draft, not auto-send, until reply quality is proven (matches the
build-stage/recoverability posture).

run_conservative triages NEW messages into a reviewed digest (state_dir/worker-digest.md)
with drafted replies, posts ONE progress note, tracks seen message ids (schedule-safe
dedup), and sends NOTHING to other agents / marks nothing read. `warden worker run
--execute` now runs this conservative tier; `--full-auto` opts into the auto-send path.

Live-verified with the LLM brain on the real inbox: produced a high-quality draft reply to
a secrets-engine coordination message and correctly flagged the llm-connect custody request
as NEEDS YOU. Conservative mode is safe to schedule (T4). 244 tests, lint clean.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

src/warden/worker.py

@@ -16,6 +16,7 @@ from __future__ import annotations
 import os
 import re
 from dataclasses import dataclass, field
+from pathlib import Path
 from typing import List, Optional, Protocol
 
 import httpx
@@ -329,7 +330,7 @@ def execute_plan(plan: WorkerPlan, hub: HubClient, *, topic_id: Optional[str] =
 
 
 def execute_plans(plans: List[WorkerPlan], hub: HubClient, *, topic_id: Optional[str] = None) -> str:
-    """Execute every plan and return a human-readable audit summary."""
+    """FULL-AUTO: execute every plan's safe actions and return an audit summary."""
     lines: List[str] = []
     for p in plans:
         results = execute_plan(p, hub, topic_id=topic_id)
@@ -339,6 +340,85 @@ def execute_plans(plans: List[WorkerPlan], hub: HubClient, *, topic_id: Optional
     return "\n".join(lines) if lines else "inbox empty — nothing to execute."
 
 
+# --- conservative tier (default for --execute): triage + draft, never auto-send ----------
+
+def default_state_dir() -> Path:
+    return Path(os.environ.get("WARDEN_STATE_DIR", str(Path.home() / ".local" / "state" / "warden")))
+
+
+def load_seen(state_dir: Path) -> set:
+    import json as _json
+
+    p = state_dir / "worker-seen.json"
+    if not p.exists():
+        return set()
+    try:
+        return set(_json.loads(p.read_text()))
+    except (ValueError, OSError):
+        return set()
+
+
+def save_seen(state_dir: Path, seen: set) -> None:
+    import json as _json
+
+    (state_dir / "worker-seen.json").write_text(_json.dumps(sorted(seen)))
+
+
+def build_digest(plans: List[WorkerPlan]) -> str:
+    """Human-reviewable digest of proposed actions + drafted replies. Sends nothing."""
+    if not plans:
+        return "No new coordination requests."
+    lines: List[str] = []
+    for p in plans:
+        tag = "NEEDS YOU" if p.escalated else "DRAFT READY"
+        lines.append(f"## [{tag}] {p.from_agent}: {p.subject}  ({p.message_id})")
+        if not p.actions:
+            lines.append("- no in-scope action — handle directly")
+        for a in p.actions:
+            if a.risk == "escalate":
+                lines.append(f"- escalated ({a.reason}): {a.summary}")
+            elif a.kind == "route_answer" and a.payload.get("answer"):
+                lines.append(f"- proposed answer: {a.payload['answer']}")
+            elif a.kind == "reply" and a.payload.get("body"):
+                lines.append(f"- proposed reply: {a.payload['body']}")
+            else:
+                lines.append(f"- {a.kind}: {a.summary}")
+        lines.append("")
+    return "\n".join(lines).rstrip()
+
+
+def run_conservative(
+    plans: List[WorkerPlan], hub: HubClient, *, topic_id: Optional[str] = None,
+    state_dir: Optional[Path] = None,
+) -> str:
+    """Triage NEW messages into a reviewed digest. No agent-facing sends, no mark-read.
+
+    Safe to schedule: it only surfaces what's waiting (with drafted replies for you to
+    approve), tracks which messages it has already digested, and posts one progress note
+    so a scheduled run is visible. The operator approves/sends the good drafts.
+    """
+    state_dir = state_dir or default_state_dir()
+    state_dir.mkdir(parents=True, exist_ok=True)
+    seen = load_seen(state_dir)
+    new = [p for p in plans if p.message_id and p.message_id not in seen]
+    digest = build_digest(new)
+    (state_dir / "worker-digest.md").write_text(digest + "\n")
+    if new:
+        n_esc = sum(1 for p in new if p.escalated)
+        try:
+            hub.add_progress(
+                summary=(
+                    f"[worker] triaged {len(new)} new message(s): {len(new) - n_esc} with "
+                    f"drafted replies, {n_esc} need you. Drafts: {state_dir / 'worker-digest.md'}"
+                ),
+                topic_id=topic_id,
+            )
+        except Exception:  # noqa: BLE001 — a note failure must not lose the digest
+            pass
+        save_seen(state_dir, seen | {p.message_id for p in new})
+    return digest
+
+
 def draft_route_answer(query: str) -> str:
     """Compute the routing answer the worker would send for a query. Read-only.