diff --git a/.agent/AGENTS.md b/.agent/AGENTS.md
index 4191b53..c324b26 100644
--- a/.agent/AGENTS.md
+++ b/.agent/AGENTS.md
@@ -7,10 +7,34 @@ same memory, skills, and protocols.
 ## Memory (read in this order)
 - `memory/personal/PREFERENCES.md` — stable user conventions
 - `memory/working/WORKSPACE.md` — current task state
-- `memory/semantic/LESSONS.md` — distilled patterns, read before decisions
+- `memory/working/REVIEW_QUEUE.md` — pending candidate lessons waiting for you
 - `memory/semantic/DECISIONS.md` — past architectural choices
+- `memory/semantic/LESSONS.md` — distilled patterns (rendered from `lessons.jsonl`)
 - `memory/episodic/AGENT_LEARNINGS.jsonl` — raw experience log (top-k by salience)
 
+## Review Queue (host-agent responsibility)
+
+Candidate lessons are clustered + staged automatically by `memory/auto_dream.py`.
+The host agent — you — does the actual review using the CLI tools below.
+
+Check `memory/working/REVIEW_QUEUE.md` at session start. If pending > 10 or
+oldest staged > 7 days, review before substantive work.
+
+Workflow:
+1. `python .agent/tools/list_candidates.py` — pending candidates, sorted by priority
+2. For each: decide accept / reject / defer based on claim, evidence_ids,
+   cluster_size, and any contradictions with existing LESSONS.md
+3. `python .agent/tools/graduate.py <id> --rationale "..."` to accept
+4. `python .agent/tools/reject.py <id> --reason "..."` to reject
+5. `python .agent/tools/reopen.py <id>` to requeue a previously-rejected item
+6. Review in a **batch**, not one-by-one — cross-candidate contradictions
+   only surface when you see multiple at once.
+
+The heuristic prefilter in `memory/validate.py` has already dropped obvious
+junk (too-short claims, exact duplicates). Everything staged needs real
+judgment. Rationale is required for graduation — rubber-stamped promotions
+are the exact failure mode this layer prevents.
+
 ## Skills
 - `skills/_index.md` — read first for discovery
 - `skills/_manifest.jsonl` — machine-readable skill metadata
@@ -24,8 +48,12 @@ same memory, skills, and protocols.
 
 ## Rules
 1. Check memory before decisions you have been corrected on before.
-2. Log every significant action to `memory/episodic/AGENT_LEARNINGS.jsonl`.
-3. Update `memory/working/WORKSPACE.md` as you work; archive on completion.
-4. Follow `protocols/permissions.md` strictly. Blocked means blocked.
-5. When a self-rewrite hook fires, propose conservative edits only.
-6. The harness is dumb on purpose. Reasoning lives in skills and memory.
+2. If `REVIEW_QUEUE.md` shows backlog past threshold, handle it before the new task.
+3. Log every significant action to `memory/episodic/AGENT_LEARNINGS.jsonl`
+   via `.agent/tools/memory_reflect.py`.
+4. Update `memory/working/WORKSPACE.md` as you work; archive on completion.
+5. Never hand-edit `memory/semantic/LESSONS.md` — it's rendered from
+   `lessons.jsonl`. Use `graduate.py` / `reject.py` instead.
+6. Follow `protocols/permissions.md`. Blocked means blocked.
+7. When a self-rewrite hook fires, propose conservative edits only.
+8. The harness is dumb on purpose. Reasoning lives in skills + the host agent.
diff --git a/.agent/harness/conductor.py b/.agent/harness/conductor.py
index a19e90a..ad5a051 100644
--- a/.agent/harness/conductor.py
+++ b/.agent/harness/conductor.py
@@ -2,35 +2,12 @@
 import os, sys
 from context_budget import build_context
 from hooks.post_execution import log_execution
+from llm import call_model
 
 RESERVED = 40000
 MAX_CTX = int(os.getenv("AGENT_MAX_CONTEXT", "128000"))
 
 
