diff --git a/.gitignore b/.gitignore
index 74e140de..3afd6529 100644
--- a/.gitignore
+++ b/.gitignore
@@ -11,3 +11,4 @@ node_modules/
 tmp/
 .playwright-mcp/
 .claude/
+.codex/
diff --git a/AGENTS.md b/AGENTS.md
new file mode 120000
index 00000000..681311eb
--- /dev/null
+++ b/AGENTS.md
@@ -0,0 +1 @@
+CLAUDE.md
\ No newline at end of file
diff --git a/Makefile b/Makefile
index 7fd5df0f..dcbb7047 100644
--- a/Makefile
+++ b/Makefile
@@ -155,6 +155,14 @@ retrieval-report:
 migrate-sidecars-dry:
 	uv run python scripts/migrate_sidecars_to_dirs.py --dry-run
 
+# Report git worktrees/branches whose work already landed (no changes made)
+prune-worktrees-dry:
+	uv run python scripts/prune_agent_worktrees.py
+
+# Remove landed worktrees and delete their branches
+prune-worktrees:
+	uv run python scripts/prune_agent_worktrees.py --apply
+
 # Build web UI vendor bundle (npm + esbuild)
 # Run after changing JS dependencies in src/decafclaw/web/static/package.json
 # Requires Node.js. Output is committed to git, so only needed for dev.
