From da8984d0e64333207ebedff2abb36d9cc5d71330 Mon Sep 17 00:00:00 2001
From: istos <ronald@istos.it>
Date: Sat, 1 Aug 2026 09:43:57 +0200
Subject: [PATCH] A phase runs itself, on a branch of its own
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

Starting a phase cuts phase/<stem> from the newest main it can see and
works the list into it: each member branched from the phase's tip, run
headless, merged back when its checks are green, the next one started.
At the end one PR into main, for a human. The human gate moves from
every card to the phase boundary, and the promise survives: the board
merges into a branch it created, inside a scope you opened.

The runner is a beat, not an agent — everything it decides is already
structured state, and an agent paid to poll would be the wrong tool at
the wrong price. It holds no registry of where a phase is. Two durable
things carry the memory, and the board already writes both: git, where
a member is finished when its branch is contained in the phase branch,
and the card, which grows a ## Phase log the runner adds one line to
per decision. The log is what tells "this member has run and it ended
badly" from "the phase has not reached it yet" — without it a
restarted board would relaunch a run that died.

Containment alone is not enough to call a member merged: a clean exit
that committed nothing leaves an empty branch that is contained. The
card has to have settled into review/ too, or a broken launch would
hide exactly where it always tries to.

Five conditions halt, each already a visible state on the card, and a
halt is written once and then held. Running the phase again is the
person's decision and is what clears it — the run is scoped to its own
log line, so a member whose run died is launchable again. A dependency
that has not landed is a wait, not a halt.

Merges are additive throughout: main into the phase branch on every
beat so a long run does not drift into one enormous conflict, members
into it as they go green, nothing rebased and nothing force-pushed. A
conflict aborts, leaves the branch as it was, and halts naming the
files that collided.

The actor rule decides who runs it, written where it already lives:
the phase card's assignee. A replica renders the phase and advances
nothing. Reachable through /api/phase/run and the ticker; the header
chip and the card actions are a separate card.

Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
---
 manager/core/.env.example |   8 +
 manager/core/board.py     |   3 +
 manager/core/config.py    |   5 +
 manager/core/httpd.py     |  11 +
 manager/core/phases.py    | 579 +++++++++++++++++++++++++
 tests/test_phase_runs.py  | 891 ++++++++++++++++++++++++++++++++++++++
 6 files changed, 1497 insertions(+)
 create mode 100644 manager/core/phases.py
 create mode 100644 tests/test_phase_runs.py

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()