diff --git a/manager/core/.env.example b/manager/core/.env.example
index 1a2385e..ca16723 100644
--- a/manager/core/.env.example
+++ b/manager/core/.env.example
@@ -124,6 +124,14 @@ BOARD_COMMIT_MOVES=
 BOARD_SYNC=
 BOARD_SYNC_INTERVAL=30
 
+# Seconds between passes of the phase runner: a phase card being run
+# (**Type:** Phase, in in-progress/) has its list worked into its own
+# `phase/<stem>` branch one card at a time. Each pass recomputes what the
+# phase needs from the cards and from git and does the one next thing, so
+# raising this only makes a phase slower to notice, never wrong. Nothing
+# runs at all while no phase is being run.
+BOARD_PHASE_INTERVAL=30
+
 # Seconds between disk polls of the stage directories.
 BOARD_WATCH_INTERVAL=2
 
diff --git a/manager/core/board.py b/manager/core/board.py
index 556ec46..9bf289d 100755
--- a/manager/core/board.py
+++ b/manager/core/board.py
@@ -14,6 +14,7 @@ task files, but the tasks work as a plain folder kanban without it. See
     events.py     hook payloads → displayable events, session registry
     sync.py       origin/main as the shared board: push on move, pull on a beat
     agents.py     headless work/review agents: launch, reap, stop, diff
+    phases.py     a phase run: its own branch, its members merged into it
     watch.py      2s disk poller narrating moves made outside the API
     httpd.py      HTTP routes, SSE stream, the page itself
     .prompts/     agent prompt templates (read fresh on every launch)
@@ -34,6 +35,7 @@ import drive
 import events
 import github
 import httpd
+import phases
 import state
 import sync
 import watch
@@ -66,6 +68,7 @@ def main() -> None:
     threading.Thread(target=watch.watcher, daemon=True).start()
     threading.Thread(target=github.poller, daemon=True).start()
     threading.Thread(target=github.reconcile, daemon=True).start()
+    threading.Thread(target=phases.beat, daemon=True).start()
     if config.SYNC:
         # Team mode: board commits publish themselves and a beat pulls what
         # the other boards published. Off, neither thread nor hook exists.
diff --git a/manager/core/config.py b/manager/core/config.py
index 1563f6d..3e80f24 100644
--- a/manager/core/config.py
+++ b/manager/core/config.py
@@ -175,6 +175,11 @@ SYNC_INTERVAL = float(setting("BOARD_SYNC_INTERVAL", "30"))
 # tasks/ stays a hand job.
 COMMIT_MOVES = flag("BOARD_COMMIT_MOVES") or SYNC
 
+# How often the phase runner takes a pass: recompute what each running
+# phase needs and do the one next thing. A beat, not an agent — it costs a
+# handful of git commands and nothing at all when no phase is running.
+PHASE_INTERVAL = float(setting("BOARD_PHASE_INTERVAL", "30"))
+
 WATCH_INTERVAL = float(setting("BOARD_WATCH_INTERVAL", "2"))
 EVENTS_CAP = int(setting("BOARD_EVENTS_CAP", "800"))
 BOARD_EVENTS_CAP = int(setting("BOARD_HISTORY_CAP", "300"))
diff --git a/manager/core/httpd.py b/manager/core/httpd.py
index 6edd014..379d9e9 100644
--- a/manager/core/httpd.py
+++ b/manager/core/httpd.py
@@ -17,6 +17,7 @@ import config
 import drive
 import events
 import github
+import phases
 import state
 import sync
 import taskfiles
@@ -34,6 +35,9 @@ def state_payload() -> dict:
         "sessions": sessions,
         "agents": agents.list_public(),
         "prs": github.public_state(),
+        # what the last pass of the phase runner saw: per running phase
+        # card, its branch and each member's state
+        "phases": phases.public_state(),
         "drive": drive.public(),
         "hasDriver": config.driver_path() is not None,
         "branches": github.task_branches(),
@@ -190,6 +194,13 @@ class Handler(BaseHTTPRequestHandler):
                 payload = self._read_body()
                 agent = agents.start_pr_fix(payload["file"], payload["stage"])
                 self._json(200, {"agent": agent})
+            elif path == "/api/phase/run":
+                payload = self._read_body()
+                # takeover carries the same meaning it does for a launch:
+                # the deliberate second click on someone else's card
+                self._json(200, {"phase": phases.start_phase(
+                    payload["file"], payload["stage"],
+                    bool(payload.get("takeover")))})
             elif path == "/api/pr/open":
                 payload = self._read_body()
                 self._json(200, {"url": github.open_pr_now(payload["file"])})