-def _call_model(system, user):
-    provider = os.getenv("AGENT_PROVIDER", "anthropic").lower()
-    if provider == "anthropic":
-        from anthropic import Anthropic
-        c = Anthropic()
-        r = c.messages.create(
-            model=os.getenv("AGENT_MODEL", "claude-sonnet-4-5"),
-            max_tokens=4096, temperature=0.3,
-            system=system,
-            messages=[{"role": "user", "content": user}],
-        )
-        return r.content[0].text
-    if provider == "openai":
-        from openai import OpenAI
-        c = OpenAI()
-        r = c.chat.completions.create(
-            model=os.getenv("AGENT_MODEL", "gpt-4o"),
-            messages=[{"role": "system", "content": system},
-                      {"role": "user", "content": user}],
-        )
-        return r.choices[0].message.content
-    raise ValueError(f"unknown provider: {provider}")
-
-
 SYSTEM_PREAMBLE = (
     "You are an agent with externalized memory, skills, and protocols.\n"
     "Your memory, skills, and constraints are in the context below.\n"
@@ -43,7 +20,7 @@ def run(user_input: str) -> str:
     context, used = build_context(user_input, budget=MAX_CTX - RESERVED)
     system = SYSTEM_PREAMBLE + context
     try:
-        result = _call_model(system, user_input)
+        result = call_model(system, user_input)
         log_execution("conductor", user_input[:100], result[:500], True)
         return result
     except Exception as e:
diff --git a/.agent/harness/context_budget.py b/.agent/harness/context_budget.py
index 4068e43..1dc7045 100644
--- a/.agent/harness/context_budget.py
+++ b/.agent/harness/context_budget.py
@@ -1,8 +1,23 @@
-"""Assemble context from memory + matched skills + protocols within a token budget."""
-import json, os
+"""Assemble context from memory + matched skills + protocols within a token budget.
+
+Query-aware: episodes and lessons are scored against user_input so the agent
+sees the memory that matters for *this* task, not just the most salient memory
+in general. Always-on slots (PREFERENCES, WORKSPACE, permissions) are loaded
+whole regardless of query — they're cheap and safety-critical.
+"""
+import json, os, re, sys
 from salience import salience_score
+from text import word_set, jaccard
 
 ROOT = os.path.join(os.path.dirname(__file__), "..")
+# skill_loader lives in tools/ — make it importable without requiring callers
+# to configure PYTHONPATH themselves
+sys.path.insert(0, os.path.join(ROOT, "tools"))
+RELEVANCE_FLOOR = 0.3  # even zero-overlap episodes surface if very salient
+
+# Keep in sync with memory/validate._extract_lesson_lines — both filters
+# want TERMINAL-only lesson content.
+_STATUS_RE = re.compile(r"status=(\w+)")
 
 
 def _read(path, limit=None):
@@ -13,11 +28,22 @@ def _read(path, limit=None):
     return content[:limit] if limit else content
 
 
-def _tokens(text):
+def _token_estimate(text):
+    """Rough chars-to-tokens estimate for budgeting."""
     return len(text) // 4
 
 
-def _top_episodes(k=5):
+def _relevance(entry_text, query_words):
+    """Fraction of query words that appear in entry. 1.0 when no query."""
+    if not query_words:
+        return 1.0
+    ew = word_set(entry_text)
+    if not ew:
+        return 0.0
+    return len(query_words & ew) / len(query_words)
+
+
+def _top_episodes(query, k=5):
     path = os.path.join(ROOT, "memory/episodic/AGENT_LEARNINGS.jsonl")
     if not os.path.exists(path):
         return ""
@@ -30,7 +56,19 @@ def _top_episodes(k=5):
             entries.append(json.loads(line))
         except json.JSONDecodeError:
             continue
-    entries.sort(key=salience_score, reverse=True)
+
+    query_words = word_set(query)
+
+    def _score(e):
+        text = " ".join([
+            e.get("action", ""),
+            e.get("reflection", ""),
+            e.get("detail", ""),
+        ])
+        rel = _relevance(text, query_words)
+        return salience_score(e) * (RELEVANCE_FLOOR + (1.0 - RELEVANCE_FLOOR) * rel)
+
+    entries.sort(key=_score, reverse=True)
     top = entries[:k]
     return "\n".join(
         f"- [{e.get('timestamp','')[:10]}] {e.get('action','')}: "
@@ -39,38 +77,104 @@ def _top_episodes(k=5):
     )
 
 
+def _lines_up_to_budget(lines, char_budget):
+    out, used = [], 0
+    for line in lines:
+        block = f"- {line}\n"
+        if used + len(block) > char_budget:
+            break
+        out.append(block)
+        used += len(block)
+    return "".join(out)
+
+
+def _top_lessons(query, lessons_md, char_budget=8000):
+    """Rank accepted lesson bullets by query overlap; fall back to original order.
+
+    Only terminal (status=accepted) lessons reach the host agent as retrievable
+    guidance. Provisional, legacy, and superseded bullets exist in LESSONS.md
+    for audit but must not be injected into the system prompt — they'd let the
+    agent act on probationary or stale memory.
+    """
+    lines = []
+    for line in (lessons_md or "").splitlines():
+        s = line.strip()
+        if not s.startswith("- ") or len(s) <= 2:
+            continue
+        # Primary status filter: HTML annotation
+        if "<!--" in s:
+            ann = s.split("<!--", 1)[1]
+            m = _STATUS_RE.search(ann)
+            if m and m.group(1) != "accepted":
+                continue
+        text = s[2:].split("<!--")[0].strip()
+        # Fallback: visual markers
+        if text.startswith("[PROVISIONAL]"):
+            continue
+        if text.startswith("~~") and text.endswith("~~"):
+            continue
+        if text:
+            lines.append(text)
+    if not lines:
+        # No accepted lessons → return empty. Returning raw markdown would
+        # leak the non-terminal content the filter is designed to block.
+        return ""
+
+    query_words = word_set(query)
+    if not query_words:
+        return _lines_up_to_budget(lines, char_budget)
+
+    scored = [(len(query_words & word_set(l)), i, l) for i, l in enumerate(lines)]
+    relevant = sorted([s for s in scored if s[0] > 0], key=lambda s: (-s[0], s[1]))
+
+    if not relevant:
+        return _lines_up_to_budget(lines, char_budget)
+    return _lines_up_to_budget([l for _, _, l in relevant], char_budget)
+
+
 def build_context(user_input: str, budget: int = 88000):
-    """Returns (context_string, tokens_used). Lean by design."""
-    from skill_loader import progressive_load  # lazy to avoid cycles
+    """Returns (context_string, tokens_used). Lean and query-aware."""
     parts, used = [], 0
 
-    # always load: personal preferences + live workspace
-    for rel in ("memory/personal/PREFERENCES.md", "memory/working/WORKSPACE.md"):
+    # always load: personal preferences + live workspace + AGENTS map + DECISIONS
+    # AGENTS.md and DECISIONS.md were missing despite AGENTS.md specifying the
+    # read order — the standalone path was not faithful to its own contract.
+    for rel in (
+        "AGENTS.md",
+        "memory/personal/PREFERENCES.md",
+        "memory/working/WORKSPACE.md",
+        "memory/working/REVIEW_QUEUE.md",
+        "memory/semantic/DECISIONS.md",
+    ):
         text = _read(rel)
         if text:
             parts.append(f"# {rel}\n{text}")
-            used += _tokens(text)
+            used += _token_estimate(text)
 
-    # semantic lessons, truncated
-    lessons = _read("memory/semantic/LESSONS.md", limit=8000)
-    if lessons:
-        parts.append(f"# LESSONS\n{lessons}")
-        used += _tokens(lessons)
+    # query-aware lessons
+    lessons_raw = _read("memory/semantic/LESSONS.md")
+    if lessons_raw:
+        lessons = _top_lessons(user_input, lessons_raw, char_budget=8000)
+        if lessons:
+            parts.append(f"# LESSONS (query-relevant)\n{lessons}")
+            used += _token_estimate(lessons)
 
-    # top episodic by salience
-    episodes = _top_episodes(k=5)
+    # query-aware top episodes
+    episodes = _top_episodes(user_input, k=5)
     if episodes:
-        parts.append(f"# RECENT EPISODES (top by salience)\n{episodes}")
-        used += _tokens(episodes)
+        parts.append(f"# RECENT EPISODES (salience x relevance)\n{episodes}")
+        used += _token_estimate(episodes)
 
-    # matched skills only
+    # matched skills only (progressive_load is already input-matched).
+    # Lazy import so a missing skill_loader doesn't kill context assembly.
     try:
+        from skill_loader import progressive_load
         skills = progressive_load(user_input)
     except Exception:
         skills = []
     for s in skills:
         block = f"## Skill: {s['name']}\n{s['content']}"
-        t = _tokens(block)
+        t = _token_estimate(block)
         if used + t < budget:
             parts.append(block)
             used += t
@@ -79,6 +183,6 @@ def build_context(user_input: str, budget: int = 88000):
     perms = _read("protocols/permissions.md")
     if perms:
         parts.append(f"# PERMISSIONS\n{perms}")
-        used += _tokens(perms)
+        used += _token_estimate(perms)
 
     return "\n\n---\n\n".join(parts), used
diff --git a/.agent/harness/hooks/_provenance.py b/.agent/harness/hooks/_provenance.py
new file mode 100644
index 0000000..0228697
--- /dev/null
+++ b/.agent/harness/hooks/_provenance.py
@@ -0,0 +1,33 @@
+"""Shared provenance helpers for episodic entries. Cached per-process."""
+import os, subprocess
+
+AGENT_ROOT = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", ".."))
+
+_CACHED_COMMIT = None
+_CACHED_RUN_ID = None
+
+
+def run_id():
+    global _CACHED_RUN_ID
+    if _CACHED_RUN_ID is None:
+        _CACHED_RUN_ID = os.environ.get("AGENT_RUN_ID", f"pid-{os.getpid()}")
+    return _CACHED_RUN_ID
+
+
+def commit_sha():
+    global _CACHED_COMMIT
+    if _CACHED_COMMIT is None:
+        try:
+            out = subprocess.run(
+                ["git", "rev-parse", "HEAD"],
+                capture_output=True, text=True, timeout=2,
+                cwd=AGENT_ROOT,
+            )
+            _CACHED_COMMIT = out.stdout.strip() if out.returncode == 0 else ""
+        except Exception:
+            _CACHED_COMMIT = ""
+    return _CACHED_COMMIT
+
+
+def build_source(skill):
+    return {"skill": skill, "run_id": run_id(), "commit_sha": commit_sha()}
diff --git a/.agent/harness/hooks/on_failure.py b/.agent/harness/hooks/on_failure.py
index 6870d33..850ff7f 100644
--- a/.agent/harness/hooks/on_failure.py
+++ b/.agent/harness/hooks/on_failure.py
@@ -1,5 +1,6 @@
 """Failures are learning. High pain score + rewrite flag after repeat offenses."""
 import json, datetime, os
+from ._provenance import build_source
 
 ROOT = os.path.join(os.path.dirname(__file__), "..", "..")
 EPISODIC = os.path.join(ROOT, "memory/episodic/AGENT_LEARNINGS.jsonl")
@@ -30,7 +31,8 @@ def _count_recent_failures(skill_name):
     return count
 
 
-def on_failure(skill_name, action, error, context=""):
+def on_failure(skill_name, action, error, context="", confidence=0.9,
+               evidence_ids=None):
     entry = {
         "timestamp": datetime.datetime.now().isoformat(),
         "skill": skill_name,
@@ -42,8 +44,13 @@ def on_failure(skill_name, action, error, context=""):
         "reflection": f"FAILURE in {skill_name}: {type(error).__name__}: "
                       f"{str(error)[:200]}",
         "context": context[:300],
+        "confidence": confidence,
+        "source": build_source(skill_name),
+        "evidence_ids": list(evidence_ids) if evidence_ids else [],
     }
-    recent = _count_recent_failures(skill_name)
+    # _count_recent_failures returns PRIOR failures only; add 1 for this one
+    # so the rewrite flag fires on the Nth failure, not the (N+1)th.
+    recent = _count_recent_failures(skill_name) + 1
     if recent >= FAILURE_THRESHOLD:
         entry["reflection"] += (
             f" | THIS SKILL HAS FAILED {recent} TIMES IN {WINDOW_DAYS}d. "
diff --git a/.agent/harness/hooks/post_execution.py b/.agent/harness/hooks/post_execution.py
index d3d590a..7bc90f4 100644
--- a/.agent/harness/hooks/post_execution.py
+++ b/.agent/harness/hooks/post_execution.py
@@ -1,12 +1,13 @@
 """Runs after every action. Appends a structured entry to episodic memory."""
 import json, datetime, os
+from ._provenance import build_source
 
 ROOT = os.path.join(os.path.dirname(__file__), "..", "..")
 EPISODIC = os.path.join(ROOT, "memory/episodic/AGENT_LEARNINGS.jsonl")
 
 
 def log_execution(skill_name, action, result, success, reflection="",
-                  importance=5):
+                  importance=5, confidence=0.5, evidence_ids=None):
     os.makedirs(os.path.dirname(EPISODIC), exist_ok=True)
     entry = {
         "timestamp": datetime.datetime.now().isoformat(),
@@ -17,6 +18,9 @@ def log_execution(skill_name, action, result, success, reflection="",
         "pain_score": 2 if success else 7,
         "importance": importance,
         "reflection": reflection,
+        "confidence": confidence,
+        "source": build_source(skill_name),
+        "evidence_ids": list(evidence_ids) if evidence_ids else [],
     }
     with open(EPISODIC, "a") as f:
         f.write(json.dumps(entry) + "\n")
diff --git a/.agent/harness/llm.py b/.agent/harness/llm.py
new file mode 100644
index 0000000..082969a
--- /dev/null
+++ b/.agent/harness/llm.py
@@ -0,0 +1,38 @@
+"""Shared model-call helper. Factored out of conductor so memory/ can reuse."""
+import os
+
+
+def llm_available():
+    """True iff provider + key are configured. Validation / dream cycle check
+    this before making calls so they degrade gracefully offline."""
+    provider = os.getenv("AGENT_PROVIDER", "anthropic").lower()
+    if provider == "anthropic":
+        return bool(os.getenv("ANTHROPIC_API_KEY"))
+    if provider == "openai":
+        return bool(os.getenv("OPENAI_API_KEY"))
+    return False
+
+
+def call_model(system, user, *, temperature=0.3, max_tokens=4096, model=None):
+    provider = os.getenv("AGENT_PROVIDER", "anthropic").lower()
+    if provider == "anthropic":
+        from anthropic import Anthropic
+        c = Anthropic()
+        r = c.messages.create(
+            model=model or os.getenv("AGENT_MODEL", "claude-sonnet-4-5"),
+            max_tokens=max_tokens, temperature=temperature,
+            system=system,
+            messages=[{"role": "user", "content": user}],
+        )
+        return r.content[0].text
+    if provider == "openai":
+        from openai import OpenAI
+        c = OpenAI()
+        r = c.chat.completions.create(
+            model=model or os.getenv("AGENT_MODEL", "gpt-4o"),
+            temperature=temperature,
+            messages=[{"role": "system", "content": system},
+                      {"role": "user", "content": user}],
+        )
+        return r.choices[0].message.content
+    raise ValueError(f"unknown provider: {provider}")
diff --git a/.agent/harness/text.py b/.agent/harness/text.py
new file mode 100644
index 0000000..8d8babf
--- /dev/null
+++ b/.agent/harness/text.py
@@ -0,0 +1,30 @@
+"""Shared text utilities for memory/retrieval scoring.
+
+Single source of truth for stopwords + lexical overlap. Both validate.py
+(dream-cycle promotion gate) and context_budget.py (retrieval) depend on it,
+so drift here would quietly make the two layers see different words.
+"""
+import re
+
+STOPWORDS = frozenset({
+    "the", "a", "an", "and", "or", "but", "is", "are", "was", "were", "be",
+    "been", "being", "have", "has", "had", "do", "does", "did", "will",
+    "would", "should", "could", "may", "might", "must", "can", "this",
+    "that", "these", "those", "of", "in", "on", "at", "to", "for", "with",
+    "by", "from", "as", "if", "then", "when", "where", "how", "why", "what",
+    "it", "its", "their", "our", "we", "you", "i", "not", "no",
+})
+
+
+def word_set(text):
+    """Lowercase content words (3+ chars, stopwords removed)."""
+    return {t.lower() for t in re.findall(r"[a-zA-Z][a-zA-Z0-9_-]{2,}", text or "")
+            if t.lower() not in STOPWORDS}
+
+
+def jaccard(a, b):
+    if not a and not b:
+        return 1.0
+    if not a or not b:
+        return 0.0
+    return len(a & b) / len(a | b)
diff --git a/.agent/memory/auto_dream.py b/.agent/memory/auto_dream.py
index 75f4cd7..1e6ffb0 100644
--- a/.agent/memory/auto_dream.py
+++ b/.agent/memory/auto_dream.py
@@ -1,13 +1,32 @@
-"""Nightly compression. Orchestrates promote + decay + archive + git snapshot."""
-import json, os, datetime, subprocess
-from promote import find_recurring_patterns, promote_to_semantic
+"""Staging-only dream cycle. Mechanical work, no reasoning.
+
+Responsibilities (in order):
+  1. load episodic entries
+  2. cluster + extract → structured patterns
+  3. stage candidates (lifecycle metadata baked in)
+  4. heuristic prefilter (length + exact-duplicate; obvious junk goes to rejected/)
+  5. decay old episodes + archive stale workspace
+  6. write REVIEW_QUEUE.md summary so the next host session sees the backlog
+
+Never:
+  - subjective validation (host agent reviews via CLI tools)
+  - promotion to LESSONS.md (graduate.py does that)
+  - git commit (unattended repo writes are dangerous on a host hook)
+"""
+import json, os
+from promote import cluster_and_extract, write_candidates
+from validate import heuristic_check
+from review_state import mark_rejected, write_review_queue_summary
 from decay import decay_old_entries
 from archive import archive_stale_workspace
 
-ROOT = os.path.abspath(os.path.dirname(__file__))  # .agent/memory/ — absolute so cron paths hold
-PROJ_ROOT = os.path.dirname(os.path.dirname(ROOT))   # project root
+ROOT = os.path.abspath(os.path.dirname(__file__))
 EPISODIC = os.path.join(ROOT, "episodic/AGENT_LEARNINGS.jsonl")
+CANDIDATES = os.path.join(ROOT, "candidates")
+SEMANTIC = os.path.join(ROOT, "semantic")
+REVIEW_QUEUE = os.path.join(ROOT, "working/REVIEW_QUEUE.md")
 PROMOTION_THRESHOLD = 7.0
+CLUSTER_SIMILARITY = 0.3
 
 
 def _load_entries():
@@ -31,33 +50,72 @@ def _write_entries(entries):
             f.write(json.dumps(e) + "\n")
 
 
-def _git_snapshot(promoted, decayed, kept):
-    msg = f"dream cycle: promoted {promoted}, decayed {decayed}, kept {kept}"
-    try:
-        subprocess.run(["git", "add", ROOT], check=False, cwd=PROJ_ROOT)
-        subprocess.run(["git", "commit", "-m", msg], check=False, cwd=PROJ_ROOT)
-    except FileNotFoundError:
-        pass
-    return msg
+def _heuristic_prefilter(candidates_dir, semantic_dir):
+    """Move obvious junk (too-short, exact duplicate) to rejected/ automatically.
+
+    Anything subjective — "is this really a useful lesson?" — is the host
+    agent's call, not this function's.
+    """
+    if not os.path.isdir(candidates_dir):
+        return 0
+    lessons_path = os.path.join(semantic_dir, "LESSONS.md")
+    existing = open(lessons_path).read() if os.path.exists(lessons_path) else ""
+    rejected = 0
+    for fname in sorted(os.listdir(candidates_dir)):
+        if not fname.endswith(".json"):
+            continue
+        path = os.path.join(candidates_dir, fname)
+        if not os.path.isfile(path):
+            continue
+        try:
+            with open(path) as f:
+                cand = json.load(f)
+        except (OSError, json.JSONDecodeError):
+            continue
+        check = heuristic_check(cand, existing)
+        if not check["passed"]:
+            reason = ", ".join(check["reasons"])
+            # Record the specific lesson(s) that triggered the duplicate
+            # rejection so write_candidates can check whether THIS blocker
+            # is still there, not just whether LESSONS.md as a whole changed.
+            mark_rejected(cand["id"], "heuristic_prefilter", reason,
+                          candidates_dir,
+                          duplicate_claims=check.get("duplicates", []))
+            rejected += 1
+    return rejected
 
 
 def run_dream_cycle():
     entries = _load_entries()
     if not entries:
-        print("dream cycle: no entries")
+        # Still refresh the review queue — candidates may have been staged in
+        # a previous cycle and the host agent loads REVIEW_QUEUE.md into every
+        # session via build_context, so a stale/missing file hides real work.
+        pending = write_review_queue_summary(CANDIDATES, REVIEW_QUEUE)
+        print(f"dream cycle: no entries (queue has {pending} pending)")
         return
 
-    recurring = find_recurring_patterns(entries)
-    promotable = [e for e in recurring.values()
-                  if e.get("_salience", 0) >= PROMOTION_THRESHOLD]
-    promoted = promote_to_semantic(promotable, semantic_dir=os.path.join(ROOT, "semantic"))
+    patterns = cluster_and_extract(entries, threshold=CLUSTER_SIMILARITY)
+    promotable = {k: p for k, p in patterns.items()
+                  if p.get("canonical_salience", 0) >= PROMOTION_THRESHOLD}
 
-    kept, archived = decay_old_entries(entries, archive_dir=os.path.join(ROOT, "episodic/snapshots"))
+    staged = write_candidates(promotable, CANDIDATES)
+    prefiltered = _heuristic_prefilter(CANDIDATES, SEMANTIC)
+
+    kept, archived = decay_old_entries(
+        entries, archive_dir=os.path.join(ROOT, "episodic/snapshots"))
     _write_entries(kept)
-    archive_stale_workspace(working_dir=os.path.join(ROOT, "working"),
-                            archive_dir=os.path.join(ROOT, "episodic/snapshots"))
+    archive_stale_workspace(
+        working_dir=os.path.join(ROOT, "working"),
+        archive_dir=os.path.join(ROOT, "episodic/snapshots"))
+
+    pending = write_review_queue_summary(CANDIDATES, REVIEW_QUEUE)
 
-    print(_git_snapshot(promoted, len(archived), len(kept)))
+    print(
+        f"dream cycle: patterns={len(patterns)} staged={staged} "
+        f"prefiltered_out={prefiltered} pending_review={pending} "
+        f"archived={len(archived)} kept={len(kept)}"
+    )
 
 
 if __name__ == "__main__":
diff --git a/.agent/memory/cluster.py b/.agent/memory/cluster.py
new file mode 100644
index 0000000..4e3964f
--- /dev/null
+++ b/.agent/memory/cluster.py
@@ -0,0 +1,123 @@
+"""Content-based clustering + deterministic pattern extraction.
+
+Phase 3's replacement for action-prefix clustering. Works without an LLM:
+similarity is Jaccard on word_set, and extraction picks a canonical episode
+rather than synthesizing a new claim. Structured candidates flow through the
+Phase 1 validation gate — if no LLM is available, they defer as before.
+"""
+import os, re, sys, hashlib
+
+sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "harness"))
+from text import word_set, jaccard
+from salience import salience_score
+
+
+def _normalize_claim(text):
+    """Lowercase, strip punctuation, collapse whitespace.
+
+    Used to derive a stable pattern id — same claim text must always produce
+    the same id so lifecycle state (decisions, rejection_count, graduation
+    status) carries across dream cycles even when the cluster membership
+    shifts by one episode. Kept in sync with validate._normalize.
+    """
+    t = re.sub(r"[^\w\s]", " ", (text or "").lower())
+    return re.sub(r"\s+", " ", t).strip()
+
+
+def _entry_features(entry):
+    """Content feature set for clustering: action + reflection + detail."""
+    text = " ".join([
+        entry.get("action", ""),
+        entry.get("reflection", ""),
+        entry.get("detail", ""),
+    ])
+    return word_set(text)
+
+
+def content_cluster(entries, threshold=0.3, min_size=2):
+    """Single-linkage agglomerative clustering on Jaccard similarity.
+
+    An entry joins every existing cluster it's similar to, and all such
+    clusters merge into one — proper single-linkage. Without the merge
+    step, ordering matters: entries [A, C, B] where A~B~C but A⊄C would
+    produce two clusters [A,B] + [C] instead of one, so recurrence
+    counts and promotion thresholds become input-order dependent.
+
+    Entries with empty feature sets are dropped (jaccard of two empty
+    sets would otherwise be 1.0). Clusters smaller than min_size are
+    filtered so singletons don't create candidate churn.
+    """
+    featured = [(e, _entry_features(e)) for e in entries]
+    featured = [(e, fs) for e, fs in featured if fs]
+
+    clusters = []  # each: list of (entry, feature_set)
+    for item in featured:
+        e_i, fs_i = item
+        matching_indices = [
+            i for i, c in enumerate(clusters)
+            if any(jaccard(fs_i, fs_j) >= threshold for _, fs_j in c)
+        ]
+        if not matching_indices:
+            clusters.append([item])
+            continue
+        # Merge the new item + every cluster it connects to into one.
+        target = clusters[matching_indices[0]]
+        target.append(item)
+        # Absorb the rest, tail-first so indexing stays valid.
+        for idx in reversed(matching_indices[1:]):
+            target.extend(clusters[idx])
+            del clusters[idx]
+
+    return [[e for e, _ in c] for c in clusters if len(c) >= min_size]
+
+
+def extract_pattern(cluster):
+    """Extractive summarization from a cluster of episodes.
+
+    Without an LLM we cannot synthesize a generalization, so:
+      - claim: canonical (highest-salience) member's reflection or action
+      - conditions: tokens shared by every cluster member
+      - name: longest shared terms + content hash (deterministic, collision-free)
+      - evidence_ids: all member timestamps
+      - cluster_size: recurrence count
+      - canonical_salience: salience of the canonical episode *boosted by*
+        cluster_size. Repetition is a learning signal; a recurring-but-moderate
+        pattern must be able to clear the promotion threshold even when no
+        single episode was extreme. salience_score already caps recurrence at 3.
+    """
+    canonical = max(cluster, key=salience_score)
+    claim = (canonical.get("reflection") or canonical.get("action") or "").strip()
+
+    feature_sets = [_entry_features(e) for e in cluster]
+    common = set.intersection(*feature_sets) if feature_sets else set()
+
+    top_terms = sorted(common, key=lambda t: (-len(t), t))[:3]
+    name_base = "_".join(top_terms) if top_terms else "untitled"
+    # Id derived from normalized claim + conditions (shared tokens). Claim
+    # alone would collide for generic canonical text (e.g., "the test
+    # failed") occurring in unrelated contexts. Conditions usually stay
+    # stable as cluster members shift (intersection of the cluster's common
+    # vocabulary), so lifecycle history carries across membership changes
+    # in the common case while genuinely-different clusters with the same
+    # canonical get distinct ids.
+    conditions_key = "|".join(sorted(common))
+    pattern_id = hashlib.md5(
+        (_normalize_claim(claim) + "||" + conditions_key).encode()
+    ).hexdigest()[:12]
+    name = f"pattern_{name_base}_{pattern_id[:6]}"
+
+    # Recurrence-aware salience: give the scoring function cluster context
+    # without mutating the source episode dict.
+    canonical_with_recurrence = dict(canonical)
+    canonical_with_recurrence["recurrence_count"] = len(cluster)
+    canonical_salience = salience_score(canonical_with_recurrence)
+
+    return {
+        "id": pattern_id,
+        "name": name,
+        "claim": claim,
+        "conditions": sorted(common),
+        "evidence_ids": [e.get("timestamp", "") for e in cluster if e.get("timestamp")],
+        "cluster_size": len(cluster),
+        "canonical_salience": canonical_salience,
+    }
diff --git a/.agent/memory/promote.py b/.agent/memory/promote.py
index eb5a3d4..66efca8 100644
--- a/.agent/memory/promote.py
+++ b/.agent/memory/promote.py
@@ -1,46 +1,166 @@
-"""Detect recurring patterns and promote high-salience entries to semantic memory."""
-import os, datetime
-from collections import defaultdict
-import sys
+"""Cluster + extract + stage candidates. No graduation here — CLI tools do that.
 
-sys.path.insert(0, os.path.join(os.path.dirname(__file__), "..", "harness"))
-from salience import salience_score
+Pipeline:
+  1. cluster_and_extract(entries) — content clusters → structured patterns
+  2. write_candidates(patterns, dir) — patterns → candidate JSON files
 
+Every staged candidate carries lifecycle metadata (status, decisions,
+rejection_count) from birth so repeated churn is visible rather than looking
+fresh each time the pattern recurs.
+"""
+import os, json, datetime, hashlib
+from cluster import content_cluster, extract_pattern
+from review_state import _lessons_sha
+from validate import extract_lesson_lines
 