diff --git a/docs/dev-sessions/2026-08-08-0000-783-synchronous-hooks/spec.md b/docs/dev-sessions/2026-08-08-0000-783-synchronous-hooks/spec.md
new file mode 100644
index 00000000..5f28efe4
--- /dev/null
+++ b/docs/dev-sessions/2026-08-08-0000-783-synchronous-hooks/spec.md
@@ -0,0 +1,34 @@
+**Concept from opencode:**
+While `opencode` has event emitters, its plugin architecture relies on an explicit lifecycle middleware chain (`Hooks` interface) that allows plugins to synchronously intercept and mutate the LLM request before execution.
+
+**How `decafclaw` could implement this:**
+`decafclaw` heavily relies on an async EventBus (`events.py`), which is great for "fire-and-forget" but prone to race conditions when used for mutation.
+
+**Proposed Implementation:**
+- Formalize a middleware/hook chain (e.g., `Context.add_interceptor(TurnLifecycle.BEFORE_LLM_CALL, my_hook)`).
+- Allow external skills to safely and synchronously alter prompt context, tool execution parameters, or LLM routing right before a request goes out.
+
+## Verifiable acceptance criteria
+
+- CRITERION: WHEN an interceptor is registered via `Context.add_interceptor(TurnLifecycle.BEFORE_LLM_CALL, hook)`, the system SHALL synchronously invoke the hook with `(ctx, messages, tools)` before sending the LLM request, allowing it to modify the arguments in-place.
+  CHECK: `pytest tests/test_interceptor_hooks.py::test_before_llm_call_hook_mutates_messages` (asserts that a test hook appending a system message successfully alters the outgoing LLM call) passes.
+  VERIFIED DISCRIMINATING: No such test or `add_interceptor` method exists today.
+
+- CRITERION: WHEN multiple interceptors are registered for the same phase, the system SHALL execute them synchronously in the order they were added.
+  CHECK: `pytest tests/test_interceptor_hooks.py::test_interceptors_execute_in_order` (asserts that hooks appending to a shared list execute in registration order) passes.
+  VERIFIED DISCRIMINATING: No such test or mechanism exists today.
+
+## Regression guards
+
+- GUARD: `make test` — The existing EventBus behavior and `run_agent_turn` loop remain functionally equivalent when no hooks are registered. Passes today.
+
+## Tier: auto-ok
+
+The criteria reduce the hook mechanism to concrete testable behavior. Assuming we start with just the `BEFORE_LLM_CALL` phase, no subjective human judgment is required to verify the implementation.
+
+## Design decisions
+
+- **Decision:** Only support the `BEFORE_LLM_CALL` phase in this initial PR.
+  - **Why:** To keep the initial implementation scoped and focused as requested in the issue, and confirmed by user.
+  - **Rejected:** Supporting multiple lifecycle phases at once.
+
diff --git a/docs/dev-sessions/2026-08-10-1200-779-concurrency-durable-input-queues/checks.md b/docs/dev-sessions/2026-08-10-1200-779-concurrency-durable-input-queues/checks.md
new file mode 100644
index 00000000..31d8040b
--- /dev/null
+++ b/docs/dev-sessions/2026-08-10-1200-779-concurrency-durable-input-queues/checks.md
@@ -0,0 +1,34 @@
+# Frozen acceptance checks
+
+**Source:** https://github.com/lmorchard/decafclaw/issues/779
+**Frozen at:** 98bb7c4 (recorded below)
+**Check files — read-only from Phase 1 onward:**
+- `tests/test_conversation_manager.py`
+
+## C1
+CRITERION: WHEN `enqueue_turn` is called, THEN the system SHALL write the incoming turn to a durable JSONL session inbox file instead of an in-memory queue.
+CHECK: `pytest tests/test_conversation_manager.py::test_enqueue_turn_writes_to_jsonl` passes.
+AT FREEZE: (pending)
+
+## C2
+CRITERION: GIVEN an active conversation, THEN the system SHALL run a detached background worker task that continuously polls and drains its JSONL inbox serially.
+CHECK: `pytest tests/test_conversation_manager.py::test_inbox_drained_by_worker` passes.
+AT FREEZE: (pending)
+
+## C3
+CRITERION: GIVEN a server restart, WHEN the system initializes, THEN it SHALL process any pending turns found in the JSONL session inbox exactly once.
+CHECK: `pytest tests/test_conversation_manager.py::test_pending_inputs_survive_restart` passes.
+AT FREEZE: (pending)
+
+## Guards
+- G1: `pytest tests/test_conversation_manager.py -k "not test_enqueue_turn_writes_to_jsonl and not test_inbox_drained_by_worker and not test_pending_inputs_survive_restart"`
+- G2: `pytest tests/test_runner.py` (and the rest of the test suite)
+
+## Adjudication
+- C1: strengthened — explicitly asserts `pending_messages` is empty to prove legacy queue is unused.
+- C2: strengthened — added active counter and delay to fake runner to prove serialization.
+- C3: strengthened — tightened assertion to `called.count("surviving turn") == 1` to prove exactly-once execution.
+- G1: accepted — the suite still passes at freeze (before changes).
+- G2: accepted — the suite still passes at freeze.
+
+## Amendments
diff --git a/docs/dev-sessions/2026-08-10-1200-779-concurrency-durable-input-queues/plan.md b/docs/dev-sessions/2026-08-10-1200-779-concurrency-durable-input-queues/plan.md
new file mode 100644
index 00000000..2fb91802
--- /dev/null
+++ b/docs/dev-sessions/2026-08-10-1200-779-concurrency-durable-input-queues/plan.md
@@ -0,0 +1,61 @@
+# Concurrency: Durable Input Queues Implementation Plan
+
+**Goal:** Migrate `conversation_manager.py`'s input handling to a durable JSONL-backed inbox to provide flawless concurrency and survive bot restarts.
+
+**Source issue:** https://github.com/lmorchard/decafclaw/issues/779 — **Tier:** `auto-ok` (All criteria are verifiable via unit tests without human judgment. Adding a JSONL-backed queue for incoming turns does not touch risk-gated paths.)
+
+**Approach:**
+- Migrate `conversation_manager.py`'s input handling from in-memory dispatch to a durable, SQLite-backed Inbox queue (similar to how `notifications.py` uses JSONL, but for incoming turns). Wait, design decisions specify "backed by JSONL files rather than SQLite".
+- Have a single `asyncio.Task` per active conversation that loops and drains this inbox.
+- Ensures that if the server crashes mid-turn, pending human inputs or schedule wakes are processed exactly once on restart.
+
+**Criteria:** 
+- C1: `enqueue_turn` writes to JSONL inbox instead of memory queue.
+- C2: Detached background worker drains JSONL inbox serially.
+- C3: Startup scan processes pending turns in JSONL inbox.
+
+---
+
+## Phase 0: Freeze the acceptance checks
+
+Write `checks.md` and author the tests the checks name, per `references/frozen-checks.md`.
+No implementation in this phase.
+
+**Files:**
+- Create: `{session-dir}/checks.md`
+- Modify: `tests/test_conversation_manager.py`
+
+**Verification — automated:**
+- [x] Every criterion's check runs and fails for the expected reason.
+- [x] Every guard runs and passes.
+- [x] Check-reviewer dispatched read-only; `## Adjudication` in `checks.md` recorded.
+- [x] Freeze commit made; sha recorded in `checks.md`.
+
+---
+
+## Phase 1: Migrate enqueue to write JSONL and run detached drain tasks
+
+Migrate `ConversationManager` to use JSONL files for the inbox, starting background worker tasks to drain them.
+
+**Advances:** C1, C2, C3
+
+**Micro-tasks (Atomic checkbox steps):**
+- [x] Add logic in `enqueue_turn` to append the incoming turn payload (JSON) to `{conv_id}/inbox.jsonl` using `sidecar_path`.
+- [x] Instead of pushing to an in-memory queue or waiting inline, `enqueue_turn` should now create/ensure a `_drain_inbox` `asyncio.Task` is running for the conversation.
+- [x] Implement `_drain_inbox` task: loops reading lines from `inbox.jsonl`. For each line, runs `run_agent_turn`, then rewrites the file without the processed line (or renames and starts a new one). It should run serially and sleep briefly or use an `asyncio.Event` to wake up when new items are added.
+- [x] Update `startup_scan` to scan for pending turns in `inbox.jsonl` files and start `_drain_inbox` tasks.
+- [x] Update test assertions that assume `enqueue_turn` directly returns a completed future or `pending_messages` exists. (Refactor breaking tests).
+
+**Files:**
+- Modify: `src/decafclaw/conversation_manager.py`
+
+**Key changes:**
+- `def enqueue_turn` writes to JSONL and triggers worker.
+- `async def _drain_inbox(self, state: ConversationState)`
+
+**Verification — automated:**
+- [x] C1's check passes: `uv run pytest tests/test_conversation_manager.py::test_enqueue_turn_writes_to_jsonl`
+- [x] C2's check passes: `uv run pytest tests/test_conversation_manager.py::test_inbox_drained_by_worker`
+- [x] C3's check passes: `uv run pytest tests/test_conversation_manager.py::test_pending_inputs_survive_restart`
+- [x] Guards still pass: `uv run pytest tests/test_conversation_manager.py -k "not test_enqueue_turn_writes_to_jsonl and not test_inbox_drained_by_worker and not test_pending_inputs_survive_restart"`
+- [x] `make test` passes (no regression)
diff --git a/docs/dev-sessions/2026-08-10-1200-779-concurrency-durable-input-queues/spec.md b/docs/dev-sessions/2026-08-10-1200-779-concurrency-durable-input-queues/spec.md
new file mode 100644
index 00000000..f89da3aa
--- /dev/null
+++ b/docs/dev-sessions/2026-08-10-1200-779-concurrency-durable-input-queues/spec.md
@@ -0,0 +1,33 @@
+**Concept from opencode:**
+User inputs and background events are never directly pushed into an actively running agent loop. They are persisted to a `SessionInputTable` in SQLite. A detached background worker constantly "drains" this queue. This provides flawless concurrency, persistent state across bot restarts, and an elegant way to handle UI interruptions.
+
+**How `decafclaw` could implement this:**
+`decafclaw` uses `conversation_manager.py` with an in-memory `asyncio` task to coordinate `enqueue_turn()`. It relies on a `_busy_flags` dictionary which can sometimes cause race conditions with fast UI inputs or background schedule wakes (`child_agent`).
+
+**Proposed Implementation:**
+- Migrate `conversation_manager.py`'s input handling from in-memory dispatch to a durable, SQLite-backed `Inbox` queue (similar to how `notifications.py` uses JSONL, but for incoming turns).
+- Have a single `asyncio.Task` per active conversation that loops and drains this inbox.
+- Ensures that if the server crashes mid-turn, pending human inputs or schedule wakes are processed exactly once on restart.
+
+### Acceptance Criteria
+
+- CRITERION: WHEN `enqueue_turn` is called, THEN the system SHALL write the incoming turn to a durable JSONL session inbox file instead of an in-memory queue.
+  CHECK: `pytest tests/test_conversation_manager.py::test_enqueue_turn_writes_to_jsonl` passes (asserting a file append occurs and the line exists).
+
+- CRITERION: GIVEN an active conversation, THEN the system SHALL run a detached background worker task that continuously polls and drains its JSONL inbox serially.
+  CHECK: `pytest tests/test_conversation_manager.py::test_inbox_drained_by_worker` passes (asserting queued items are dispatched and subsequently removed from the file/queue).
+
+- CRITERION: GIVEN a server restart, WHEN the system initializes, THEN it SHALL process any pending turns found in the JSONL session inbox exactly once.
+  CHECK: `pytest tests/test_conversation_manager.py::test_pending_inputs_survive_restart` passes (asserting a manager initialized against a pre-populated JSONL file processes those turns on startup).
+
+### Regression Guards
+
+- GUARD: Existing integrations (HTTP server, Mattermost) still successfully enqueue turns and receive responses without knowing about the JSONL layer. Passes today.
+- GUARD: The `busy` state lock mechanism continues to prevent concurrent processing of turns for the same conversation. Passes today.
+
+## Tier: auto-ok
+**Reason:** All criteria are verifiable via unit tests without human judgment. Adding a JSONL-backed queue for incoming turns does not touch risk-gated paths (no secrets, no auth, no production data migration since the previous queue was purely in-memory).
+
+### Design decisions
+- The session inbox will be backed by JSONL files rather than SQLite, per user correction on the initial proposal. This aligns with how `notifications.py` uses JSONL.
+
diff --git a/docs/dev-sessions/2026-08-13-145-steering-messages/spec.md b/docs/dev-sessions/2026-08-13-145-steering-messages/spec.md
new file mode 100644
index 00000000..3e21bd8c
--- /dev/null
+++ b/docs/dev-sessions/2026-08-13-145-steering-messages/spec.md
@@ -0,0 +1,37 @@
+## Summary
+
+Allow users to send messages while the agent is mid-turn, with two modes: **steering** (interrupt after current tool call) and **follow-up** (queue for after the agent finishes).
+
+## Motivation
+
+Inspired by Pi/OpenClaw's steering and follow-up queues. Currently, DecafClaw's `busy` flag blocks new input during a turn. Users can't say "actually stop, try a different approach" or queue up additional context while the agent is working.
+
+## Design Notes
+
+- Steering messages interrupt the agent loop after the current tool call completes (not mid-tool)
+- Follow-up messages queue and are delivered after the current turn ends
+- Could use the existing EventBus to signal steering interrupts
+- Web UI and Mattermost would both need to support sending messages while busy
+- Need to consider how this interacts with the reflection judge (skip reflection on steered turns?)
+
+## Prior Art
+
+- Pi agent core implements steering and follow-up queues with "one-at-a-time" or "all" delivery modes
+- OpenClaw inherits this from the Pi SDK
+
+## Verifiable acceptance criteria
+
+- CRITERION: WHEN a user sends a steering message while the agent is executing tool calls, THE SYSTEM SHALL interrupt the agent loop after the current tool call completes and ingest the steering message.
+  - CHECK: `pytest tests/test_steering.py::test_steering_interrupts_after_tool_call` passes.
+
+- CRITERION: WHEN a user sends a follow-up message while the agent is busy, THE SYSTEM SHALL queue the message and deliver it as a new turn after the current agent turn finishes.
+  - CHECK: `pytest tests/test_steering.py::test_follow_up_message_queued` passes.
+
+## Regression guards
+
+- GUARD: `pytest tests/test_agent.py` passes and existing agent turn flow is preserved.
+
+## Tier: auto-ok
+
+**Reason:** Approved by human review.
+
diff --git a/docs/dev-sessions/20260813T191028Z-garden-folders/plan.md b/docs/dev-sessions/20260813T191028Z-garden-folders/plan.md
new file mode 100644
index 00000000..00654d1b
--- /dev/null
+++ b/docs/dev-sessions/20260813T191028Z-garden-folders/plan.md
@@ -0,0 +1,51 @@
+# Plan
+
+## Phase 0: Freeze acceptance checks
+
+- [x] write `checks.md` with criteria from the spec.
+- [x] author tests `tests/test_garden_folders.py`.
+- [x] run tests to observe failure.
+- [x] review tests, adjudicate, and record freeze sha.
+
+## Phase 1: Define configuration and models
+
+Advances: C1, C3
+
+1. [x] Define `GardenConfig` dataclass in `src/decafclaw/skills/garden/tools.py` with `dry_run: bool = False`.
+2. [x] Add the `skill_config` handling to tools.py so `tool_vault_reorganize_folders` can access it.
+3. [x] Update `init` in `src/decafclaw/skills/garden/tools.py` if necessary to register the config.
+
+- [x] Check: `pytest tests/test_garden_folders.py::test_garden_folder_move_dry_run_and_respect_user_folders` passes (partially, depends on Phase 2).
+
+## Phase 2: Implement Folder Reorganization Tool
+
+Advances: C1, C3
+
+1. [x] Create a new async tool `tool_vault_reorganize_folders` in `src/decafclaw/skills/garden/tools.py`.
+2. [x] Implement clustering logic.
+3. [x] Determine target folder `agent/pages/{cluster_topic}/`.
+4. [x] Check if it's `dry_run`. If `True`, just log/return planned moves. If `False`, move the files using `Path.rename()`.
+5. [x] Expose this tool in `TOOL_DEFINITIONS`.
+
+- [x] Check: `pytest tests/test_garden_folders.py::test_garden_detects_and_suggests_cluster_folder_moves` passes.
+- [x] Check: `pytest tests/test_garden_folders.py::test_garden_folder_move_dry_run_and_respect_user_folders` passes.
+
+## Phase 3: Implement Wiki-links Updating
+
+Advances: C2
+
+1. [x] When a file is moved, find all other pages in the vault that contain `[[OldName]]` wiki-links.
+2. [x] Update them to point to `[[NewFolder/OldName|OldName]]`.
+
+- [x] Check: `pytest tests/test_garden_folders.py::test_garden_folder_move_updates_links` passes.
+
+## Phase 4: Update Skill Prompt
+
+Advances: C1
+
+1. [x] Edit `src/decafclaw/skills/garden/SKILL.md` to add `Step 2.5: Reorganize Clusters into Folders` instructing the agent to call `vault_reorganize_folders`.
+
+## Phase 5: Verification
+
+- [x] Run all criteria checks.
+- [x] Run guards.
diff --git a/docs/dev-sessions/20260813T191028Z-garden-folders/spec.md b/docs/dev-sessions/20260813T191028Z-garden-folders/spec.md
new file mode 100644
index 00000000..47975c36
--- /dev/null
+++ b/docs/dev-sessions/20260813T191028Z-garden-folders/spec.md
@@ -0,0 +1,45 @@
+## Context
+
+Follow-up from #170 (vault folder support). Once folders are available, the garden skill should be able to suggest or execute page moves into folders during its periodic maintenance sweeps.
+
+## Idea
+
+During garden maintenance:
+- Detect clusters of related pages at the vault root (or in any folder)
+- When 3+ pages share a clear topic, suggest consolidating them into a folder
+- Could be fully automatic or produce a "proposed reorganization" for user review
+- Should update [[wiki-links]] if pages move (or rely on stem-based resolution to handle it)
+
+## Considerations
+
+- Need to be conservative — dont break existing links or surprise users
+- May want a dry-run mode that logs proposed moves without executing
+- Should respect any user-created folder structure (dont flatten what the user organized)
+- Embedding re-indexing happens automatically on rename (implemented in #170)
+
+## Related
+
+- #170 — vault folder support (parent feature)
+
+## Verifiable acceptance criteria
+
+- CRITERION: WHEN the garden maintenance sweep runs and detects 3+ agent pages sharing a common topic cluster at the vault root (or any folder), THE GARDEN SKILL SHALL execute moving those pages into a dedicated subdirectory under `agent/pages/` (unless `dry_run` is enabled via configuration).
+  - CHECK: `pytest tests/test_garden_folders.py::test_garden_detects_and_suggests_cluster_folder_moves` passes.
+- CRITERION: WHEN pages are moved into a folder during garden reorganization, THE VAULT SYSTEM SHALL update existing `[[wiki-links]]` pointing to those pages (or correctly resolve them via stem-based resolution).
+  - CHECK: `pytest tests/test_garden_folders.py::test_garden_folder_move_updates_links` passes.
+- CRITERION: GIVEN `dry_run` configuration is enabled WHEN garden runs page reorganization THEN it SHALL log or report proposed moves without modifying files on disk or flattening user-created folder hierarchies.
+  - CHECK: `pytest tests/test_garden_folders.py::test_garden_folder_move_dry_run_and_respect_user_folders` passes.
+
+## Regression guards
+
+- GUARD: `pytest tests/test_vault_tools.py tests/test_garden_recompute.py` passes — existing vault operations and importance recompute remain fully functional.
+
+## Tier: auto-ok
+
+Reason: All acceptance criteria have concrete runnable tests, and the design decision regarding default behavior (`dry_run=False` by default with config support to enable `dry_run`) has been approved by the repo owner.
+
+## Design decisions
+
+- **Decision:** Folder reorganization executes by default (`dry_run=False`), with a configuration setting (`garden.dry_run = True`) available to enable dry-run mode.
+  - **Why:** Aligned with owner feedback ("let's not make this dry_run=True by default. Let's support a configuration setting to disable, but enable by default").
+  - **Rejected:** Dry-run by default requiring explicit opt-in.
\ No newline at end of file
diff --git a/scripts/prune_agent_worktrees.py b/scripts/prune_agent_worktrees.py
new file mode 100755
index 00000000..7adaaf0d
--- /dev/null
+++ b/scripts/prune_agent_worktrees.py
@@ -0,0 +1,258 @@
+#!/usr/bin/env python3
+"""Prune git worktrees and local branches whose work has already landed.
+
+Agent-driven development leaves a lot of debris: this repo accumulated 21
+worktrees and 30 local branches, most of them holding work that had already
+merged. This script decides which of those are safe to drop and, with
+``--apply``, drops them.
+
+A branch is considered LANDED if either:
+
+* it is an ancestor of ``origin/main`` (merge commit), or
+* it has a MERGED pull request and its last commit predates the merge
+  (squash merge, which leaves the branch off main's ancestry).
+
+Anything else is reported as UNLANDED and never touched -- that includes
+abandoned experiments, which are the whole reason this isn't just
+``git worktree prune``.
+
+Dry run by default. See ``make prune-worktrees-dry`` / ``make prune-worktrees``.
+"""
+
+import argparse
+import json
+import shutil
+import subprocess
+import sys
+from dataclasses import dataclass, field
+from pathlib import Path
+
+REPO_ROOT = Path(__file__).resolve().parent.parent
+PROTECTED = {"main"}
+
+
+def git(*args: str, check: bool = True) -> str:
+    """Run a git command in the repo root and return stripped stdout."""
+    proc = subprocess.run(
+        ["git", *args],
+        cwd=REPO_ROOT,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if check and proc.returncode != 0:
+        raise RuntimeError(f"git {' '.join(args)} failed: {proc.stderr.strip()}")
+    return proc.stdout.strip()
+
+
+@dataclass
+class Branch:
+    name: str
+    verdict: str = "UNLANDED"
+    reason: str = ""
+    worktree: Path | None = None
+    dirty_tracked: list[str] = field(default_factory=list)
+    untracked: list[str] = field(default_factory=list)
+
+    @property
+    def landed(self) -> bool:
+        return self.verdict == "LANDED"
+
+
+def scan_worktrees() -> tuple[dict[str, Path], list[tuple[Path, str]]]:
+    """Return (branch name -> worktree path, [(path, sha) for detached ones]).
+
+    Detached worktrees have no branch to key off, so a branch-driven scan would
+    miss them entirely and ``git worktree prune`` only reaps directories that
+    are already gone. They get reported so a human can decide.
+    """
+    attached: dict[str, Path] = {}
+    detached: list[tuple[Path, str]] = []
+    path: Path | None = None
+    sha = ""
+    claimed = False
+    for line in git("worktree", "list", "--porcelain").splitlines():
+        if line.startswith("worktree "):
+            if path is not None and path != REPO_ROOT and not claimed:
+                detached.append((path, sha))
+            path = Path(line.removeprefix("worktree "))
+            claimed = False
+        elif line.startswith("HEAD "):
+            sha = line.removeprefix("HEAD ")[:8]
+        elif line.startswith("branch ") and path is not None:
+            claimed = True
+            if path != REPO_ROOT:
+                attached[line.removeprefix("branch refs/heads/")] = path
+    if path is not None and path != REPO_ROOT and not claimed:
+        detached.append((path, sha))
+    return attached, detached
+
+
+def merged_prs() -> dict[str, str]:
+    """Map head branch -> mergedAt timestamp, for every merged PR."""
+    if shutil.which("gh") is None:
+        print("warning: gh not found; squash-merged branches will read as UNLANDED")
+        return {}
+    proc = subprocess.run(
+        ["gh", "pr", "list", "--state", "merged", "--limit", "500", "--json", "headRefName,mergedAt"],
+        cwd=REPO_ROOT,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if proc.returncode != 0:
+        print(f"warning: gh pr list failed: {proc.stderr.strip()}")
+        return {}
+    return {pr["headRefName"]: pr["mergedAt"] for pr in json.loads(proc.stdout or "[]") if pr.get("mergedAt")}
+
+
+def classify(name: str, prs: dict[str, str]) -> tuple[str, str]:
+    """Decide whether a branch's work is already in origin/main."""
+    ancestor = subprocess.run(
+        ["git", "merge-base", "--is-ancestor", name, "origin/main"],
+        cwd=REPO_ROOT,
+        capture_output=True,
+        check=False,
+    )
+    if ancestor.returncode == 0:
+        return "LANDED", "ancestor of origin/main"
+
+    merged_at = prs.get(name)
+    if merged_at:
+        last = git("log", "-1", "--format=%cI", name)
+        if last <= merged_at:
+            return "LANDED", f"squash-merged {merged_at[:10]}"
+        return "UNLANDED", f"commits after PR merged {merged_at[:10]}"
+
+    ahead = git("rev-list", "--count", f"origin/main..{name}")
+    return "UNLANDED", f"no merged PR, {ahead} commit(s) ahead"
+
+
+def inspect_worktree(wt: Path) -> tuple[list[str], list[str]]:
+    """Return (tracked modifications, untracked paths) for a worktree."""
+    proc = subprocess.run(
+        ["git", "status", "--porcelain"],
+        cwd=wt,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    tracked, untracked = [], []
+    for line in proc.stdout.splitlines():
+        status, _, path = line[:2], line[2:3], line[3:]
+        (untracked if status == "??" else tracked).append(path)
+    return tracked, untracked
+
+
+def collect(only: str | None) -> tuple[list[Branch], list[tuple[Path, str]]]:
+    wts, detached = scan_worktrees()
+    prs = merged_prs()
+    current = git("rev-parse", "--abbrev-ref", "HEAD")
+
+    branches = []
+    for name in git("for-each-ref", "--format=%(refname:short)", "refs/heads/").splitlines():
+        if name in PROTECTED or name == current:
+            continue
+        if only and only not in name and only not in str(wts.get(name, "")):
+            continue
+        verdict, reason = classify(name, prs)
+        b = Branch(name=name, verdict=verdict, reason=reason, worktree=wts.get(name))
+        if b.worktree and b.worktree.exists():
+            b.dirty_tracked, b.untracked = inspect_worktree(b.worktree)
+        branches.append(b)
+    return branches, detached
+
+
+def report(branches: list[Branch], detached: list[tuple[Path, str]]) -> None:
+    for verdict in ("LANDED", "UNLANDED"):
+        rows = [b for b in branches if b.verdict == verdict]
+        print(f"\n=== {verdict} ({len(rows)}) ===")
+        for b in sorted(rows, key=lambda x: x.name):
+            marks = []
+            if b.worktree:
+                marks.append("worktree")
+            if b.dirty_tracked:
+                marks.append(f"{len(b.dirty_tracked)} TRACKED EDIT(S)")
+            if b.untracked:
+                marks.append(f"{len(b.untracked)} untracked")
+            print(f"  {b.name:<44} {b.reason:<34} {', '.join(marks)}")
+            for path in b.dirty_tracked:
+                print(f"      ! {path}")
+
+    if detached:
+        print(f"\n=== DETACHED worktrees ({len(detached)}) - not pruned, review by hand ===")
+        for path, sha in detached:
+            in_main = (
+                subprocess.run(
+                    ["git", "merge-base", "--is-ancestor", sha, "origin/main"],
+                    cwd=REPO_ROOT,
+                    capture_output=True,
+                    check=False,
+                ).returncode
+                == 0
+            )
+            state = "in origin/main" if in_main else "NOT in origin/main"
+            print(f"  {str(path):<60} {sha}  {state}")
+
+
+def apply(branches: list[Branch], force: bool) -> int:
+    failures = 0
+    for b in sorted(branches, key=lambda x: x.name):
+        if not b.landed:
+            continue
+        if b.dirty_tracked and not force:
+            print(f"SKIP     {b.name}: {len(b.dirty_tracked)} tracked edit(s); use --force")
+            continue
+        if b.worktree:
+            out = subprocess.run(
+                ["git", "worktree", "remove", "--force", str(b.worktree)],
+                cwd=REPO_ROOT,
+                capture_output=True,
+                text=True,
+                check=False,
+            )
+            if out.returncode != 0:
+                print(f"FAILED   worktree {b.worktree}: {out.stderr.strip()}")
+                failures += 1
+                continue
+            print(f"REMOVED  worktree {b.worktree}")
+        sha = git("rev-parse", "--short", b.name)
+        out = subprocess.run(
+            ["git", "branch", "-D", b.name],
+            cwd=REPO_ROOT,
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        if out.returncode != 0:
+            print(f"FAILED   branch {b.name}: {out.stderr.strip()}")
+            failures += 1
+        else:
+            print(f"DELETED  branch {b.name} ({sha}) - recover with: git branch {b.name} {sha}")
+    return failures
+
+
+def main() -> int:
+    p = argparse.ArgumentParser(description=__doc__)
+    p.add_argument("--apply", action="store_true", help="actually remove worktrees and delete branches")
+    p.add_argument("--force", action="store_true", help="also prune landed worktrees that have tracked edits")
+    p.add_argument("--only", metavar="SUBSTR", help="limit to branches or worktree paths containing SUBSTR")
+    args = p.parse_args()
+
+    git("fetch", "origin", "--quiet")
+    branches, detached = collect(args.only)
+    report(branches, detached)
+
+    landed = [b for b in branches if b.landed]
+    if not args.apply:
+        print(f"\nDry run. {len(landed)} branch(es) would be pruned; re-run with --apply.")
+        return 0
+
+    print(f"\nPruning {len(landed)} landed branch(es)...")
+    failures = apply(branches, args.force)
+    git("worktree", "prune")
+    return 1 if failures else 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())