diff --git a/manager/core/phases.py b/manager/core/phases.py
new file mode 100644
index 0000000..e92b611
--- /dev/null
+++ b/manager/core/phases.py
@@ -0,0 +1,579 @@
+"""A phase runs itself, on a branch of its own.
+
+A phase is a card that lists its cards (`taskfiles.weave_phases`). Running
+one means working that list into a single integration branch: `phase/<stem>`
+cut from the newest main, each member branched from the phase's tip, run
+headless, merged back when its checks are green, and the next one started.
+At the end one PR into `main`, for a human. The board never merges into
+`main` — a phase branch is the board's own, and merging into it is
+bookkeeping in the same family as committing a move.
+
+**The runner is a beat, not an agent.** Everything it decides is already
+structured state — a card's stage, a PR's CI verdict, whether one branch is
+contained in another — so an agent paid to poll would be the wrong tool at
+the wrong price.
+
+**The beat is stateless.** On each pass it recomputes, from disk and from
+git, which members are finished, which is first unfinished and what that
+one needs. It holds no registry of where a phase "is": a restarted board
+resumes a phase by looking, and the same logic answers "what now?" whether
+the last event was a launch, a merge or a crash. Two things carry the
+memory, and both are durable:
+
+- **git** — a member is merged when its branch is contained in the phase
+  branch. That is what makes a restart safe from repeating a merge.
+- **the phase card** — a `## Phase log` section the runner appends one line
+  to per decision (a run started, a member started, a member merged, a
+  halt). It is the record a person reads, and the only thing that can tell
+  "the phase already started this member and its run ended badly" from "the
+  phase has not reached this member yet". Without it a restart would
+  silently relaunch a run that died.
+
+**Halt, never skip.** Five conditions stop a phase, each of them already a
+visible state on the card: a member that declined (`NOT READY`), a run that
+exited non-zero, a clean exit that committed nothing, CI red, and a merge
+into the phase branch that is not mechanical. A phase that steps over a
+failed card builds the rest on a foundation that never landed. Halting is
+recorded in the log and nothing retries by itself; running the phase again
+is a person's decision and appends the line that clears the halt.
+
+**One board runs it.** The actor rule decides, and the phase card's
+**Assignee** is where it is written down — the same claim that gates
+starting work. A replica renders the phase and advances nothing.
+"""
+
+from __future__ import annotations
+
+import re
+import subprocess
+import threading
+import time
+from pathlib import Path
+
+import agents
+import config
+import github
+import state
+import taskfiles
+
+LOG_HEADING = "Phase log"
+
+# One log line: `- <date> <time> · <entry>`. The entry is what is parsed;
+# the stamp in front of it is for the reader.
+LOG_LINE_RE = re.compile(r"^-\s+\d{4}-\d{2}-\d{2}\s+\d{2}:\d{2}\s+·\s+(.+?)\s*$")
+STARTED_RE = re.compile(r"^(\d+) started\b")
+HALTED_RE = re.compile(r"^halted(?: at (\d+))? — (.*)$")
+RUN_RE = re.compile(r"^run started\b")
+
+# What the last computed pass saw, for the API to render without paying for
+# a git walk on every request. A display cache, not a decision: nothing here
+# is ever read back to decide anything.
+SNAPSHOTS: dict[str, dict] = {}
+
+_LOCK = threading.Lock()          # one pass at a time, whoever asks for it
+
+
+def _git(*args: str, cwd=None, timeout: float = 60) -> subprocess.CompletedProcess:
+    try:
+        return subprocess.run(["git", "-C", str(cwd or config.REPO), *args],
+                              capture_output=True, text=True, timeout=timeout)
+    except subprocess.TimeoutExpired:
+        return subprocess.CompletedProcess(args, 128, "", "timed out")
+    except OSError as exc:
+        return subprocess.CompletedProcess(args, 128, "", str(exc))
+
+
+def _branch_exists(branch: str) -> bool:
+    return _git("rev-parse", "--verify", "--quiet", branch).returncode == 0
+
+
+def _contains(branch: str, tip: str) -> bool:
+    """Is `branch` already in `tip`? The stateless record of a merge."""
+    return _git("merge-base", "--is-ancestor", branch, tip).returncode == 0
+
+
+def _say(filename: str, summary: str) -> None:
+    state.record_board_event({"kind": "phase", "actor": "board",
+                              "file": filename, "summary": summary})
+
+
+# ── the log on the phase card ──────────────────────────────────────────
+
+
+def log_entries(text: str) -> list[str]:
+    """The `## Phase log` section's entries, in the order they were made."""
+    section = re.search(rf"^##\s+{LOG_HEADING}\s*$(.*?)(?=^##\s|\Z)", text,
+                        re.MULTILINE | re.DOTALL)
+    if not section:
+        return []
+    entries = []
+    for line in section.group(1).splitlines():
+        match = LOG_LINE_RE.match(line.strip())
+        if match:
+            entries.append(match.group(1))
+    return entries
+
+
+def _record(phase: dict, entry: str) -> None:
+    """One line into the log, where it reaches git like every other write
+    the board makes to a task file. The stage is asked of the disk: a pass
+    that finishes a phase moves the card, and the record is written to
+    wherever the card actually is."""
+    stage = taskfiles.find_stage_of(phase["file"]) or phase["stage"]
+    stamp = time.strftime("%Y-%m-%d %H:%M")
+    taskfiles.append_to_section(phase["file"], stage, LOG_HEADING,
+                                f"- {stamp} · {entry}", "phase log")
+
+
+def _this_run(entries: list[str]) -> list[str]:
+    """The entries since the last `run started` — the run in progress.
+
+    A run is scoped to its own line, so running a phase again after a halt
+    starts from a clean slate: a member whose run died is launchable again,
+    which is exactly what asking for the run again meant. Nothing is lost
+    by scoping it, because what has *merged* is answered by git, not here.
+    """
+    for index in range(len(entries) - 1, -1, -1):
+        if RUN_RE.match(entries[index]):
+            return entries[index + 1:]
+    return entries
+
+
+def _started(entries: list[str]) -> set[str]:
+    """Members this run has already launched, by canonical number — `07`,
+    `7` and `#007` are one card wherever a number is read or written."""
+    return {taskfiles.canonical_number(match.group(1)) for match in
+            (STARTED_RE.match(e) for e in _this_run(entries)) if match}
+
+
+def _halt_reason(entries: list[str]) -> str | None:
+    """The halt still in force, or None. A `run started` line after a halt
+    clears it — that is what running the phase again means."""
+    for entry in reversed(entries):
+        halted = HALTED_RE.match(entry)
+        if halted:
+            return (f"{halted.group(1)}: {halted.group(2)}" if halted.group(1)
+                    else halted.group(2))
+        if RUN_RE.match(entry):
+            return None
+    return None
+
+
+# ── reading the board ──────────────────────────────────────────────────
+
+
+def _cards() -> tuple[dict[str, dict], dict[str, dict]]:
+    """Every card, by filename and by canonical number, woven — so a
+    member already knows the phase that holds it."""
+    tasks = [task for stage in taskfiles.collect()["stages"]
+             for task in stage["tasks"]]
+    by_file = {task["file"]: task for task in tasks}
+    by_number = {}
+    for task in tasks:
+        if task["number"]:
+            by_number.setdefault(taskfiles.canonical_number(task["number"]), task)
+    return by_file, by_number
+
+
+def _running_on(filename: str) -> bool:
+    with state.LOCK:
+        return any(r["task"] == filename and r["status"] == "running"
+                   for r in state.AGENTS.values())
+
+
+def _failure_note(filename: str) -> str:
+    """What the run that just died said, when this board still remembers
+    it. A restart forgets, and the state says the same thing either way."""
+    with state.LOCK:
+        records = [r for r in state.AGENTS.values()
+                   if r["task"] == filename and r.get("failure")]
+    if not records:
+        return ""
+    latest = max(records, key=lambda r: r["started"])
+    excerpt = (latest["failure"].get("excerpt") or "").splitlines()
+    return excerpt[-1].strip()[:120] if excerpt else ""
+
+
+def _member_state(task: dict, branch: str, started: set[str]) -> tuple[str, str]:
+    """One member → (state, why). The five halting conditions and the three
+    ways a member can be in flight, all derived from what the card and git
+    already say.
+
+    - `merged`  — it is settled *and* its work is in the phase branch (or
+                  there was never a branch to bring)
+    - `running` — an agent is on it
+    - `ready`   — it reached review/ and its checks are not against it
+    - `waiting` — checks still running
+    - `halt`    — one of the five; the reason is for the log and the ticker
+    - `pending` — the phase has not reached it yet
+
+    Containment alone does not mean merged, and the difference matters: a
+    run that exits cleanly without committing leaves a branch that *is*
+    contained in the phase branch — an empty one. Reading that as merged is
+    exactly how a broken launch would hide, so the card has to have settled
+    into review/ or done/ as well.
+    """
+    # An agent on the card outranks everything the card says: whatever it is
+    # doing, the phase must not act on the same task underneath it — which
+    # is also what keeps a card moved by hand mid-run from being launched
+    # or merged twice.
+    if _running_on(task["file"]):
+        return "running", ""
+    member_branch = f"task/{task['file'][:-3]}"
+    has_branch = _branch_exists(member_branch)
+    settled = task["stage"] in ("review", "done")
+    if settled and (not has_branch or _contains(member_branch, branch)):
+        return "merged", ""
+    number = taskfiles.canonical_number(task["number"]) if task["number"] else ""
+    if number not in started and not has_branch:
+        return "pending", ""
+
+    # The phase started this one (or someone did). Where it ended decides.
+    if settled:
+        # Green means the checks are not against it: red halts, running
+        # waits, and a member with no checks at all advances — a project
+        # without CI must not deadlock every phase it runs.
+        ci = (github.PR_STATE.get(task["file"]) or {}).get("ci")
+        if ci == "fail":
+            return "halt", "its CI is red"
+        if ci == "running":
+            return "waiting", "its checks are still running"
+        return "ready", ""
+    if task["stage"] == "in-progress":
+        note = _failure_note(task["file"])
+        return "halt", ("its run ended without reaching review/"
+                        + (f" — {note}" if note else ""))
+    return "halt", (f"it is back in {task['stage']}/ — the run declined it, "
+                    f"or someone walked it back")
+
+
+def _snapshot(phase: dict, by_file: dict[str, dict]) -> dict:
+    """The whole phase, recomputed. Nothing is remembered between passes."""
+    branch = agents.phase_branch(phase["file"])
+    entries = log_entries(phase["body"])
+    started = _started(entries)
+    members = []
+    for listed in phase["members"]:
+        task = by_file.get(listed["file"])
+        if task is None:                      # vanished between reads
+            continue
+        member_state, why = _member_state(task, branch, started)
+        # canonical throughout: the log, `**Depends on:**` and a filename's
+        # own number all have to be the same key or none of them match
+        members.append({"number": taskfiles.canonical_number(task["number"])
+                        if task["number"] else None, "file": task["file"],
+                        "title": task["title"], "stage": task["stage"],
+                        "state": member_state, "why": why,
+                        "dependsOn": task["dependsOn"]})
+    return {"file": phase["file"], "branch": branch, "stage": phase["stage"],
+            "members": members, "halted": _halt_reason(entries),
+            # the branch is the whole of "this phase has been started"
+            "started": _branch_exists(branch)}
+
+
+def public_state() -> dict:
+    """What the API shows: the last pass's reading, per phase card."""
+    return {name: dict(snapshot) for name, snapshot in SNAPSHOTS.items()}
+
+
+# ── the phase branch ───────────────────────────────────────────────────
+
+
+class _Halt(Exception):
+    """A condition the phase stops on: the reason a person needs, and the
+    member it happened at when there is one."""
+
+    def __init__(self, reason: str, member: dict | None = None):
+        super().__init__(reason)
+        self.member = member
+
+
+def _worktree(phase: dict) -> Path:
+    """The phase branch needs a working tree to merge into — the same
+    arrangement every other branch here gets, recreated from the branch if
+    it was cleaned up."""
+    branch = agents.phase_branch(phase["file"])
+    path = config.WORKTREES / phase["file"][:-3]
+    if path.exists():
+        return path
+    config.WORKTREES.mkdir(parents=True, exist_ok=True)
+    result = _git("worktree", "add", str(path), branch)
+    if result.returncode != 0:
+        raise _Halt(f"could not put a worktree on {branch}: "
+                    f"{result.stderr.strip()[:140]}")
+    return path
+
+
+def _merge_into_phase(phase: dict, branch: str, what: str,
+                      member: dict | None = None) -> bool:
+    """Merge `branch` into the phase branch — additively, never rebasing,
+    never force-pushing. Returns whether anything landed.
+
+    A conflict is not something to be clever about: abort, leave the branch
+    exactly as it was, and halt naming the files that collided.
+    """
+    worktree = _worktree(phase)
+    phase_branch = agents.phase_branch(phase["file"])
+    if _contains(branch, phase_branch):
+        return False
+    result = _git("merge", "--no-ff", "--no-edit", "-m",
+                  f"phase: merge {branch} into {phase_branch}", branch,
+                  cwd=worktree, timeout=180)
+    if result.returncode != 0:
+        conflicted = _git("diff", "--name-only", "--diff-filter=U",
+                          cwd=worktree).stdout.split()
+        _git("merge", "--abort", cwd=worktree)
+        raise _Halt(f"merging {what} into {phase_branch} conflicts"
+                    + (f" on {', '.join(conflicted[:4])}" if conflicted else "")
+                    + " — a person has to settle it", member)
+    return True
+
+
+def _push_phase(phase: dict) -> None:
+    """Publish the phase branch. Best effort: a phase runs perfectly well
+    with no remote, and a member's PR simply does not open without one."""
+    rname = github.remote()
+    if not rname:
+        return
+    _git("push", "-u", rname, agents.phase_branch(phase["file"]), timeout=180)
+
+
+def _freshen(phase: dict) -> None:
+    """Merge main into the phase branch on the beat, so a phase that runs
+    for hours does not drift into one enormous conflict at the end. A
+    conflict here halts the phase like any other."""
+    point, _ = agents._fresh_branch_point()
+    if _merge_into_phase(phase, point or "main", "main"):
+        _say(phase["file"], f"merged main into {agents.phase_branch(phase['file'])}")
+        _push_phase(phase)
+
+
+# ── the beat ───────────────────────────────────────────────────────────
+
+
+def _mine(phase: dict) -> bool:
+    """State syncs; reactions don't. Only the board whose user holds the
+    phase card advances it — every replica renders the same phase and
+    launches nothing. Outside team mode there is one board, and it acts."""
+    if not config.COMMIT_MOVES:
+        return True
+    me = taskfiles.actor_name()
+    return bool(me) and phase.get("assignee") == me
+
+
+def _unfinished_dependencies(member: dict, snapshot: dict,
+                             by_number: dict[str, dict]) -> list[str]:
+    """`**Depends on:**` guards rather than orders: the list says what runs
+    next, this says whether it may. A dependency inside the phase is
+    finished when the phase has merged it; one outside is finished when its
+    card reaches done/ — the only claim the board can make about a card it
+    is not running."""
+    inside = {m["number"]: m for m in snapshot["members"] if m["number"]}
+    unfinished = []
+    for number in member["dependsOn"]:
+        held = inside.get(number)
+        if held is not None:
+            if held["state"] != "merged":
+                unfinished.append(number)
+            continue
+        card = by_number.get(number)
+        if card is None or card["stage"] != "done":
+            unfinished.append(number)
+    return unfinished
+
+
+def _launch(phase: dict, member: dict) -> None:
+    """Start one member: move its card to in-progress if it is not there
+    yet (moving a card there is the commitment work starts from), then
+    launch the ordinary headless work agent on it — which branches from the
+    phase's tip rather than from main."""
+    if member["stage"] != "in-progress":
+        taskfiles.move_task(member["file"], member["stage"], "in-progress",
+                            actor="phase")
+    # Recorded before the launch, not after: a crash in between must leave
+    # the log saying a launch may have happened, never the reverse.
+    _record(phase, f"{member['number']} started")
+    try:
+        agents.start_agent(member["file"], "in-progress")
+    except ValueError as exc:
+        raise _Halt(f"{member['number']} would not launch: {exc}")
+    _say(phase["file"], f"{phase['file']}: started {member['number']} — "
+                        f"{member['title']}")
+
+
+def _merge_member(phase: dict, member: dict) -> None:
+    branch = f"task/{member['file'][:-3]}"
+    phase_branch = agents.phase_branch(phase["file"])
+    if _branch_exists(branch):
+        _merge_into_phase(phase, branch, f"{member['number']}'s branch", member)
+        _push_phase(phase)
+    _record(phase, f"{member['number']} merged into {phase_branch}")
+    _say(phase["file"], f"{phase['file']}: merged {member['number']} into {phase_branch}")
+
+
+def _finish(phase: dict) -> None:
+    """Every member is in. Push the branch, move the card to review/ and
+    open the one PR this whole run exists to produce — from there the
+    ordinary review apparatus applies unchanged."""
+    _push_phase(phase)
+    _record(phase, "every card merged — opening the phase PR")
+    if phase["stage"] != "review":
+        taskfiles.move_task(phase["file"], phase["stage"], "review", actor="phase")
+    _say(phase["file"], f"{phase['file']}: every card merged — "
+                        f"{agents.phase_branch(phase['file'])} is ready for review")
+    github.maybe_open_pr(phase["file"])
+
+
+def _halt(phase: dict, member: dict | None, reason: str) -> None:
+    """Stop, and say so once. The log holds the halt from here on, so the
+    next pass reads it rather than saying the same thing again."""
+    at = f" at {member['number']}" if member and member["number"] else ""
+    _record(phase, f"halted{at} — {reason}")
+    _say(phase["file"], f"{phase['file']} halted{at} — {reason}")
+
+
+def _do_pass(phase: dict, snapshot: dict, by_number: dict[str, dict]) -> list[str]:
+    """The one next thing, whatever it is. Returns what the phase is
+    waiting on when a dependency holds it; raises `_Halt` when it stops."""
+    _freshen(phase)
+    for member in snapshot["members"]:
+        if member["state"] == "merged":
+            continue
+        if member["state"] in ("running", "waiting"):
+            return []                            # nothing to do but wait
+        if member["state"] == "halt":
+            raise _Halt(member["why"], member)
+        if member["state"] == "ready":
+            _merge_member(phase, member)
+            member["state"] = "merged"
+            phase = _reread(phase) or phase      # the log rewrote the card
+            continue
+        waiting_on = _unfinished_dependencies(member, snapshot, by_number)
+        if waiting_on:
+            return waiting_on                    # guarded, not skipped
+        _launch(phase, member)
+        return []
+    _finish(phase)
+    return []
+
+
+def advance(phase: dict, by_file: dict[str, dict],
+            by_number: dict[str, dict]) -> dict:
+    """One phase, one pass. Returns the snapshot the API then shows.
+
+    Nothing is carried between passes: the state is read at the top, acted
+    on once, and read back at the bottom.
+    """
+    snapshot = _snapshot(phase, by_file)
+    SNAPSHOTS[phase["file"]] = snapshot
+    if snapshot["halted"] or not snapshot["started"] or not _mine(phase):
+        return snapshot
+    waiting_on: list[str] = []
+    try:
+        waiting_on = _do_pass(phase, snapshot, by_number)
+    except _Halt as stop:
+        _halt(phase, stop.member, str(stop))
+    except (ValueError, OSError) as exc:
+        _halt(phase, None, f"the run could not continue: {exc}")
+    # The pass wrote to the card and to git, so what the API shows is read
+    # back rather than patched — the same recompute, once more.
+    before, fresh = snapshot, _reread(phase)
+    if fresh:
+        snapshot = _snapshot(fresh, _cards()[0])
+    if waiting_on:
+        snapshot["waitingOn"] = waiting_on
+    SNAPSHOTS[phase["file"]] = snapshot
+    if snapshot != before:
+        state.broadcast({"type": "board"})   # a pass that did nothing is quiet
+    return snapshot
+
+
+def _reread(phase: dict) -> dict | None:
+    """The card as it stands now — the runner writes to it as it goes, and
+    may have moved it. None once it has left the board entirely."""
+    return _cards()[0].get(phase["file"])
+
+
+def advance_all() -> dict:
+    """One pass over every phase card being run on this board."""
+    with _LOCK:
+        by_file, by_number = _cards()
+        live = {}
+        for task in by_file.values():
+            if task["isPhase"] and task["stage"] == "in-progress":
+                live[task["file"]] = advance(task, by_file, by_number)
+        for name in [n for n in SNAPSHOTS if n not in live]:
+            SNAPSHOTS.pop(name, None)
+        return live
+
+
+def beat() -> None:
+    """The runner: a plain thread, silent when no phase is running."""
+    while True:
+        time.sleep(config.PHASE_INTERVAL)
+        try:
+            advance_all()
+        except Exception:                # noqa: BLE001 — a beat never dies
+            pass
+
+
+# ── starting one ───────────────────────────────────────────────────────
+
+
+def start_phase(filename: str, stage: str, takeover: bool = False) -> dict:
+    """Run a phase — the first time, or again after a halt.
+
+    Cutting `phase/<stem>` from the newest main is the whole of "starting":
+    everything after it is the beat looking at what is there. Running a
+    halted phase again is the person's decision that cleared the halt, so
+    it appends the line that clears it and takes one pass immediately.
+    """
+    if Path(filename).name != filename or not filename.endswith(".md"):
+        raise ValueError("bad filename")
+    if stage != "in-progress":
+        raise ValueError("a phase runs from in-progress/ — move the card there first")
+    path = config.TASKS / stage / filename
+    if not path.is_file():
+        raise ValueError(f"{filename} is not in {stage}/ — refresh the board")
+
+    by_file, by_number = _cards()
+    phase = by_file.get(filename)
+    if phase is None or not phase["isPhase"]:
+        raise ValueError(f"{filename} is not a phase — a phase card is "
+                         f"**Type:** Phase with a ## Cards section")
+    if phase["phaseDrift"]:
+        raise ValueError(f"{filename} does not resolve: {phase['phaseDrift'][0]} "
+                         f"— fix the list before running it")
+    if not phase["members"]:
+        # Nothing to run, so nothing is started: no branch is cut and the
+        # card stays exactly where its author put it.
+        raise ValueError(f"{filename} lists no cards — nothing to run")
+
+    agents.claim_for_launch(filename, stage, takeover)
+    with _LOCK:                       # never alongside a pass of the beat
+        return _start(phase, filename)
+
+
+def _start(phase: dict, filename: str) -> dict:
+    """Cut the branch (once), record the run, and take the first pass."""
+    branch = agents.phase_branch(filename)
+    note = ""
+    if not _branch_exists(branch):
+        # The same rule and the same timeout a task branch is cut by: the
+        # newest main this checkout can see, and never a wait on the network.
+        point, note = agents._fresh_branch_point()
+        worktree = config.WORKTREES / filename[:-3]
+        config.WORKTREES.mkdir(parents=True, exist_ok=True)
+        result = (_git("worktree", "add", "--no-track", "-b", branch,
+                       str(worktree), point) if point
+                  else _git("worktree", "add", "-b", branch, str(worktree)))
+        if result.returncode != 0:
+            raise ValueError(f"could not cut {branch}: {result.stderr.strip()[:200]}")
+        _push_phase(phase)
+    _record(phase, f"run started on {branch}")
+    _say(filename, f"phase {filename} is running on {branch}"
+                   + (f" — {note}" if note else ""))
+    state.broadcast({"type": "board"})
+
+    by_file, by_number = _cards()         # the log just rewrote the card
+    return advance(by_file.get(filename) or phase, by_file, by_number)
diff --git a/tests/test_phase_runs.py b/tests/test_phase_runs.py
new file mode 100644
index 0000000..749a080
--- /dev/null
+++ b/tests/test_phase_runs.py
@@ -0,0 +1,891 @@
+"""A phase runs itself, on a branch of its own (task 49).
+
+The runner is a beat, not an agent: `phases.advance_all()` is one pass, and
+these cases drive it by hand rather than waiting on a thread. Everything
+under it is real — a real git repo, real worktrees, real headless launches
+through a real (stub) adapter — because what this card is about is the
+arithmetic of branches, and mocking git would be mocking the subject.
+
+    python3 -m unittest discover -s tests -v
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import shutil
+import stat
+import subprocess
+import sys
+import tempfile
+import time
+import unittest
+from pathlib import Path
+
+REPO = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(REPO / "manager" / "core"))
+
+import agents  # noqa: E402
+import config  # noqa: E402
+import github  # noqa: E402
+import phases  # noqa: E402
+import state  # noqa: E402
+import taskfiles  # noqa: E402
+
+PHASE = "40-ship-the-site.md"
+PHASE_BRANCH = "phase/40-ship-the-site"
+ONE = "31-stand-up-site.md"
+TWO = "32-serve-it.md"
+PR_URL = "https://github.com/acme/widget/pull/7"
+
+# An agent that does the job: one file, one commit, a clean exit.
+WORKS = """#!/usr/bin/env python3
+import os, subprocess
+cwd = os.environ["AGENT_CWD"]
+name = os.environ["BOARD_TASK"].split("-")[0]
+open(os.path.join(cwd, name + ".txt"), "w").write("card " + name + "\\n")
+subprocess.run(["git", "-C", cwd, "add", "-A"], check=True)
+subprocess.run(["git", "-C", cwd, "-c", "user.email=a@b", "-c", "user.name=stub",
+                "commit", "-q", "-m", "card " + name], check=True)
+print("WORK REPORT: card", name, "built")
+"""
+
+# The same, but it rewrites a file main also owns — the collision a phase
+# branch has to refuse rather than guess at.
+COLLIDES = """#!/usr/bin/env python3
+import os, subprocess
+cwd = os.environ["AGENT_CWD"]
+open(os.path.join(cwd, "code.txt"), "w").write("the member's line\\n")
+subprocess.run(["git", "-C", cwd, "add", "-A"], check=True)
+subprocess.run(["git", "-C", cwd, "-c", "user.email=a@b", "-c", "user.name=stub",
+                "commit", "-q", "-m", "collide"], check=True)
+print("WORK REPORT: done")
+"""
+
+DECLINES = """#!/usr/bin/env python3
+print("NOT READY: the card does not say which store to write to")
+"""
+
+DIES = """#!/usr/bin/env python3
+import sys
+print("API Error: 500 overloaded")
+sys.exit(1)
+"""
+
+COMMITS_NOTHING = """#!/usr/bin/env python3
+print("WORK REPORT: I read a lot and wrote nothing")
+"""
+
+FAKE_GH = f'''#!/usr/bin/env python3
+import json, os, sys
+args = sys.argv[1:]
+with open(os.environ["GH_LOG"], "a") as fh:
+    fh.write(json.dumps(args) + "\\n")
+if args[:2] == ["pr", "create"]:
+    print("{PR_URL}")
+elif args[:2] == ["pr", "view"] and "--jq" in args:
+    print("{PR_URL}")
+elif args[:2] == ["pr", "view"]:
+    print(json.dumps({{"reviews": [], "reviewRequests": [],
+                      "statusCheckRollup": [], "state": "OPEN",
+                      "mergeable": "MERGEABLE"}}))
+else:
+    sys.exit(0)
+'''
+
+
+def git(cwd: Path, *args: str) -> subprocess.CompletedProcess:
+    return subprocess.run(["git", "-C", str(cwd), *args],
+                          capture_output=True, text=True)
+
+
+def wait_for(pred, timeout: float = 30.0) -> bool:
+    deadline = time.time() + timeout
+    while time.time() < deadline:
+        if pred():
+            return True
+        time.sleep(0.05)
+    return False
+
+
+def card(title: str, *, status: str = "Backlog", kind: str | None = None,
+         depends: str | None = None, cards: str | None = None) -> str:
+    text = f"# {title}\n\n**Status:** {status}\n**Priority:** High\n"
+    if kind:
+        text += f"**Type:** {kind}\n"
+    if depends:
+        text += f"**Depends on:** {depends}\n"
+    text += "\nWhat this card is for, at enough length to be a brief.\n"
+    if cards is not None:
+        text += f"\n## Cards\n\n{cards}\n"
+    return text
+
+
+class PhaseCase(unittest.TestCase):
+    """One repo, one phase card, two members — and an adapter whose
+    behaviour each test writes."""
+
+    REMOTE = False
+
+    def setUp(self):
+        self.tmp = Path(tempfile.mkdtemp(prefix="bench-phase-run-")).resolve()
+        self.addCleanup(shutil.rmtree, self.tmp, True)
+        self.repo = self.tmp / "repo"
+
+        if self.REMOTE:
+            self.origin = self.tmp / "origin.git"
+            subprocess.run(["git", "init", "-q", "--bare", "-b", "main",
+                            str(self.origin)], check=True, capture_output=True)
+            subprocess.run(["git", "clone", "-q", str(self.origin), str(self.repo)],
+                           check=True, capture_output=True)
+        else:
+            self.repo.mkdir()
+            git(self.repo, "init", "-q", "-b", "main")
+        git(self.repo, "config", "user.email", "t@t")
+        git(self.repo, "config", "user.name", "tester")
+        (self.repo / "code.txt").write_text("shipped\n", encoding="utf-8")
+
+        self.tasks = self.repo / "tasks"
+        for slug in config.STAGE_DIRS:
+            (self.tasks / slug).mkdir(parents=True)
+        git(self.repo, "add", "-A")
+        git(self.repo, "commit", "-q", "-m", "root")
+        if self.REMOTE:
+            git(self.repo, "push", "-q", "origin", "main")
+
+        local = self.tmp / "local"
+        (local / "adapters" / config.ADAPTER).mkdir(parents=True)
+        self.adapter = local / "adapters" / config.ADAPTER / "run"
+        self.adapter_is(WORKS)
+
+        gh = self.tmp / "gh"
+        gh.write_text(FAKE_GH, encoding="utf-8")
+        gh.chmod(gh.stat().st_mode | stat.S_IEXEC)
+        self.gh_log = self.tmp / "gh.log"
+        os.environ["GH_LOG"] = str(self.gh_log)
+        self.addCleanup(os.environ.pop, "GH_LOG", None)
+
+        self.patch(REPO=self.repo, TASKS=self.tasks, TM_ROOT=self.repo,
+                   LOCAL=local, WORKTREES=self.tmp / "worktrees",
+                   AGENT_DIR=self.tmp / "agent", SESSIONS_DIR=self.tmp / "sessions",
+                   GH_BIN=str(gh), COMMIT_MOVES=False, SYNC=False,
+                   FETCH_TIMEOUT=10.0)
+
+        for registry in (state.AGENTS, state.BOARD_EVENTS, state.EXPECTED_MOVES,
+                         github.PR_STATE, phases.SNAPSHOTS):
+            registry.clear()
+            self.addCleanup(registry.clear)
+
+        self.sent: list[dict] = []
+        self.addCleanup(setattr, state, "broadcast", state.broadcast)
+        state.broadcast = self.sent.append
+
+        self.write_cards()
+
+    def write_cards(self, *, depends: str | None = None,
+                    listed: str = "- 31 — Stand up site/\n- 32 — Serve it\n") -> None:
+        self.write(ONE, card("31 — Stand up site/"))
+        self.write(TWO, card("32 — Serve it", depends=depends))
+        self.write(PHASE, card("40 — Ship the site", status="In Progress",
+                               kind="Phase", cards=listed), "in-progress")
+
+    def write(self, filename: str, text: str, stage: str = "backlog") -> None:
+        (self.tasks / stage / filename).write_text(text, encoding="utf-8")
+
+    def patch(self, **values) -> None:
+        for attr, value in values.items():
+            self.addCleanup(setattr, config, attr, getattr(config, attr))
+            setattr(config, attr, value)
+
+    def adapter_is(self, script: str) -> None:
+        self.adapter.write_text(script, encoding="utf-8")
+        self.adapter.chmod(self.adapter.stat().st_mode | stat.S_IEXEC)
+
+    # — driving the beat —
+
+    def start(self) -> dict:
+        snapshot = phases.start_phase(PHASE, "in-progress")
+        self.settle()
+        return snapshot
+
+    def advance(self) -> dict:
+        snapshots = phases.advance_all()
+        self.settle()
+        return snapshots.get(PHASE, {})
+
+    def settle(self) -> None:
+        """Wait for any agent the pass launched to be fully reaped — the
+        reaper's last act is the agents broadcast, so that is the honest
+        signal that the card has landed wherever it lands."""
+        self.assertTrue(
+            wait_for(lambda: not any(r["status"] == "running"
+                                     for r in state.AGENTS.values())),
+            "an agent the phase launched never ended")
+        running = [r for r in state.AGENTS.values() if r.get("mode") == "work"]
+        if running:
+            self.assertTrue(wait_for(
+                lambda: self.sent.count({"type": "agents"}) >= len(running)),
+                "the reaper never announced the ending")
+
+    # — reading the world back —
+
+    def stage_of(self, filename: str) -> str | None:
+        for slug in config.STAGE_DIRS:
+            if (self.tasks / slug / filename).is_file():
+                return slug
+        return None
+
+    def text(self, filename: str) -> str:
+        return (self.tasks / self.stage_of(filename) / filename).read_text(
+            encoding="utf-8")
+
+    def log(self) -> list[str]:
+        return phases.log_entries(self.text(PHASE))
+
+    def branch_exists(self, branch: str) -> bool:
+        return git(self.repo, "rev-parse", "--verify", "--quiet",
+                   branch).returncode == 0
+
+    def tip(self, ref: str) -> str:
+        return git(self.repo, "rev-parse", ref).stdout.strip()
+
+    def merges_on(self, branch: str) -> list[str]:
+        return git(self.repo, "log", "--merges", "--format=%s",
+                   branch).stdout.strip().splitlines()
+
+    def summaries(self) -> list[str]:
+        return [e["summary"] for e in state.BOARD_EVENTS]
+
+    def gh_calls(self) -> list[list[str]]:
+        if not self.gh_log.is_file():
+            return []
+        return [json.loads(line) for line in
+                self.gh_log.read_text(encoding="utf-8").splitlines() if line.strip()]
+
+
+class StartingAPhase(PhaseCase):
+    def test_the_branch_is_cut_from_main_and_member_one_starts_on_it(self):
+        main = self.tip("main")
+
+        self.start()
+
+        self.assertTrue(self.branch_exists(PHASE_BRANCH))
+        self.assertEqual(git(self.repo, "merge-base", PHASE_BRANCH,
+                             "main").stdout.strip(), main,
+                         "the phase branch starts at the newest main")
+        self.assertEqual(self.stage_of(ONE), "review",
+                         "member one ran and reached review/")
+        self.assertEqual(git(self.repo, "merge-base", f"task/{ONE[:-3]}",
+                             PHASE_BRANCH).stdout.strip(), main,
+                         "member one branched from the phase's tip")
+
+    def test_the_phase_gets_a_worktree_of_its_own(self):
+        self.start()
+        worktree = config.WORKTREES / PHASE[:-3]
+        self.assertTrue(worktree.is_dir())
+        self.assertEqual(git(worktree, "branch", "--show-current").stdout.strip(),
+                         PHASE_BRANCH)
+
+    def test_the_run_is_recorded_on_the_card(self):
+        self.start()
+        self.assertIn(f"run started on {PHASE_BRANCH}", self.log())
+        self.assertIn("31 started", self.log())
+        self.assertIn("## Phase log", self.text(PHASE))
+
+    def test_only_the_first_member_is_launched(self):
+        self.start()
+        self.assertFalse(self.branch_exists(f"task/{TWO[:-3]}"),
+                         "sequential: member two waits for member one")
+        self.assertEqual(self.stage_of(TWO), "backlog")
+
+    def test_an_empty_phase_refuses_rather_than_leaving_a_branch(self):
+        self.write(PHASE, card("40 — Ship the site", status="In Progress",
+                               kind="Phase", cards=""), "in-progress")
+
+        with self.assertRaises(ValueError) as caught:
+            phases.start_phase(PHASE, "in-progress")
+
+        self.assertIn("lists no cards", str(caught.exception))
+        self.assertFalse(self.branch_exists(PHASE_BRANCH))
+        self.assertFalse((config.WORKTREES / PHASE[:-3]).exists())
+
+    def test_a_list_that_does_not_resolve_refuses(self):
+        """Drift is an authoring mistake; a runner must not act it out."""
+        self.write_cards(listed="- 31 — Stand up site/\n- 99 — nothing here\n")
+
+        with self.assertRaises(ValueError) as caught:
+            phases.start_phase(PHASE, "in-progress")
+
+        self.assertIn("99", str(caught.exception))
+        self.assertFalse(self.branch_exists(PHASE_BRANCH))
+
+    def test_an_ordinary_card_is_not_a_phase(self):
+        self.write(PHASE, card("40 — Just a card", status="In Progress"),
+                   "in-progress")
+
+        with self.assertRaises(ValueError) as caught:
+            phases.start_phase(PHASE, "in-progress")
+
+        self.assertIn("not a phase", str(caught.exception))
+
+    def test_a_phase_runs_from_in_progress_only(self):
+        self.write(PHASE, card("40 — Ship the site", kind="Phase",
+                               cards="- 31 — Stand up site/\n"), "to-do")
+
+        with self.assertRaises(ValueError) as caught:
+            phases.start_phase(PHASE, "to-do")
+
+        self.assertIn("in-progress", str(caught.exception))
+
+
+class AdvancingOnGreen(PhaseCase):
+    def test_member_twos_worktree_holds_member_ones_work(self):
+        """The whole point of the branch: card two can see card one."""
+        self.start()
+
+        self.advance()
+
+        self.assertTrue(self.branch_exists(f"task/{TWO[:-3]}"))
+        worktree = config.WORKTREES / TWO[:-3]
+        self.assertTrue((worktree / "31.txt").is_file(),
+                        "member two started from a tip that already had 31")
+        self.assertTrue((worktree / "32.txt").is_file())
+
+    def test_the_merge_lands_on_the_phase_branch_and_is_recorded(self):
+        self.start()
+
+        self.advance()
+
+        self.assertEqual(
+            git(self.repo, "merge-base", "--is-ancestor",
+                f"task/{ONE[:-3]}", PHASE_BRANCH).returncode, 0)
+        self.assertEqual(self.merges_on(PHASE_BRANCH),
+                         [f"phase: merge task/{ONE[:-3]} into {PHASE_BRANCH}"])
+        self.assertIn(f"31 merged into {PHASE_BRANCH}", self.log())
+        self.assertTrue(any("merged 31" in s for s in self.summaries()))
+
+    def test_the_phase_never_rebases_or_moves_a_members_branch(self):
+        self.start()
+        before = self.tip(f"task/{ONE[:-3]}")
+
+        self.advance()
+
+        self.assertEqual(self.tip(f"task/{ONE[:-3]}"), before,
+                         "merging is additive: the member's branch is untouched")
+
+    def test_a_finished_phase_opens_one_pr_and_moves_to_review(self):
+        self.start()
+        self.advance()          # merge 31, launch 32
+        self.advance()          # merge 32, finish
+
+        self.assertEqual(self.stage_of(PHASE), "review")
+        self.assertEqual(len(self.merges_on(PHASE_BRANCH)), 2)
+        self.assertIn("every card merged — opening the phase PR", self.log())
+
+    def test_the_run_is_over_and_the_beat_leaves_it_alone(self):
+        self.start()
+        self.advance()
+        self.advance()
+        state.BOARD_EVENTS.clear()
+
+        phases.advance_all()
+
+        self.assertEqual(self.summaries(), [],
+                         "a phase card out of in-progress/ is not run any more")
+
+
+class DependenciesGuard(PhaseCase):
+    def test_a_member_waiting_on_an_unfinished_card_is_not_launched(self):
+        self.write_cards(depends="99")
+        self.write("99-something-else.md", card("99 — Something else"))
+        self.start()
+
+        snapshot = self.advance()
+
+        self.assertEqual(snapshot["waitingOn"], ["99"])
+        self.assertFalse(self.branch_exists(f"task/{TWO[:-3]}"))
+        self.assertEqual(self.stage_of(TWO), "backlog")
+        self.assertIsNone(snapshot["halted"], "a guard is a wait, not a halt")
+
+    def test_the_same_member_launches_once_the_dependency_is_done(self):
+        self.write_cards(depends="99")
+        self.write("99-something-else.md",
+                   card("99 — Something else", status="Done"), "done")
+        self.start()
+
+        self.advance()
+
+        self.assertTrue(self.branch_exists(f"task/{TWO[:-3]}"))
+
+    def test_a_padded_number_is_the_same_card_everywhere(self):
+        """`07`, `7` and `#007` are one card — in the list, in the log and
+        in a dependency alike, or none of them match."""
+        self.write("07-early.md", card("07 — An early card"))
+        self.write(TWO, card("32 — Serve it", depends="07"))
+        self.write(PHASE, card("40 — Ship the site", status="In Progress",
+                               kind="Phase",
+                               cards="- 07 — An early card\n- 32 — Serve it\n"),
+                   "in-progress")
+
+        self.start()
+        self.advance()
+
+        self.assertIn("7 started", self.log())
+        self.assertTrue(self.branch_exists("task/07-early"))
+        self.assertTrue(self.branch_exists(f"task/{TWO[:-3]}"),
+                        "the dependency on 07 resolved against member 7")
+
+    def test_a_dependency_inside_the_phase_counts_as_finished_when_merged(self):
+        self.write_cards(depends="31")
+        self.start()
+
+        self.advance()
+
+        self.assertTrue(self.branch_exists(f"task/{TWO[:-3]}"),
+                        "31 is merged into the phase branch — that is finished")
+
+
+class HaltNeverSkip(PhaseCase):
+    def halted(self) -> str:
+        return phases._halt_reason(self.log()) or ""
+
+    def test_a_member_that_declines_halts_the_phase(self):
+        self.adapter_is(DECLINES)
+        self.start()
+
+        self.assertEqual(self.stage_of(ONE), "to-do", "the card walks back")
+        snapshot = self.advance()
+
+        self.assertIn("31:", self.halted())
+        self.assertIn("declined", self.halted())
+        self.assertIsNotNone(snapshot["halted"])
+        self.assertFalse(self.branch_exists(f"task/{TWO[:-3]}"),
+                         "no further member starts")
+
+    def test_a_run_that_dies_halts_the_phase(self):
+        self.adapter_is(DIES)
+        self.start()
+
+        self.advance()
+
+        self.assertEqual(self.stage_of(ONE), "in-progress")
+        self.assertIn("without reaching review/", self.halted())
+        self.assertIn("API Error: 500", self.halted())
+        self.assertFalse(self.branch_exists(f"task/{TWO[:-3]}"))
+
+    def test_a_clean_run_that_committed_nothing_halts_the_phase(self):
+        self.adapter_is(COMMITS_NOTHING)
+        self.start()
+
+        self.advance()
+
+        self.assertEqual(self.stage_of(ONE), "in-progress")
+        self.assertIn("without reaching review/", self.halted())
+        self.assertFalse(self.branch_exists(f"task/{TWO[:-3]}"))
+
+    def test_red_ci_halts_the_phase(self):
+        self.start()
+        github.PR_STATE[ONE] = {"verdict": "red", "ci": "fail", "url": PR_URL}
+
+        self.advance()
+
+        self.assertIn("CI is red", self.halted())
+        self.assertEqual(self.merges_on(PHASE_BRANCH), [],
+                         "nothing is merged on a red member")
+
+    def test_checks_still_running_is_a_wait_not_a_halt(self):
+        self.start()
+        github.PR_STATE[ONE] = {"verdict": "pending", "ci": "running", "url": PR_URL}
+
+        snapshot = self.advance()
+
+        self.assertIsNone(snapshot["halted"])
+        self.assertEqual(self.merges_on(PHASE_BRANCH), [])
+
+    def test_a_halt_is_said_once_and_then_held(self):
+        self.adapter_is(DIES)
+        self.start()
+        self.advance()
+        state.BOARD_EVENTS.clear()
+        lines = len(self.log())
+
+        self.advance()
+
+        self.assertEqual(self.summaries(), [], "a halted phase is quiet")
+        self.assertEqual(len(self.log()), lines, "and writes nothing more")
+
+    def test_running_it_again_clears_the_halt_and_carries_on(self):
+        self.adapter_is(DIES)
+        self.start()
+        self.advance()
+        self.assertTrue(self.halted())
+        self.adapter_is(WORKS)
+
+        phases.start_phase(PHASE, "in-progress")
+        self.settle()
+
+        self.assertEqual(self.halted(), "",
+                         "the person's decision to run it again clears the halt")
+        self.assertEqual(self.stage_of(ONE), "review")
+
+
+class ConflictsAreRefused(PhaseCase):
+    """The third place in this codebase that merges branches: it aborts
+    cleanly, leaves no half-merged branch, and names the collision."""
+
+    def test_a_member_that_collides_with_main_halts_and_leaves_no_mess(self):
+        self.adapter_is(COLLIDES)
+        self.start()
+        (self.repo / "code.txt").write_text("main moved on\n", encoding="utf-8")
+        git(self.repo, "commit", "-q", "-am", "main moved on")
+
+        self.advance()
+
+        self.assertIn("conflicts", phases._halt_reason(self.log()) or "")
+        self.assertIn("code.txt", phases._halt_reason(self.log()) or "")
+        worktree = config.WORKTREES / PHASE[:-3]
+        merge_head = git(worktree, "rev-parse", "--git-path", "MERGE_HEAD").stdout.strip()
+        self.assertFalse((worktree / merge_head).is_file() or
+                         Path(merge_head).is_file(),
+                         "no merge is left in progress")
+        self.assertEqual(git(worktree, "status", "--porcelain").stdout, "",
+                         "the phase worktree is left clean")
+        self.assertEqual(
+            git(self.repo, "merge-base", "--is-ancestor",
+                f"task/{ONE[:-3]}", PHASE_BRANCH).returncode, 1,
+            "nothing half-merged: the member's branch did not land")
+
+
+class TheBranchIsKeptFresh(PhaseCase):
+    def test_main_is_merged_into_the_phase_branch_on_the_beat(self):
+        self.start()
+        (self.repo / "later.txt").write_text("landed on main\n", encoding="utf-8")
+        git(self.repo, "add", "-A")
+        git(self.repo, "commit", "-q", "-m", "landed on main")
+
+        self.advance()
+
+        self.assertEqual(
+            git(self.repo, "merge-base", "--is-ancestor", "main",
+                PHASE_BRANCH).returncode, 0,
+            "a phase that runs for hours does not drift")
+        self.assertTrue(any("merged main into" in s for s in self.summaries()))
+
+    def test_an_unchanged_main_is_merged_nowhere(self):
+        self.start()
+        state.BOARD_EVENTS.clear()
+
+        self.advance()
+
+        self.assertEqual([s for s in self.summaries() if "merged main" in s], [])
+
+
+class ARestartResumesByLooking(PhaseCase):
+    """The runner holds no registry: git and the card carry the memory."""
+
+    def restart(self) -> None:
+        """What a board restart costs the runner: everything it held in
+        memory about running agents and past passes."""
+        state.AGENTS.clear()
+        phases.SNAPSHOTS.clear()
+
+    def test_a_merged_member_is_not_merged_twice(self):
+        self.start()
+        self.advance()          # merge 31, launch 32
+        self.restart()
+
+        self.advance()
+
+        self.assertEqual(len([m for m in self.merges_on(PHASE_BRANCH)
+                              if ONE[:-3] in m]), 1)
+        self.assertEqual(len([e for e in self.log() if e.startswith("31 merged")]), 1)
+
+    def test_a_started_member_is_not_started_twice(self):
+        """The log is what tells "already run, and it ended badly" from
+        "not reached yet" — without it a restart would relaunch a dead run."""
+        self.adapter_is(DECLINES)
+        self.start()
+        self.restart()
+
+        self.advance()
+
+        self.assertEqual(len([e for e in self.log() if e == "31 started"]), 1)
+        self.assertIn("31:", phases._halt_reason(self.log()) or "")
+
+    def test_a_run_lost_to_a_restart_halts_rather_than_relaunching(self):
+        self.start()
+        # put the card back where a run that never finished would leave it
+        taskfiles.move_task(ONE, "review", "in-progress", actor="tester")
+        self.restart()
+
+        self.advance()
+
+        self.assertIn("without reaching review/", phases._halt_reason(self.log()) or "")
+
+
+class HandMovesDoNotDoubleLaunch(PhaseCase):
+    def test_a_card_moved_by_hand_under_a_running_agent_is_left_alone(self):
+        self.start()
+        state.AGENTS.clear()
+        state.AGENTS["live"] = {"id": "live", "task": TWO, "status": "running",
+                                "started": time.time(), "mode": "work"}
+        taskfiles.move_task(TWO, "backlog", "review", actor="you")
+        state.BOARD_EVENTS.clear()
+
+        snapshot = phases.advance_all()[PHASE]
+
+        self.assertEqual([m["state"] for m in snapshot["members"] if m["file"] == TWO],
+                         ["running"], "an agent is on it — the phase waits")
+        self.assertEqual(len(state.AGENTS), 1, "no second launch on the same task")
+        self.assertFalse(self.branch_exists(f"task/{TWO[:-3]}"))
+
+    def test_a_member_finished_before_the_phase_reached_it_needs_no_run(self):
+        self.write(TWO, card("32 — Serve it", status="Done"), "done")
+        (self.tasks / "backlog" / TWO).unlink()
+        self.start()
+
+        self.advance()
+
+        self.assertFalse(self.branch_exists(f"task/{TWO[:-3]}"),
+                         "a card with no branch that is already done is finished")
+        self.assertEqual(self.stage_of(PHASE), "review")
+
+
+class OneBoardRunsIt(PhaseCase):
+    """State syncs; reactions don't — the assignee is where "who runs it"
+    is written down, and a replica advances nothing."""
+
+    def setUp(self):
+        super().setUp()
+        self.patch(COMMIT_MOVES=True)
+
+    def test_the_board_that_holds_the_card_runs_it(self):
+        self.start()
+
+        self.assertIn("**Assignee:** tester", self.text(PHASE))
+        self.assertEqual(self.stage_of(ONE), "review")
+
+    def test_someone_elses_phase_advances_nowhere(self):
+        self.write(PHASE, card("40 — Ship the site", status="In Progress",
+                               kind="Phase",
+                               cards="- 31 — Stand up site/\n- 32 — Serve it\n"
+                               ).replace("**Priority:** High",
+                                         "**Priority:** High\n**Assignee:** elena"),
+                   "in-progress")
+        git(self.repo, "checkout", "-q", "-b", "phase/40-ship-the-site")
+        git(self.repo, "checkout", "-q", "main")
+        state.BOARD_EVENTS.clear()
+
+        snapshot = phases.advance_all()[PHASE]
+
+        self.assertEqual(len(state.AGENTS), 0, "a replica launches nothing")
+        self.assertEqual(self.summaries(), [])
+        self.assertEqual([m["state"] for m in snapshot["members"]],
+                         ["pending", "pending"], "but it still renders the phase")
+
+    def test_starting_someone_elses_phase_refuses_and_names_them(self):
+        self.write(PHASE, card("40 — Ship the site", status="In Progress",
+                               kind="Phase",
+                               cards="- 31 — Stand up site/\n"
+                               ).replace("**Priority:** High",
+                                         "**Priority:** High\n**Assignee:** elena"),
+                   "in-progress")
+
+        with self.assertRaises(ValueError) as caught:
+            phases.start_phase(PHASE, "in-progress")
+
+        self.assertIn("elena holds", str(caught.exception))
+        self.assertFalse(self.branch_exists(PHASE_BRANCH))
+
+
+class ThePhasePR(PhaseCase):
+    """From review/ onwards a phase card is an ordinary card — which means
+    its branch has to be the one the apparatus finds."""
+
+    REMOTE = True
+
+    def test_the_pr_goes_from_the_phase_branch_into_main(self):
+        self.start()
+        self.advance()
+        self.advance()
+
+        create = [c for c in self.gh_calls() if c[:2] == ["pr", "create"]]
+        self.assertEqual(len(create), 1, "one PR at the end, not one per card")
+        self.assertIn("--head", create[0])
+        self.assertEqual(create[0][create[0].index("--head") + 1], PHASE_BRANCH)
+        self.assertEqual(create[0][create[0].index("--base") + 1], "main")
+        self.assertIn(f"**PR:** {PR_URL}", self.text(PHASE))
+        self.assertEqual(self.stage_of(PHASE), "review")
+
+    def test_the_body_is_the_member_list(self):
+        self.start()
+        self.advance()
+        self.advance()
+
+        create = [c for c in self.gh_calls() if c[:2] == ["pr", "create"]][0]
+        body = create[create.index("--body") + 1]
+        self.assertIn("31 — 31 — Stand up site/", body)
+        self.assertIn("32 — 32 — Serve it", body)
+
+    def test_a_members_own_pr_is_based_on_the_phase_branch(self):
+        """A member's branch was cut from the phase's, so main is not a base
+        whose diff means anything — and a PR into main would invite exactly
+        the merge this whole design refuses to make."""
+        self.start()
+
+        github.maybe_open_pr(ONE)
+
+        create = [c for c in self.gh_calls() if c[:2] == ["pr", "create"]]
+        self.assertEqual(create[0][create[0].index("--base") + 1], PHASE_BRANCH)
+        self.assertEqual(create[0][create[0].index("--head") + 1], f"task/{ONE[:-3]}")
+
+    def test_the_phase_branch_is_published_so_the_base_exists(self):
+        self.start()
+        self.assertEqual(
+            git(self.repo, "rev-parse", f"origin/{PHASE_BRANCH}").returncode, 0,
+            "a member's PR needs its base on the remote")
+
+
+class TheCardsBranchIsFound(PhaseCase):
+    def test_a_phase_card_wears_its_own_branch(self):
+        self.start()
+        self.assertEqual(github.branch_of(PHASE), PHASE_BRANCH)
+        self.assertEqual(github.branch_of(ONE), f"task/{ONE[:-3]}")
+        self.assertIn(PHASE[:-3], github.task_branches())
+        self.assertIn(ONE[:-3], github.task_branches())
+
+    def test_a_card_with_no_branch_at_all_names_the_task_branch(self):
+        self.assertEqual(github.branch_of(PHASE), f"task/{PHASE[:-3]}")
+
+
+class TheBranchPointOfAMember(PhaseCase):
+    def test_a_member_of_a_running_phase_branches_from_the_phase(self):
+        self.start()
+        point, note = agents.phase_branch_point(TWO)
+        self.assertEqual(point, PHASE_BRANCH)
+        self.assertIn("the phase's own branch", note)
+
+    def test_a_member_of_a_phase_that_has_not_started_branches_as_usual(self):
+        self.assertEqual(agents.phase_branch_point(TWO), (None, None))
+
+    def test_a_card_in_no_phase_branches_as_usual(self):
+        self.write("77-alone.md", card("77 — On its own"))
+        self.start()
+        self.assertEqual(agents.phase_branch_point("77-alone.md"), (None, None))
+
+    def test_the_ticker_names_the_branch_point(self):
+        self.start()
+        self.advance()
+        self.assertTrue(any("the phase's own branch" in s for s in self.summaries()),
+                        "an unusual branch point is always narrated")
+
+
+class ThePhaseLog(unittest.TestCase):
+    """The record on the card, read back — the only thing that can tell an
+    already-run member from one the phase has not reached."""
+
+    def entries(self, *lines: str) -> list[str]:
+        body = "\n".join(f"- 2026-08-01 09:0{i} · {line}"
+                         for i, line in enumerate(lines))
+        return phases.log_entries(f"# 40\n\n## Phase log\n\n{body}\n")
+
+    def test_entries_are_read_in_order(self):
+        self.assertEqual(self.entries("run started on phase/40-x", "31 started"),
+                         ["run started on phase/40-x", "31 started"])
+
+    def test_a_card_with_no_log_has_no_entries(self):
+        self.assertEqual(phases.log_entries("# 40\n\n**Status:** In Progress\n"), [])
+
+    def test_the_section_ends_where_the_next_heading_begins(self):
+        text = ("# 40\n\n## Phase log\n\n- 2026-08-01 09:00 · 31 started\n\n"
+                "## Work report\n\n- 2026-08-01 09:01 · 32 started\n")
+        self.assertEqual(phases.log_entries(text), ["31 started"])
+
+    def test_started_members_are_the_ones_it_launched(self):
+        entries = self.entries("run started on phase/40-x", "31 started",
+                               "31 merged into phase/40-x", "32 started")
+        self.assertEqual(phases._started(entries), {"31", "32"})
+
+    def test_a_halt_holds_until_a_run_clears_it(self):
+        entries = self.entries("31 started", "halted at 31 — its CI is red")
+        self.assertEqual(phases._halt_reason(entries), "31: its CI is red")
+
+    def test_a_later_run_clears_the_halt(self):
+        entries = self.entries("halted at 31 — its CI is red",
+                               "run started on phase/40-x")
+        self.assertIsNone(phases._halt_reason(entries))
+
+    def test_a_phase_that_never_halted_is_not_halted(self):
+        self.assertIsNone(phases._halt_reason(
+            self.entries("run started on phase/40-x", "31 started")))
+
+
+class AppendingToASection(unittest.TestCase):
+    """taskfiles' third door: a running record that stays in one place."""
+
+    def setUp(self):
+        tmp = Path(tempfile.mkdtemp(prefix="bench-section-")).resolve()
+        self.addCleanup(shutil.rmtree, tmp, True)
+        self.tasks = tmp / "tasks"
+        (self.tasks / "in-progress").mkdir(parents=True)
+        self.addCleanup(setattr, config, "TASKS", config.TASKS)
+        config.TASKS = self.tasks
+        self.addCleanup(setattr, config, "COMMIT_MOVES", config.COMMIT_MOVES)
+        config.COMMIT_MOVES = False
+        self.path = self.tasks / "in-progress" / PHASE
+        self.path.write_text(card("40 — Ship the site", status="In Progress",
+                                  kind="Phase", cards="- 31 — one\n"),
+                             encoding="utf-8")
+
+    def append(self, line: str) -> None:
+        taskfiles.append_to_section(PHASE, "in-progress", "Phase log", line, "log")
+
+    def test_the_first_line_creates_the_section(self):
+        self.append("- one")
+        self.assertIn("## Phase log\n\n- one\n", self.path.read_text())
+
+    def test_later_lines_join_it(self):
+        self.append("- one")
+        self.append("- two")
+        text = self.path.read_text()
+        self.assertIn("## Phase log\n\n- one\n- two\n", text)
+        self.assertEqual(text.count("## Phase log"), 1)
+
+    def test_a_section_appended_after_it_does_not_break_the_record(self):
+        self.append("- one")
+        taskfiles.append_to_task(PHASE, "in-progress",
+                                 "\n\n---\n\n## Work report\n\nsomething\n", "report")
+        self.append("- two")
+        text = self.path.read_text()
+        self.assertLess(text.index("- two"), text.index("## Work report"),
+                        "the log stays in one place a person can read")
+        self.assertEqual(phases.log_entries(text), [])   # not stamped lines
+
+    def test_the_cards_section_is_left_intact(self):
+        self.append("- one")
+        task = taskfiles.read_task(self.path, "in-progress")
+        self.assertEqual(task["cards"], ["31"])
+        self.assertEqual(task["phaseDrift"], [])
+
+
+class TheDocumentedPromise(unittest.TestCase):
+    """AGENTS.md quotes "the board never merges" as an absolute. After this
+    card it has to say what it always meant."""
+
+    @classmethod
+    def setUpClass(cls):
+        cls.doc = (REPO / "AGENTS.md").read_text(encoding="utf-8")
+
+    def test_the_promise_is_about_main(self):
+        self.assertIn("the board never merges into `main`", self.doc)
+        self.assertNotIn("Merging remains yours — the board never merges.", self.doc)
+
+    def test_the_module_map_names_the_runner(self):
+        self.assertIn("phases.py", self.doc)
+
+    def test_the_runner_is_described(self):
+        self.assertIn("phase/<task-stem>", self.doc)
+
+
+if __name__ == "__main__":
+    unittest.main()