-def find_recurring_patterns(entries):
-    """Cluster by (skill, action-prefix); entries appearing 2+ times get boosted."""
-    groups = defaultdict(list)
-    for e in entries:
-        key = f"{e.get('skill','general')}::{e.get('action','')[:50]}"
-        groups[key].append(e)
 
-    recurring = {}
-    for key, group in groups.items():
-        if len(group) < 2:
-            continue
-        best = max(group, key=salience_score)
-        best["recurrence_count"] = len(group)
-        best["_salience"] = salience_score(best)
-        recurring[key] = best
-    return recurring
+def cluster_and_extract(entries, threshold=0.3):
+    """Cluster entries by content similarity, extract a pattern per cluster."""
+    clusters = content_cluster(entries, threshold=threshold)
+    return {p["name"]: p for p in (extract_pattern(c) for c in clusters)}
 
 
-def promote_to_semantic(entries, semantic_dir):
-    """Append lessons that aren't already present. Returns count promoted."""
-    if not entries:
-        return 0
-    lessons_path = os.path.join(semantic_dir, "LESSONS.md")
-    existing = open(lessons_path).read() if os.path.exists(lessons_path) else ""
-    new = []
-    for e in entries:
-        line = f"- {e.get('reflection') or e.get('action') or 'unknown'}"
-        if line.strip() and line not in existing:
-            new.append(line)
-    if not new:
+def _slug(pattern_or_key):
+    """Slug for a pattern. Prefer pattern['id'] (claim-derived, stable across
+    cluster membership changes); fall back to md5(key) for legacy callers."""
+    if isinstance(pattern_or_key, dict) and pattern_or_key.get("id"):
+        return pattern_or_key["id"]
+    return hashlib.md5(str(pattern_or_key).encode()).hexdigest()[:12]
+
+
+def _find_prior(slug, candidates_dir):
+    """Look up any prior record for this slug across lifecycle subdirs.
+
+    Returns (prev_dict, location) where location is one of
+    'staged' | 'rejected' | 'graduated' | None. A slug can only live in
+    one place at a time; the caller is responsible for cleaning up the
+    old location when moving the candidate back to staged.
+    """
+    staged_path = os.path.join(candidates_dir, f"{slug}.json")
+    if os.path.isfile(staged_path):
+        try:
+            with open(staged_path) as f:
+                return json.load(f), "staged"
+        except (OSError, json.JSONDecodeError):
+            pass
+    for sub in ("rejected", "graduated"):
+        path = os.path.join(candidates_dir, sub, f"{slug}.json")
+        if os.path.isfile(path):
+            try:
+                with open(path) as f:
+                    return json.load(f), sub
+            except (OSError, json.JSONDecodeError):
+                pass
+    return {}, None
+
+
+def write_candidates(patterns, candidates_dir):
+    """Stage each pattern as a candidate JSON with lifecycle metadata.
+
+    Checks all three lifecycle subdirs (staged / rejected / graduated) for an
+    existing record with the same slug, and preserves its history.
+      - staged already: append a new 'staged' decision, keep original staged_at.
+      - rejected previously: move back to staged with rejection_count and
+        decision log intact. The reviewer sees this as a recurring pattern,
+        not a fresh one.
+      - graduated previously: skip entirely. The lesson already lives in
+        lessons.jsonl; re-staging would only create work the heuristic
+        prefilter would then reject on exact-duplicate grounds.
+    """
+    if not patterns:
         return 0
-    os.makedirs(semantic_dir, exist_ok=True)
-    with open(lessons_path, "a") as f:
-        f.write(f"\n## Auto-promoted {datetime.date.today().isoformat()}\n")
-        for line in new:
-            f.write(line + "\n")
-    return len(new)
+    os.makedirs(candidates_dir, exist_ok=True)
+    written = 0
+    # Read LESSONS.md once — used to check whether specific duplicates that
+    # blocked a prior heuristic rejection are still present.
+    lessons_path = os.path.join(
+        os.path.dirname(candidates_dir), "semantic", "LESSONS.md")
+    lessons_text = ""
+    if os.path.exists(lessons_path):
+        try:
+            lessons_text = open(lessons_path).read()
+        except OSError:
+            pass
+    current_terminal_lessons = set(extract_lesson_lines(lessons_text))
+
+    for key, p in patterns.items():
+        claim = (p.get("claim") or "").strip()
+        if not claim:
+            continue
+        # Prefer the claim+conditions id from extract_pattern — stable slug
+        # means lifecycle state carries across cluster membership changes.
+        slug = _slug(p)
+        prev, prev_loc = _find_prior(slug, candidates_dir)
+
+        # Fully-accepted lesson — terminal, never resurrect.
+        if prev_loc == "graduated" and prev.get("status") != "provisional":
+            continue
+
+        # For rejected + provisional-graduated, re-stage ONLY when something
+        # material has changed since the last decision. Comparing reviewer
+        # identity ("heuristic" vs "human") was a blunt proxy; what actually
+        # matters is whether evidence or the specific blocker shifted.
+        if prev_loc in ("rejected", "graduated"):
+            last = (prev.get("decisions") or [])[-1] if prev.get("decisions") else {}
+            prev_evidence = set(last.get("evidence_snapshot", []))
+            new_evidence = set(p.get("evidence_ids", []))
+            # Only NEW supporting episodes count as a change worth re-review.
+            # Equality comparison would trigger on routine decay (old evidence
+            # archived out of the cluster), even though nothing new arrived
+            # and the original blocker is unchanged.
+            evidence_changed = bool(new_evidence - prev_evidence)
+
+            # Did the specific lesson(s) that triggered this rejection go
+            # away? Uses stamped duplicate_claims rather than a whole-file
+            # LESSONS.md hash — unrelated graduations no longer cause
+            # heuristic-rejected candidates to churn.
+            stamped_dups = last.get("duplicate_claims") or []
+            if stamped_dups:
+                blocker_still_present = any(
+                    d in current_terminal_lessons for d in stamped_dups)
+            else:
+                # No specific stamp (older rejection or human reject).
+                # Provisional and human-rejected cases gate on evidence alone.
+                blocker_still_present = True
+
+            if not evidence_changed and blocker_still_present:
+                continue
+
+        now = datetime.datetime.now().isoformat()
+        decisions = prev.get("decisions", [])
+        decisions.append({"ts": now, "action": "staged", "reviewer": "auto_dream"})
+
+        # Preserve original staged_at so priority + backlog age signals stay
+        # meaningful across re-detections.
+        staged_at = prev.get("staged_at") or now
+
+        candidate = {
+            "id": slug,
+            "key": key,
+            "name": p.get("name", key),
+            "claim": claim,
+            "conditions": p.get("conditions", []),
+            "evidence_ids": p.get("evidence_ids", []),
+            "cluster_size": p.get("cluster_size", 1),
+            "canonical_salience": p.get("canonical_salience", 0.0),
+            "staged_at": staged_at,
+            "status": "staged",
+            "decisions": decisions,
+            "rejection_count": prev.get("rejection_count", 0),
+        }
+
+        staged_path = os.path.join(candidates_dir, f"{slug}.json")
+        with open(staged_path, "w") as f:
+            json.dump(candidate, f, indent=2)
+
+        # The slug must live in exactly one lifecycle location. Remove any
+        # prior copy in rejected/ or graduated/ (the latter only for
+        # provisional re-review — accepted never gets here because it's
+        # skipped above).
+        if prev_loc in ("rejected", "graduated"):
+            try:
+                os.remove(os.path.join(candidates_dir, prev_loc, f"{slug}.json"))
+            except OSError:
+                pass
+        written += 1
+    return written
diff --git a/.agent/memory/render_lessons.py b/.agent/memory/render_lessons.py
new file mode 100644
index 0000000..97cafb0
--- /dev/null
+++ b/.agent/memory/render_lessons.py
@@ -0,0 +1,223 @@
+"""Render semantic/LESSONS.md from structured semantic/lessons.jsonl.
+
+lessons.jsonl is the source of truth. LESSONS.md is a derived view. Graduate.py
+and reject.py write to lessons.jsonl and call render_lessons to regenerate
+the markdown.
+
+Preserves user content above the sentinel. On first call, the existing
+`## Auto-promoted entries will be appended below` line (shipped in the
+template) is treated as the sentinel; subsequent calls replace everything
+from the sentinel onward. If the sentinel is missing it's appended at the
+end of the file. This means hand-curated preambles and seed bullets above
+the sentinel survive every render.
+"""
+import os, json, datetime, hashlib
+from collections import defaultdict
+
+
+LESSONS_JSONL = "lessons.jsonl"
+LESSONS_MD = "LESSONS.md"
+
+SENTINEL = "## Auto-promoted entries will be appended below"
+
+
+def append_lesson(lesson, semantic_dir):
+    """Append a lesson to semantic/lessons.jsonl. Returns the written path."""
+    os.makedirs(semantic_dir, exist_ok=True)
+    path = os.path.join(semantic_dir, LESSONS_JSONL)
+    with open(path, "a") as f:
+        f.write(json.dumps(lesson) + "\n")
+    return path
+
+
+def load_lessons(semantic_dir):
+    path = os.path.join(semantic_dir, LESSONS_JSONL)
+    if not os.path.exists(path):
+        return []
+    out = []
+    for line in open(path):
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            out.append(json.loads(line))
+        except json.JSONDecodeError:
+            continue
+    return out
+
+
+def _bullet_for(lesson, superseded_by):
+    claim = lesson.get("claim", "")
+    conf = lesson.get("confidence", "?")
+    status = lesson.get("status", "accepted")
+    ev = lesson.get("evidence_ids", [])
+    lid = lesson.get("id", "?")
+    ann = f"status={status} confidence={conf} evidence={len(ev)} id={lid}"
+    sup_by = superseded_by.get(lid)
+    if sup_by:
+        return f"- ~~{claim}~~  <!-- {ann} superseded_by={sup_by} -->"
+    if status == "provisional":
+        return f"- [PROVISIONAL] {claim}  <!-- {ann} -->"
+    return f"- {claim}  <!-- {ann} -->"
+
+
+def _build_auto_section(lessons):
+    # Only accepted supersessions flip the old lesson to strikethrough.
+    # A provisional --supersedes would otherwise blank the active lesson
+    # before its replacement has been accepted, leaving no active guidance
+    # on that topic at all (retrieval skips both provisional and
+    # strikethrough).
+    superseded_by = {}
+    for L in lessons:
+        if L.get("status") != "accepted":
+            continue
+        sup = L.get("supersedes")
+        if sup:
+            superseded_by[sup] = L.get("id")
+
+    groups = defaultdict(list)
+    for L in lessons:
+        month = (L.get("accepted_at") or "")[:7] or "unknown"
+        groups[month].append(L)
+
+    lines = []
+    for month in sorted(groups.keys(), reverse=True):
+        lines.append(f"### {month}")
+        lines.append("")
+        for L in groups[month]:
+            lines.append(_bullet_for(L, superseded_by))
+        lines.append("")
+    return "\n".join(lines).rstrip() + "\n" if lines else ""
+
+
+def migrate_legacy_bullets(semantic_dir):
+    """Import any bullets below the sentinel not yet in lessons.jsonl.
+
+    Upgrade safety: installations that ran the old markdown-only promotion
+    have auto-promoted bullets below the sentinel. Without this pass, the
+    first call to render_lessons with an empty lessons.jsonl would rewrite
+    LESSONS.md with an empty auto-section and lose all of them silently.
+    Migrated entries land with status='legacy' so they're visually distinct
+    and can be reviewed + superseded by the host agent later.
+    """
+    md_path = os.path.join(semantic_dir, LESSONS_MD)
+    if not os.path.exists(md_path):
+        return 0
+    content = open(md_path).read()
+    if SENTINEL not in content:
+        return 0
+
+    below = content.split(SENTINEL, 1)[1]
+    bullets = []
+    for line in below.splitlines():
+        s = line.strip()
+        if not s.startswith("- ") or len(s) <= 2:
+            continue
+        text = s[2:].split("<!--")[0].strip()
+        # Skip superseded entries — they're historical, not content to re-ingest
+        if text.startswith("~~") and text.endswith("~~"):
+            continue
+        # Strip provisional prefix if present
+        if text.startswith("[PROVISIONAL]"):
+            text = text[len("[PROVISIONAL]"):].strip()
+        if text:
+            bullets.append(text)
+
+    if not bullets:
+        return 0
+
+    existing_claims = {(L.get("claim") or "").strip().lower()
+                       for L in load_lessons(semantic_dir)}
+    try:
+        accepted_at = datetime.datetime.fromtimestamp(
+            os.path.getmtime(md_path)).isoformat()
+    except OSError:
+        accepted_at = datetime.datetime.now().isoformat()
+
+    migrated = 0
+    for claim in bullets:
+        if claim.strip().lower() in existing_claims:
+            continue
+        lid = "lesson_legacy_" + hashlib.md5(claim.lower().encode()).hexdigest()[:12]
+        lesson = {
+            "id": lid, "claim": claim,
+            "conditions": [], "evidence_ids": [],
+            "status": "legacy", "accepted_at": accepted_at,
+            "reviewer": "render_lessons_migration",
+            "rationale": "Imported from pre-restructure LESSONS.md bullets below sentinel",
+            "cluster_size": 1, "canonical_salience": 5.0,
+            "confidence": 0.7, "support_count": 0, "contradiction_count": 0,
+            "supersedes": None, "source_candidate": None,
+        }
+        append_lesson(lesson, semantic_dir)
+        existing_claims.add(claim.strip().lower())
+        migrated += 1
+    return migrated
+
+
+def _dedupe_by_id(lessons):
+    """Keep the latest entry per lesson id.
+
+    lessons.jsonl is append-only, so a provisional→accepted state transition
+    writes two rows with the same id. The render should show only the latest
+    state for each lesson; the jsonl is preserved for audit.
+    """
+    latest = {}
+    order = []
+    for L in lessons:
+        lid = L.get("id")
+        if not lid:
+            # No id? Treat as-is, keyed by its position so we keep it.
+            lid = f"_anon_{len(order)}"
+        if lid not in latest:
+            order.append(lid)
+        latest[lid] = L
+    return [latest[lid] for lid in order]
+
+
+def render_lessons(semantic_dir):
+    """Re-render LESSONS.md. Preserves hand-curated content above the sentinel.
+
+    Auto-migrates legacy auto-promoted bullets below the sentinel into
+    lessons.jsonl before rendering, so upgrades from the old markdown-only
+    format don't silently erase past promotions. Deduplicates entries by
+    lesson id so a provisional-then-accepted lesson renders once, not twice.
+    """
+    # First, absorb any un-registered bullets below the sentinel
+    migrate_legacy_bullets(semantic_dir)
+    lessons = _dedupe_by_id(load_lessons(semantic_dir))
+    auto_section = _build_auto_section(lessons)
+
+    path = os.path.join(semantic_dir, LESSONS_MD)
+
+    if os.path.exists(path):
+        existing = open(path).read()
+        if SENTINEL in existing:
+            prefix = existing.split(SENTINEL)[0].rstrip()
+            new = f"{prefix}\n\n{SENTINEL}\n\n{auto_section}"
+        else:
+            new = existing.rstrip() + f"\n\n{SENTINEL}\n\n{auto_section}"
+    else:
+        header = (
+            "# Lessons\n\n"
+            "> _Auto-managed below. Hand-curated preamble + seed lessons "
+            "above the sentinel are preserved across renders._\n"
+        )
+        new = f"{header}\n{SENTINEL}\n\n{auto_section}"
+
+    os.makedirs(semantic_dir, exist_ok=True)
+    with open(path, "w") as f:
+        f.write(new)
+    return path
+
+
+def render_lessons_as_text(semantic_dir):
+    return open(render_lessons(semantic_dir)).read()
+
+
+if __name__ == "__main__":
+    import sys
+    sem = sys.argv[1] if len(sys.argv) > 1 else os.path.join(
+        os.path.dirname(os.path.abspath(__file__)), "semantic")
+    path = render_lessons(sem)
+    print(f"rendered: {path}")
diff --git a/.agent/memory/review_state.py b/.agent/memory/review_state.py
new file mode 100644
index 0000000..7eee543
--- /dev/null
+++ b/.agent/memory/review_state.py
@@ -0,0 +1,265 @@
+"""Candidate lifecycle + decision log.
+
+Each candidate JSON under memory/candidates/ carries:
+  status:    staged | provisional | accepted | rejected | superseded
+  decisions: append-only list of {ts, action, reviewer, notes, **fields}
+
+Host-agent CLI tools (.agent/tools/graduate.py, reject.py, reopen.py) call
+into this module to transition state. Rejection and re-stage preserve full
+history so a candidate that keeps reappearing is visibly churning rather
+than looking novel each time.
+"""
+import os, json, datetime, hashlib
+
+
+def _now():
+    return datetime.datetime.now().isoformat()
+
+
+def _touch(candidate, action, reviewer, notes="", **fields):
+    decisions = candidate.setdefault("decisions", [])
+    decisions.append({
+        "ts": _now(),
+        "action": action,
+        "reviewer": reviewer,
+        "notes": notes,
+        **fields,
+    })
+
+
+def _lessons_sha(candidates_dir):
+    """Short hash of current LESSONS.md, used to stamp decisions.
+
+    Re-staging logic uses this to tell 'semantic state changed since this
+    decision' apart from 'nothing has changed, skip to avoid churn'.
+    """
+    lessons_path = os.path.join(
+        os.path.dirname(candidates_dir), "semantic", "LESSONS.md")
+    if not os.path.exists(lessons_path):
+        return ""
+    try:
+        with open(lessons_path, "rb") as f:
+            return hashlib.md5(f.read()).hexdigest()[:12]
+    except OSError:
+        return ""
+
+
+def _stamp_evidence_and_lessons(cand, candidates_dir):
+    """Attach evidence + lessons snapshots to the most recent decision.
+
+    Called on every terminal-ish transition (rejected, graduated). Lets
+    write_candidates decide whether a later re-detection represents genuinely
+    new information or the same state we already judged.
+    """
+    if not cand.get("decisions"):
+        return
+    last = cand["decisions"][-1]
+    last["evidence_snapshot"] = list(cand.get("evidence_ids", []))
+    last["lessons_sha"] = _lessons_sha(candidates_dir)
+
+
+def load_candidate(path):
+    with open(path) as f:
+        return json.load(f)
+
+
+def save_candidate(candidate, path):
+    with open(path, "w") as f:
+        json.dump(candidate, f, indent=2)
+
+
+def stage_candidate(candidate_path, reviewer="auto_dream"):
+    """Mark a freshly-written candidate as staged with an initial decision entry."""
+    cand = load_candidate(candidate_path)
+    cand.setdefault("status", "staged")
+    _touch(cand, "staged", reviewer)
+    save_candidate(cand, candidate_path)
+
+
+def _default_queue_path(candidates_dir):
+    """By convention, memory/candidates/ sits next to memory/working/REVIEW_QUEUE.md."""
+    memory_dir = os.path.dirname(candidates_dir)
+    return os.path.join(memory_dir, "working", "REVIEW_QUEUE.md")
+
+
+def _refresh_queue(candidates_dir):
+    """Keep REVIEW_QUEUE.md in sync after any lifecycle transition.
+
+    build_context loads this file into every host session, so a stale file
+    makes reviewed items keep appearing as pending and reopened items stay
+    invisible until the next dream cycle.
+    """
+    try:
+        write_review_queue_summary(candidates_dir, _default_queue_path(candidates_dir))
+    except Exception:
+        # Never let queue bookkeeping break a graduation / rejection action.
+        pass
+
+
+def mark_graduated(candidate_id, reviewer, rationale, candidates_dir,
+                   provisional=False):
+    """Move a staged candidate to candidates/graduated/ with an accept decision.
+
+    Returns the graduated candidate dict. Caller is responsible for writing
+    the structured lesson entry to semantic/lessons.jsonl and re-rendering
+    LESSONS.md — this function only handles the candidate side.
+    """
+    src = os.path.join(candidates_dir, f"{candidate_id}.json")
+    if not os.path.exists(src):
+        raise FileNotFoundError(f"candidate not found: {candidate_id}")
+    cand = load_candidate(src)
+    cand["status"] = "provisional" if provisional else "accepted"
+    cand["accepted_at"] = _now()
+    cand["reviewer"] = reviewer
+    cand["rationale"] = rationale
+    _touch(cand, "graduated", reviewer, notes=rationale,
+           provisional=provisional)
+    _stamp_evidence_and_lessons(cand, candidates_dir)
+
+    graduated_dir = os.path.join(candidates_dir, "graduated")
+    os.makedirs(graduated_dir, exist_ok=True)
+    dst = os.path.join(graduated_dir, f"{candidate_id}.json")
+    save_candidate(cand, dst)
+    os.remove(src)
+    _refresh_queue(candidates_dir)
+    return cand
+
+
+def mark_rejected(candidate_id, reviewer, reason, candidates_dir, **extra_stamp):
+    """Move a staged candidate to candidates/rejected/ with a reject decision.
+
+    rejection_count tracks how many times this id has been rejected — if it
+    keeps coming back, the reviewer sees churn instead of a 'fresh' item.
+
+    extra_stamp kwargs are merged into the decision entry. heuristic_prefilter
+    uses this to record which specific lessons triggered the duplicate rejection
+    (duplicate_claims=[...]); write_candidates later checks whether those
+    specific lessons are still present before re-staging, so unrelated LESSONS
+    edits don't cause the candidate to churn.
+    """
+    src = os.path.join(candidates_dir, f"{candidate_id}.json")
+    if not os.path.exists(src):
+        raise FileNotFoundError(f"candidate not found: {candidate_id}")
+    cand = load_candidate(src)
+    cand["status"] = "rejected"
+    cand["rejection_count"] = cand.get("rejection_count", 0) + 1
+    _touch(cand, "rejected", reviewer, notes=reason, **extra_stamp)
+    _stamp_evidence_and_lessons(cand, candidates_dir)
+
+    rejected_dir = os.path.join(candidates_dir, "rejected")
+    os.makedirs(rejected_dir, exist_ok=True)
+    dst = os.path.join(rejected_dir, f"{candidate_id}.json")
+    save_candidate(cand, dst)
+    os.remove(src)
+    _refresh_queue(candidates_dir)
+    return cand
+
+
+def mark_reopened(candidate_id, reviewer, candidates_dir):
+    """Move a rejected candidate back to the staged pool with history intact."""
+    src = os.path.join(candidates_dir, "rejected", f"{candidate_id}.json")
+    if not os.path.exists(src):
+        raise FileNotFoundError(f"rejected candidate not found: {candidate_id}")
+    cand = load_candidate(src)
+    cand["status"] = "staged"
+    _touch(cand, "reopened", reviewer)
+
+    dst = os.path.join(candidates_dir, f"{candidate_id}.json")
+    save_candidate(cand, dst)
+    os.remove(src)
+    _refresh_queue(candidates_dir)
+    return cand
+
+
+def _age_factor(staged_at):
+    """1.0 at stage time, grows to 2.0 for candidates ~14 days old."""
+    try:
+        staged = datetime.datetime.fromisoformat(staged_at)
+    except (ValueError, TypeError):
+        return 1.0
+    age_days = (datetime.datetime.now() - staged).days
+    return 1.0 + min(1.0, age_days / 14.0)
+
+
+def candidate_priority(candidate):
+    """priority = cluster_size * canonical_salience * age_factor.
+
+    Reviewers attack high-priority items first. Older + more-recurrent +
+    higher-salience patterns deserve attention ahead of one-offs.
+    """
+    return (
+        max(1, candidate.get("cluster_size", 1)) *
+        max(0.1, candidate.get("canonical_salience", 0.1)) *
+        _age_factor(candidate.get("staged_at", ""))
+    )
+
+
+def list_candidates(candidates_dir, status="staged", sort_by="priority"):
+    """Return candidate dicts with the given status, sorted by the key."""
+    if status == "staged":
+        search_dir = candidates_dir
+    else:
+        search_dir = os.path.join(candidates_dir, status)
+    if not os.path.isdir(search_dir):
+        return []
+
+    out = []
+    for fname in os.listdir(search_dir):
+        if not fname.endswith(".json"):
+            continue
+        path = os.path.join(search_dir, fname)
+        if not os.path.isfile(path):
+            continue
+        try:
+            with open(path) as f:
+                out.append(json.load(f))
+        except (OSError, json.JSONDecodeError):
+            continue
+
+    if sort_by == "priority":
+        out.sort(key=candidate_priority, reverse=True)
+    elif sort_by == "age":
+        out.sort(key=lambda c: c.get("staged_at", ""))
+    return out
+
+
+def write_review_queue_summary(candidates_dir, summary_path):
+    """Emit a compact REVIEW_QUEUE.md so the host agent sees the backlog.
+
+    On-demand review without a surfacing mechanism grows silent backlog.
+    This file sits in memory/working/ and gets loaded by context_budget into
+    every host session — impossible to miss.
+    """
+    pending = list_candidates(candidates_dir, status="staged")
+    os.makedirs(os.path.dirname(summary_path), exist_ok=True)
+    if not pending:
+        with open(summary_path, "w") as f:
+            f.write("# Review Queue\n\n_No pending candidates._\n")
+        return 0
+
+    staged_ats = [c.get("staged_at", "") for c in pending if c.get("staged_at")]
+    oldest = min(staged_ats) if staged_ats else ""
+    lines = ["# Review Queue", ""]
+    lines.append(f"**Pending:** {len(pending)}")
+    if oldest:
+        lines.append(f"**Oldest staged:** {oldest}")
+    lines.append("")
+    lines.append("Run `python .agent/tools/list_candidates.py` for detail, then:")
+    lines.append("- `python .agent/tools/graduate.py <id> --rationale \"...\"` to accept")
+    lines.append("- `python .agent/tools/reject.py <id> --reason \"...\"` to reject")
+    lines.append("- Review in a batch so cross-candidate contradictions are caught.")
+    lines.append("")
+    lines.append("## Priority order (top 10)")
+    lines.append("")
+    for cand in pending[:10]:
+        prio = candidate_priority(cand)
+        claim_preview = (cand.get("claim") or "")[:80]
+        lines.append(
+            f"- **{cand.get('id')}** (priority={prio:.2f}, "
+            f"size={cand.get('cluster_size', '?')}, "
+            f"rejections={cand.get('rejection_count', 0)}) "
+            f"— {claim_preview}"
+        )
+    with open(summary_path, "w") as f:
+        f.write("\n".join(lines) + "\n")
+    return len(pending)
diff --git a/.agent/memory/validate.py b/.agent/memory/validate.py
new file mode 100644
index 0000000..338df24
--- /dev/null
+++ b/.agent/memory/validate.py
@@ -0,0 +1,104 @@
+"""Heuristic pre-filter for candidate lessons. Deterministic, no LLM.
+
+The host agent (Claude Code, Codex, Windsurf) does actual reasoning via the
+CLI tools in .agent/tools/ (graduate.py, reject.py). This module catches
+obvious junk — too-short claims, exact duplicates — before the reviewer
+sees the candidate at all. Anything subjective is the host's job.
+"""
+import re
+
+MIN_CLAIM_LEN = 20
+LENGTH_SATURATE = 100
+CLUSTER_SATURATE = 5
+
+_STATUS_RE = re.compile(r"status=(\w+)")
+
+
+def _normalize(text):
+    """Lowercase, strip punctuation, collapse whitespace. For exact-dup detection."""
+    t = re.sub(r"[^\w\s]", " ", (text or "").lower())
+    return re.sub(r"\s+", " ", t).strip()
+
+
+def extract_lesson_lines(lessons_md):
+    """Extract accepted lesson claims from rendered markdown.
+
+    Only TERMINAL lessons count for duplicate detection. Non-terminal status
+    values — `provisional` (probationary), `legacy` (imported from pre-restructure
+    LESSONS.md), and anything superseded — must be skipped so recurrences can
+    reach the host agent for re-review, evidence accumulation, or supersession.
+    Status is read from the HTML annotation written by render_lessons; visual
+    markers (`[PROVISIONAL]`, `~~...~~`) are the fallback for unannotated lines.
+    """
+    out = []
+    for line in (lessons_md or "").splitlines():
+        s = line.strip()
+        if not s.startswith("- ") or len(s) <= 2:
+            continue
+        # Primary signal: status in HTML annotation
+        if "<!--" in s:
+            ann = s.split("<!--", 1)[1]
+            m = _STATUS_RE.search(ann)
+            if m and m.group(1) != "accepted":
+                continue
+        text = s[2:].split("<!--")[0].strip()
+        # Fallback: visual markers (for pre-existing bullets without annotations)
+        if text.startswith("[PROVISIONAL]"):
+            continue
+        if text.startswith("~~") and text.endswith("~~"):
+            continue
+        if text:
+            out.append(text)
+    return out
+
+
+def check_exact_duplicate(claim, existing_lessons_md):
+    """Return lesson lines whose normalized form matches the candidate."""
+    nc = _normalize(claim)
+    if not nc:
+        return []
+    return [l for l in extract_lesson_lines(existing_lessons_md)
+            if _normalize(l) == nc]
+
+
+def heuristic_check(candidate, existing_lessons_md=""):
+    """Deterministic pre-filter. Takes a candidate dict, not a claim string.
+
+    passed=False means obvious junk: too short, or exact duplicate of a
+    lesson already in LESSONS.md. Everything else should reach the host
+    agent for real review — overlap != contradiction.
+
+    Returns {passed, confidence, reasons, duplicates}.
+      confidence — structural quality hint for reviewer priority only, not a gate.
+    """
+    claim = (candidate.get("claim") or "").strip()
+    reasons, duplicates = [], []
+
+    if len(claim) < MIN_CLAIM_LEN:
+        reasons.append("claim_too_short")
+
+    if claim:
+        duplicates = check_exact_duplicate(claim, existing_lessons_md)
+        if duplicates:
+            reasons.append(f"exact_duplicate_of_{len(duplicates)}_lessons")
+
+    cluster_size = candidate.get("cluster_size", 1)
+    length_score = min(1.0, len(claim) / LENGTH_SATURATE)
+    size_score = min(1.0, cluster_size / CLUSTER_SATURATE)
+    confidence = round(0.5 * length_score + 0.5 * size_score, 3)
+
+    return {
+        "passed": not reasons,
+        "confidence": confidence,
+        "reasons": reasons,
+        "duplicates": duplicates,
+    }
+
+
+def validate_candidate(claim_or_candidate, existing_lessons_md="", bootstrap=False):
+    """Backwards-compat shim. Old callers passed a claim string; new callers
+    pass a candidate dict. bootstrap is accepted but ignored — heuristic
+    check has no threshold to loosen."""
+    if isinstance(claim_or_candidate, str):
+        return heuristic_check({"claim": claim_or_candidate}, existing_lessons_md)
+    return heuristic_check(claim_or_candidate, existing_lessons_md)
diff --git a/.agent/tools/graduate.py b/.agent/tools/graduate.py
new file mode 100644
index 0000000..628d02f
--- /dev/null
+++ b/.agent/tools/graduate.py
@@ -0,0 +1,112 @@
+"""Graduate a staged candidate to semantic memory.
+
+The host agent reviews a candidate, decides it's worth keeping, and calls
+this tool with a rationale. A heuristic re-check (length + exact duplicate
+against current LESSONS.md) runs automatically so last-minute issues get
+caught. The rationale is REQUIRED — rubber-stamped promotions are the
+whole failure mode this layer is designed to prevent.
+"""
+import os, sys, json, argparse, hashlib, datetime
+
+BASE = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+sys.path.insert(0, os.path.join(BASE, "memory"))
+
+from review_state import mark_graduated
+from validate import heuristic_check
+from render_lessons import append_lesson, render_lessons
+
+CANDIDATES = os.path.join(BASE, "memory/candidates")
+SEMANTIC = os.path.join(BASE, "memory/semantic")
+
+
+def _lesson_id(candidate):
+    """1:1 with the candidate's own id (claim + conditions, stable).
+
+    Using md5(claim) alone collides when two distinct patterns happen to
+    share canonical text — different clusters about different situations
+    with the same "the test failed" reflection, etc. Keying off the
+    candidate id keeps them distinct and matches how write_candidates
+    identifies the pattern lifecycle.
+    """
+    cid = candidate.get("id") or ""
+    if cid:
+        return f"lesson_{cid}"
+    # Fallback for older candidate dicts without `id`
+    claim = (candidate.get("claim") or "").strip().lower()
+    return "lesson_" + hashlib.md5(claim.encode()).hexdigest()[:12]
+
+
+def main():
+    p = argparse.ArgumentParser(description="Graduate a staged candidate.")
+    p.add_argument("candidate_id")
+    p.add_argument("--rationale", required=True,
+                   help="Why this lesson should be accepted. Required, not optional.")
+    p.add_argument("--reviewer", default="host-agent")
+    p.add_argument("--provisional", action="store_true",
+                   help="Accept as provisional (probationary) rather than full.")
+    p.add_argument("--supersedes", default=None,
+                   help="ID of an existing lesson this replaces.")
+    args = p.parse_args()
+
+    cand_path = os.path.join(CANDIDATES, f"{args.candidate_id}.json")
+    if not os.path.exists(cand_path):
+        print(f"ERROR: candidate not found: {args.candidate_id}", file=sys.stderr)
+        sys.exit(1)
+    with open(cand_path) as f:
+        cand = json.load(f)
+
+    lessons_md = os.path.join(SEMANTIC, "LESSONS.md")
+    existing = open(lessons_md).read() if os.path.exists(lessons_md) else ""
+    # When superseding, exclude the target lesson from the duplicate check —
+    # replacing a lesson with structurally-better content but same wording
+    # is exactly what supersession is for.
+    if args.supersedes:
+        existing = "\n".join(
+            line for line in existing.splitlines()
+            if f"id={args.supersedes}" not in line
+        )
+    check = heuristic_check(cand, existing)
+    if not check["passed"]:
+        print(f"ERROR: candidate fails heuristic check: {check['reasons']}",
+              file=sys.stderr)
+        sys.exit(2)
+
+    # Atomicity: write to semantic memory BEFORE moving the candidate.
+    # If we crash mid-graduation, the staged candidate remains and the
+    # reviewer can retry. Append-only lessons.jsonl is deduped by id at
+    # render time, so a retry that re-appends produces one lesson bullet,
+    # not two. Crashing AFTER the move would leave the candidate lost and
+    # the lesson unlogged — the failure mode this reorder prevents.
+    accepted_at = datetime.datetime.now().isoformat()
+    lesson = {
+        "id": _lesson_id(cand),
+        "claim": cand.get("claim"),
+        "conditions": cand.get("conditions", []),
+        "evidence_ids": cand.get("evidence_ids", []),
+        "status": "provisional" if args.provisional else "accepted",
+        "accepted_at": accepted_at,
+        "reviewer": args.reviewer,
+        "rationale": args.rationale,
+        "cluster_size": cand.get("cluster_size", 1),
+        "canonical_salience": cand.get("canonical_salience", 0.0),
+        "confidence": check["confidence"],
+        "support_count": 0,
+        "contradiction_count": 0,
+        "supersedes": args.supersedes,
+        "source_candidate": args.candidate_id,
+    }
+    append_lesson(lesson, SEMANTIC)
+    md_path = render_lessons(SEMANTIC)
+
+    # Semantic writes survived — now move the candidate file.
+    mark_graduated(
+        args.candidate_id, args.reviewer, args.rationale, CANDIDATES,
+        provisional=args.provisional,
+    )
+
+    print(f"graduated {args.candidate_id} → lesson {lesson['id']}")
+    print(f"re-rendered: {md_path}")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/.agent/tools/list_candidates.py b/.agent/tools/list_candidates.py
new file mode 100644
index 0000000..82c67fd
--- /dev/null
+++ b/.agent/tools/list_candidates.py
@@ -0,0 +1,62 @@
+"""List candidate lessons by status + priority.
+
+Host-agent workflow: run this, pick the top N, review each with graduate.py
+or reject.py. Priority = cluster_size * canonical_salience * age_factor, so
+recurring + salient + aging items get attention first.
+"""
+import os, sys, json, argparse
+
+BASE = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+sys.path.insert(0, os.path.join(BASE, "memory"))
+
+from review_state import list_candidates, candidate_priority
+
+CANDIDATES = os.path.join(BASE, "memory/candidates")
+
+
+def main():
+    p = argparse.ArgumentParser(description="List candidate lessons.")
+    p.add_argument("--status", default="staged",
+                   choices=["staged", "rejected", "graduated"])
+    p.add_argument("--sort", default="priority",
+                   choices=["priority", "age"])
+    p.add_argument("--limit", type=int, default=0,
+                   help="Max to return; 0 = no limit.")
+    p.add_argument("--format", default="human", choices=["human", "json"])
+    args = p.parse_args()
+
+    items = list_candidates(CANDIDATES, status=args.status, sort_by=args.sort)
+    if args.limit:
+        items = items[:args.limit]
+
+    if args.format == "json":
+        print(json.dumps(
+            [{**c, "priority": round(candidate_priority(c), 3)} for c in items],
+            indent=2))
+        return
+
+    if not items:
+        print(f"No candidates with status={args.status}.")
+        return
+
+    print(f"=== {len(items)} candidate(s) — status={args.status} ===\n")
+    for c in items:
+        prio = candidate_priority(c)
+        print(f"# {c.get('id')}  (priority={prio:.2f})")
+        print(f"  claim:      {c.get('claim', '')}")
+        print(f"  cluster:    {c.get('cluster_size', '?')} episode(s)")
+        print(f"  salience:   {c.get('canonical_salience', 0):.2f}")
+        print(f"  conditions: {c.get('conditions', [])}")
+        print(f"  evidence:   {len(c.get('evidence_ids', []))} episode(s)")
+        print(f"  rejections: {c.get('rejection_count', 0)}")
+        print(f"  staged:     {c.get('staged_at', '?')}")
+        dec = c.get("decisions", [])
+        if dec:
+            last = dec[-1]
+            print(f"  last:       {last.get('action')} by "
+                  f"{last.get('reviewer')} @ {last.get('ts', '')[:19]}")
+        print()
+
+
+if __name__ == "__main__":
+    main()
diff --git a/.agent/tools/memory_reflect.py b/.agent/tools/memory_reflect.py
index bb88e98..7101bb0 100644
--- a/.agent/tools/memory_reflect.py
+++ b/.agent/tools/memory_reflect.py
@@ -7,12 +7,16 @@
 
 
 def reflect(skill_name, action, outcome, success=True, importance=5,
-            reflection="", error=None):
+            reflection="", error=None, confidence=None, evidence_ids=None):
     if success:
         return log_execution(skill_name, action, outcome, True,
-                             reflection=reflection, importance=importance)
+                             reflection=reflection, importance=importance,
+                             confidence=0.5 if confidence is None else confidence,
+                             evidence_ids=evidence_ids)
     return on_failure(skill_name, action, error or outcome,
-                      context=reflection)
+                      context=reflection,
+                      confidence=0.9 if confidence is None else confidence,
+                      evidence_ids=evidence_ids)
 
 
 if __name__ == "__main__":
@@ -24,7 +28,11 @@ def reflect(skill_name, action, outcome, success=True, importance=5,
     p.add_argument("--fail", action="store_true")
     p.add_argument("--importance", type=int, default=5)
     p.add_argument("--note", default="")
+    p.add_argument("--confidence", type=float, default=None)
+    p.add_argument("--evidence", nargs="*", default=None,
+                   help="Space-separated episode/lesson IDs this entry builds on.")
     args = p.parse_args()
     print(reflect(args.skill, args.action, args.outcome,
                   success=not args.fail, importance=args.importance,
-                  reflection=args.note))
+                  reflection=args.note, confidence=args.confidence,
+                  evidence_ids=args.evidence))
diff --git a/.agent/tools/reject.py b/.agent/tools/reject.py
new file mode 100644
index 0000000..2567807
--- /dev/null
+++ b/.agent/tools/reject.py
@@ -0,0 +1,38 @@
+"""Reject a staged candidate. Moves to rejected/ with a decision log entry.
+
+The candidate is not deleted. Its decision history, including the reject
+reason, is preserved. If the same pattern recurs and gets re-staged by
+auto_dream, rejection_count will show the history — so the reviewer sees
+churn instead of treating it as a fresh item.
+"""
+import os, sys, argparse
+
+BASE = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+sys.path.insert(0, os.path.join(BASE, "memory"))
+
+from review_state import mark_rejected
+
+CANDIDATES = os.path.join(BASE, "memory/candidates")
+
+
+def main():
+    p = argparse.ArgumentParser(description="Reject a staged candidate.")
+    p.add_argument("candidate_id")
+    p.add_argument("--reason", required=True,
+                   help="Why it's being rejected. Required.")
+    p.add_argument("--reviewer", default="host-agent")
+    args = p.parse_args()
+
+    try:
+        rejected = mark_rejected(args.candidate_id, args.reviewer,
+                                 args.reason, CANDIDATES)
+    except FileNotFoundError as e:
+        print(f"ERROR: {e}", file=sys.stderr)
+        sys.exit(1)
+
+    print(f"rejected {args.candidate_id} "
+          f"(rejection_count={rejected.get('rejection_count', 1)})")
+
+
+if __name__ == "__main__":
+    main()
diff --git a/.agent/tools/reopen.py b/.agent/tools/reopen.py
new file mode 100644
index 0000000..3375dd7
--- /dev/null
+++ b/.agent/tools/reopen.py
@@ -0,0 +1,34 @@
+"""Move a rejected candidate back to the staged pool.
+
+Use this when a previous rejection was wrong, or when new evidence changes
+the picture. Decision history and rejection_count survive the reopen — the
+next reviewer sees the full churn log.
+"""
+import os, sys, argparse
+
+BASE = os.path.abspath(os.path.join(os.path.dirname(__file__), ".."))
+sys.path.insert(0, os.path.join(BASE, "memory"))
+
+from review_state import mark_reopened
+
+CANDIDATES = os.path.join(BASE, "memory/candidates")
+
+
+def main():
+    p = argparse.ArgumentParser(description="Reopen a rejected candidate.")
+    p.add_argument("candidate_id")
+    p.add_argument("--reviewer", default="host-agent")
+    args = p.parse_args()
+
+    try:
+        cand = mark_reopened(args.candidate_id, args.reviewer, CANDIDATES)
+    except FileNotFoundError as e:
+        print(f"ERROR: {e}", file=sys.stderr)
+        sys.exit(1)
+
+    print(f"reopened {args.candidate_id} "
+          f"(rejection_count={cand.get('rejection_count', 0)})")
+
+
+if __name__ == "__main__":
+    main()