Merge pull request #41 from 12vectors/task/49-a-phase-runs-itself-on-its-own-branch

49 — A phase runs itself, on a branch of its own

AGENTS.md

@@ -44,7 +44,7 @@ Drivers know apps, adapters know vendors, `local/` knows this project.
 Module map for `manager/core/` (dependencies flow strictly left to right):
 
 ```
-config → state / reports → taskfiles → events / github / drive / sync → agents → watch / httpd → board.py
+config → state / reports → taskfiles → events / github / drive / sync → agents → phases → watch / httpd → board.py
 ```
 
 - `config.py` — paths, stages, settings, prompt/adapter/driver resolution
@@ -57,6 +57,8 @@ config → state / reports → taskfiles → events / github / drive / sync →
 - `drive.py` — runs the project driver, tracks the one live drive
 - `sync.py` — origin/main as the shared board: push on move, pull on a beat
 - `agents.py` — headless work/review jobs, launched through the adapter
+- `phases.py` — a phase run: its own branch, its cards merged into it one
+  at a time. A beat, not an agent; it holds no registry of where a phase is
 - `watch.py` — 2s disk poller narrating moves made outside the API, and
   the gate that keeps a move a pull applied from triggering anything
 - `httpd.py` — HTTP routes, the SSE stream, serving the page
@@ -468,6 +470,12 @@ resolution commit, and refuses semantic ones, naming the collision for a
 human to settle. GitHub computes mergeability lazily, so an UNKNOWN
 reading keeps the chip's last state rather than flapping.
 
+A phase member's PR is opened against **its phase's branch**, not `main`:
+its branch was cut from there, so that is the only base whose diff is the
+member's own work — and a PR into `main` carrying a whole phase is exactly
+the merge this design refuses to make. Every other card, the phase card
+itself included, opens into `main` as it always did.
+
 The card wears that verdict in the design system's state colours:
 approved → pine (`--calm`) border and an `approved` pill; changes asked →
 terracotta (`--alarm`) and a `changes asked` pill; otherwise it stays the
@@ -475,7 +483,10 @@ neutral `waiting on you`. Tool chips (CI, copilot, PR, drive) are
 destinations, not statuses: they live in the card's footer row, never
 squeezed into the author row — `CI ✓` (pine), `CI ✕` (terracotta), `◌`
 while in flight — with hover actions staying in the status pill's slot.
-Merging remains yours — the board never merges.
+Merging into `main` remains yours — the board never merges into `main`. It
+does merge into a branch of its own: a phase's integration branch is the
+board's, and merging into it is bookkeeping in the same family as
+committing a move (see "A phase runs itself, on a branch of its own").
 
 The agent's first duty is to judge whether the task is actionable. If the
 task still has open questions — unresolved decisions only its author can
@@ -793,7 +804,83 @@ that would otherwise surface later as a runner behaving oddly.
 `**Depends on:**` is the other half, and it guards rather than orders: the
 list says what runs next, a member's dependencies say whether it *may*. The
 board parses the numbers out of the line and shows them; acting on them
-belongs to whatever runs a phase.
+belongs to the runner below.
+
+### A phase runs itself, on a branch of its own
+
+Running a phase works its list into a single integration branch. Starting
+one cuts `phase/<task-stem>` from the newest `origin/main` it can see — the
+same rule and the same timeout a task branch is cut by — and gives it a
+worktree beside the task worktrees. From there each member is branched
+**from the phase's tip**, run headless exactly as **▸ start work** runs any
+card, and merged back into the phase branch when its checks are green;
+then the next one starts. At the end one PR, from the phase branch into
+`main`, for a human.
+
+That is why the branch exists. Members of a phase are related by
+definition, so card two branched from `main` could not see card one's work
+while card one sat unmerged in `review/` — it would conflict, or quietly
+build the same thing twice. Gating on a merge into `main` would fix the
+branch point and destroy the point, because `main` is merged by a person
+and the phase would stall on every card. So the human gate moves from every
+card to the phase boundary, and the promise survives intact: the board
+merges into a branch it created, inside a scope you opened, and `main`
+still waits for your click.
+
+**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. It is a plain thread (`BOARD_PHASE_INTERVAL`, 30s), silent
+when no phase is running.
+
+**The beat is stateless.** Each pass recomputes which members are finished,
+which is first unfinished and what that one needs; it holds no registry of
+where a phase *is*. Two durable things carry the memory instead, and both
+are things the board already writes: **git**, where a member is finished
+when its branch is contained in the phase branch, and **the phase card**,
+which grows a `## Phase log` section the runner adds one line to per
+decision — a run started, a member started, a member merged, a halt. The
+log is the record a person reads, and the only thing that can tell "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. So a restart
+resumes a phase by looking, and the same logic answers "what now?" whether
+the last event was a launch, a merge or a crash.
+
+**Advance on green.** A member is finished when its card reaches `review/`
+and its checks are not against it. Green is read from the same PR poll the
+board already runs: red halts the phase, running holds it, and a member
+with no checks at all advances — a project without CI must not deadlock
+every phase it runs. A member with no branch at all that is already in
+`review/` or `done/` is simply finished; there is nothing to bring.
+
+**Halt, never skip.** Five conditions stop a phase, each already a visible
+state on the card: a member that exits `NOT READY`, a run that exits
+non-zero, a clean exit that committed nothing, CI red, and a merge into the
+phase branch that is not mechanical. A phase that stepped over a failed
+card would build the rest on a foundation that never landed. The halt is
+written into the log and then held — said once, not once a beat — and
+nothing retries by itself. Running the phase again is a person's decision,
+and it is what appends the line that clears the halt. A member whose
+`**Depends on:**` names something unfinished is a *wait*, not a halt: the
+phase idles until the dependency lands (merged, for a card inside the
+phase; `done/`, for one outside).
+
+**Merges are additive, always.** Nothing here rebases and nothing
+force-pushes. `main` is merged into the phase branch on every beat, so a
+phase that runs for hours does not drift into one enormous conflict at the
+end; a conflict there halts the phase like any other, aborted cleanly, with
+the colliding files named. When every member is in, the branch is pushed, a
+PR into `main` is opened with the member list as its body, the `**PR:**`
+line is written into the phase card and the card moves to `review/` — where
+the existing apparatus applies unchanged: the CI chip, **◔ review PR**,
+**⚑ copilot**, and drag-to-`done/` for **merge & clean up**.
+
+**One board runs it.** State syncs; reactions don't, so the phase card's
+**Assignee** is where "who runs it" is written down — the same claim that
+gates starting work. A replica renders the phase and advances nothing.
+
+Members run one at a time. Running independent members in parallel, and
+any UI beyond the API and the ticker, are separate cards.
 
 An optional **Assignee** line records who holds the card:
 

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
 

manager/core/agents.py

@@ -22,8 +22,17 @@ import config
 import events
 import reports
 import state
-from taskfiles import (actor_name, append_to_task, find_stage_of, move_task,
-                       read_task, set_assignee)
+from taskfiles import (actor_name, append_to_task, collect, find_stage_of,
+                       move_task, read_task, set_assignee)
+
+# A phase runs on an integration branch of its own — phases.py owns the
+# behaviour, the name lives here because this is where a launch decides
+# what to branch from.
+PHASE_BRANCH_PREFIX = "phase/"
+
+
+def phase_branch(phase_file: str) -> str:
+    return PHASE_BRANCH_PREFIX + phase_file[:-3]
 
 
 def _report_of(record: dict, text: str | None = None) -> str:
@@ -201,7 +210,38 @@ def _fresh_branch_point() -> tuple[str | None, str | None]:
     return "origin/main", None
 
 
-def _claim_for_launch(filename: str, stage: str, takeover: bool) -> None:
+def phase_branch_point(filename: str) -> tuple[str | None, str | None]:
+    """Where a *member of a running phase* branches from: the phase's own
+    branch, not main.
+
+    That is the whole reason a phase has a branch. Related cards run one
+    after another, so card two branched from main could not see card one's
+    work while card one sat unmerged in review/ — it would conflict, or
+    quietly build the same thing twice. Branching from the phase tip is
+    what makes the list add up.
+
+    (None, None) for a card in no phase, or one whose phase has not been
+    started — then the ordinary fresh branch point applies.
+    """
+    phase = None
+    try:
+        for stage in collect()["stages"]:
+            for task in stage["tasks"]:
+                if task["file"] == filename:
+                    phase = task.get("phase")
+    except OSError:
+        return None, None
+    if not phase:
+        return None, None
+    branch = phase_branch(phase["file"])
+    exists = subprocess.run(["git", "-C", str(config.REPO), "rev-parse",
+                             "--verify", "--quiet", branch], capture_output=True)
+    if exists.returncode != 0:
+        return None, None
+    return branch, f"branched from {branch}, the phase's own branch"
+
+
+def claim_for_launch(filename: str, stage: str, takeover: bool = False) -> None:
     """One agent per task is a board-memory rule; across machines the card
     file is the only thing every board can see, so the claim is what gates
     a launch here.
@@ -238,7 +278,7 @@ def start_agent(filename: str, stage: str, takeover: bool = False) -> dict:
     # Moving a card to in-progress is the commitment; only then does work start.
     _validate(filename, stage, {"in-progress"},
               "work starts from in-progress/ — move the card there first")
-    _claim_for_launch(filename, stage, takeover)
+    claim_for_launch(filename, stage, takeover)
 
     stem = filename[:-3]
     branch = f"task/{stem}"
@@ -267,7 +307,11 @@ def start_agent(filename: str, stage: str, takeover: bool = False) -> dict:
             base = _git("merge-base", "main", branch).stdout.strip()
             result = _git("worktree", "add", str(worktree), branch)
         else:
-            point, base_note = _fresh_branch_point()
+            # A phase member starts from its phase's tip; everything else
+            # from the newest main this checkout can see.
+            point, base_note = phase_branch_point(filename)
+            if point is None:
+                point, base_note = _fresh_branch_point()
             if point:
                 base = _git("rev-parse", point).stdout.strip()
                 result = _git("worktree", "add", "--no-track", "-b", branch,

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.

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

manager/core/github.py

@@ -27,7 +27,8 @@ import config
 import drive as drive_mod
 import reports
 import state
-from taskfiles import STATUS_RE, commit_edit, find_stage_of, move_task, read_task
+from taskfiles import (STATUS_RE, collect, commit_edit, find_stage_of,
+                       move_task, read_task)
 
 PR_STATE: dict[str, dict] = {}   # filename -> {verdict, detail, url, ts}
 _OPENING: set[str] = set()       # filenames with a PR-open in flight
@@ -52,6 +53,75 @@ def _branch_exists(branch: str) -> bool:
     return _run(["git", "rev-parse", "--verify", "--quiet", branch]).returncode == 0
 
 
+def branch_of(filename: str) -> str:
+    """The branch a card's work lives on. Ordinary cards get `task/<stem>`;
+    a phase runs on `phase/<stem>`, its own integration branch (phases.py),
+    and from review/ onwards it is reviewed, driven and completed through
+    the same apparatus as any other card."""
+    phase = f"phase/{filename[:-3]}"
+    return phase if _branch_exists(phase) else f"task/{filename[:-3]}"
+
+
+def _woven(filename: str, stage: str) -> dict:
+    """The card as the board shows it. A card's phase is *derived* across
+    the whole board (`taskfiles.weave_phases`), and both what a PR is based
+    on and what a phase's PR says depend on that reading, so a lone
+    `read_task` is not enough here."""
+    for group in collect()["stages"]:
+        for task in group["tasks"]:
+            if task["file"] == filename:
+                return task
+    return read_task(config.TASKS / stage / filename, stage)
+
+
+def _remote_has_branch(rname: str, branch: str) -> bool:
+    """Whether the remote already carries this branch. Sync only ever
+    fetches origin/main, so a phase branch the running board pushed is not a
+    local ref on any other board even when it is published — asking the
+    remote directly is the only way another board sees it."""
+    return bool(_run(["git", "ls-remote", "--heads", rname, branch],
+                     timeout=30).stdout.strip())
+
+
+def _pr_base(task: dict) -> str:
+    """What a PR is opened against. A phase member's branch was cut from
+    its phase's branch, so that is the only base whose diff is the member's
+    own work — a PR into main would carry the whole phase, and invite a
+    merge into main that this board exists not to make. Everything else,
+    the phase card included, goes into main.
+
+    The phase branch is the board's own and is published as the phase runs,
+    so a board that did not run the phase may know it only through the
+    remote — that still makes it the base, not main."""
+    phase = task.get("phase")
+    if phase:
+        branch = f"phase/{phase['file'][:-3]}"
+        if _branch_exists(branch):
+            return branch
+        rname = remote()
+        if rname and _remote_has_branch(rname, branch):
+            return branch
+    return "main"
+
+
+def _pr_body(filename: str, task: dict) -> str:
+    """The PR's body. A phase's PR is the one a whole run produces, so it
+    says what is in it: the member list, in the order it ran."""
+    if task.get("isPhase"):
+        cards = "\n".join(f"- {m['number']} — {m['title']}"
+                          for m in task.get("members") or [])
+        return (f"Phase: `{filename}` — tracked in `.task-manager/tasks/review/`.\n\n"
+                f"Opened by the board when every card in the phase had been "
+                f"merged into its branch.\n\n## Cards in this phase\n\n"
+                f"{cards or '_none_'}\n")
+    body = (f"Task: `{filename}` — tracked in `.task-manager/tasks/review/`.\n\n"
+            f"Opened by the board when the card moved to review.")
+    summary = _agent_report(filename)
+    if summary:
+        body += f"\n\n## Agent summary\n\n{summary}"
+    return body
+
+
 def _write_pr_line(filename: str, url: str) -> None:
     """The url joins the header — and in team mode commits itself, so the
     gate that stops a second board opening a second PR travels to the other
@@ -115,14 +185,14 @@ def open_pr_now(filename: str) -> str:
 
 
 def _open_pr(filename: str) -> str:
-    branch = f"task/{filename[:-3]}"
+    branch = branch_of(filename)
     if not _branch_exists(branch):
         # nothing to publish — a hand-moved card without agent work
         raise _Quiet(f"{filename} has no {branch} branch — nothing to open a PR from")
     stage = find_stage_of(filename)
     if stage != "review":
         raise _Quiet(f"{filename} is not in review/ — PRs open from there")
-    task = read_task(config.TASKS / stage / filename, stage)
+    task = _woven(filename, stage)
     if task.get("pr"):
         raise _Quiet(f"{filename} already has a PR: {task['pr']}")
 
@@ -132,24 +202,29 @@ def _open_pr(filename: str) -> str:
                          ("no git remote configured" if rname is None
                           else "gh is not installed"))
 
-    # The PR's diff is computed against the remote main — refuse to open one
-    # that would drag unpushed main commits along with it.
-    _run(["git", "fetch", rname, "main"], timeout=120)
-    ahead = _run(["git", "rev-list", "--count", f"{rname}/main..main"]).stdout.strip()
-    if ahead.isdigit() and int(ahead) > 0:
-        raise ValueError(f"won't open a PR for {filename}: main is {ahead} commits "
-                         f"ahead of {rname} — push main first, then move the card again")
+    base = _pr_base(task)
+    if base == "main":
+        # The PR's diff is computed against the remote main — refuse to open one
+        # that would drag unpushed main commits along with it.
+        _run(["git", "fetch", rname, "main"], timeout=120)
+        ahead = _run(["git", "rev-list", "--count", f"{rname}/main..main"]).stdout.strip()
+        if ahead.isdigit() and int(ahead) > 0:
+            raise ValueError(f"won't open a PR for {filename}: main is {ahead} commits "
+                             f"ahead of {rname} — push main first, then move the card again")
+    elif _branch_exists(base):
+        # A phase branch is the board's own: publish it so the member's PR
+        # has a base on the remote to be opened against. When only the
+        # remote carries it — a member PR opened from a board that did not
+        # run the phase — it is already there, and there is nothing local
+        # to push.
+        _run(["git", "push", "-u", rname, base], timeout=180)
 
     push = _run(["git", "push", "-u", rname, branch], timeout=180)
     if push.returncode != 0:
         raise ValueError(f"push failed for {branch}: {push.stderr.strip()[:140]}")
 
-    body = (f"Task: `{filename}` — tracked in `.task-manager/tasks/review/`.\n\n"
-            f"Opened by the board when the card moved to review.")
-    summary = _agent_report(filename)
-    if summary:
-        body += f"\n\n## Agent summary\n\n{summary}"
-    result = _run([config.GH_BIN, "pr", "create", "--head", branch, "--base", "main",
+    body = _pr_body(filename, task)
+    result = _run([config.GH_BIN, "pr", "create", "--head", branch, "--base", base,
                    "--title", task["title"], "--body", body], timeout=120)
     if result.returncode != 0:
         # The rare double-fire: two attempts crossed and GitHub already has
@@ -417,7 +492,7 @@ def complete_task(filename: str, stage: str) -> dict:
 def _complete(filename: str, stage: str) -> dict:
     """The steps themselves, run under the claim complete_task holds."""
     stem = filename[:-3]
-    branch = f"task/{stem}"
+    branch = branch_of(filename)   # a phase's own branch, or task/<stem>
 
     # 1. the app must not keep running code that is about to be merged away
     d = drive_mod.DRIVE
@@ -512,11 +587,17 @@ def _merge_on_origin(filename: str, stage: str, branch: str) -> None:
 
 
 def task_branches() -> list[str]:
-    """Stems of all task/* branches — the UI uses this to say honestly
-    whether a review card has work attached."""
+    """Stems of all task/* and phase/* branches — the UI uses this to say
+    honestly whether a card has work attached, and a phase card's work is
+    its own integration branch."""
     result = _run(["git", "for-each-ref", "--format=%(refname:short)",
-                   "refs/heads/task/"])
-    return [ref[len("task/"):] for ref in result.stdout.split() if ref.startswith("task/")]
+                   "refs/heads/task/", "refs/heads/phase/"])
+    stems = []
+    for ref in result.stdout.split():
+        for prefix in ("task/", "phase/"):
+            if ref.startswith(prefix):
+                stems.append(ref[len(prefix):])
+    return stems
 
 
 def reconcile() -> None:

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"])})

manager/core/phases.py

@@ -0,0 +1,605 @@
+"""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 _write_log(phase: dict, entry: str) -> bool:
+    """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. Returns whether the line landed."""
+    stage = taskfiles.find_stage_of(phase["file"]) or phase["stage"]
+    stamp = time.strftime("%Y-%m-%d %H:%M")
+    return taskfiles.append_to_section(phase["file"], stage, LOG_HEADING,
+                                       f"- {stamp} · {entry}", "phase log")
+
+
+def _record(phase: dict, entry: str) -> None:
+    """Record a decision, and refuse to act if it cannot be written down.
+
+    The log is the durable memory a restart reads to tell "this member has
+    already started" from "the phase has not reached it yet". An action that
+    outran its own record — a launch or a merge with no line behind it —
+    is exactly what a restarted board would repeat. So a log that will not
+    take the line is itself a halt, raised here before the action it was
+    meant to precede ever happens. The halt path writes best-effort
+    (`_write_log`) so that recording the halt can never raise in turn."""
+    if not _write_log(phase, entry):
+        raise _Halt("could not write the phase log — refusing to act without "
+                    "the record a restart reads")
+
+
+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.
+
+    No local git identity is the one case that does not gate: it is exactly
+    where `agents.claim_for_launch` cannot write an assignee and so cannot
+    refuse a launch either. Gating the beat on it while the launch went
+    through would strand a phase — branch cut, run recorded — that then
+    never advances. So a board with no name is the lone actor here, the same
+    as it is for starting work."""
+    if not config.COMMIT_MOVES:
+        return True
+    me = taskfiles.actor_name()
+    if not me:
+        return True
+    return 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 ""
+    _write_log(phase, f"halted{at} — {reason}")   # best effort: never re-raise
+    _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)
+    if not _write_log(phase, f"run started on {branch}"):
+        raise ValueError(f"could not record the run on {filename} — its phase "
+                         f"log must be writable to run the phase safely")
+    _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)

manager/core/taskfiles.py

@@ -450,6 +450,40 @@ def append_to_task(filename: str, stage: str, text: str, what: str) -> bool:
     return True
 
 
+def append_to_section(filename: str, stage: str, heading: str, line: str,
+                      what: str) -> bool:
+    """Add one line under `## <heading>`, creating the section at the end of
+    the card when it is not there yet.
+
+    The third door, and the narrowest: a running record where every entry is
+    one line and belongs under one heading — the phase log. `append_to_task`
+    would scatter those lines through the file as other sections (a work
+    report, a review) landed between them, and the record would stop being
+    readable in the one place a person looks. It commits like every other
+    board-made write.
+    """
+    path = config.TASKS / stage / filename
+    try:
+        text = path.read_text(encoding="utf-8")
+    except OSError:
+        return False
+    section = re.compile(rf"^##\s+{re.escape(heading)}\s*$(.*?)(?=^##\s|\Z)",
+                         re.MULTILINE | re.DOTALL).search(text)
+    if section:
+        body = section.group(1).strip("\n")
+        updated = f"## {heading}\n\n{body}\n{line}\n\n" if body else \
+                  f"## {heading}\n\n{line}\n\n"
+        text = text[:section.start()] + updated + text[section.end():]
+    else:
+        text = text.rstrip("\n") + f"\n\n## {heading}\n\n{line}\n"
+    try:
+        path.write_text(text, encoding="utf-8")
+    except OSError:
+        return False
+    commit_edit(filename, stage, what)
+    return True
+
+
 def move_task(filename: str, source: str, target: str, actor: str = "you") -> dict:
     """Move a task file between stage directories and fix its Status line.
 

site/templates/home.html

@@ -165,8 +165,8 @@ Task board for ~/your-repo/.task-manager/tasks
     <div class="why-reason">
       <span class="why-lead">Nothing merges without you</span>
       <span class="why-text">Finished work arrives as a PR with CI on the
-        card. bench never merges — the button stays yours, and you stop
-        being the thing everything queues behind.</span>
+        card. bench never merges into main — the button stays yours, and
+        you stop being the thing everything queues behind.</span>
     </div>
   </div>
 </section>

tests/test_actor_acts.py

@@ -430,7 +430,7 @@ class ClaimsGateLaunches(Boards):
                              BRANCH).returncode, 1)
 
     def test_the_deliberate_takeover_reassigns_the_card(self):
-        agents._claim_for_launch(FILENAME, "in-progress", True)
+        agents.claim_for_launch(FILENAME, "in-progress", True)
 
         self.assertIn("**Assignee:** elena", self.text(self.elena))
         self.assertEqual(self.text(self.elena).count("**Assignee:**"), 1)
@@ -440,7 +440,7 @@ class ClaimsGateLaunches(Boards):
                             for s in self.summaries()))
 
     def test_the_takeover_reaches_the_other_board(self):
-        agents._claim_for_launch(FILENAME, "in-progress", True)
+        agents.claim_for_launch(FILENAME, "in-progress", True)
         self.assertEqual(sync.push_now(), "ok")
 
         self.use(self.ada)
@@ -450,7 +450,7 @@ class ClaimsGateLaunches(Boards):
     def test_an_unclaimed_card_claims_on_launch(self):
         self.place(self.elena, "in-progress", card("In Progress"), commit=False)
 
-        agents._claim_for_launch(FILENAME, "in-progress", False)
+        agents.claim_for_launch(FILENAME, "in-progress", False)
 
         self.assertIn("**Assignee:** elena", self.text(self.elena))
         self.assertTrue(any("claimed" in s for s in self.summaries()))
@@ -460,7 +460,7 @@ class ClaimsGateLaunches(Boards):
                    commit=False)
         head = git(self.elena, "rev-parse", "HEAD").stdout
 
-        agents._claim_for_launch(FILENAME, "in-progress", False)
+        agents.claim_for_launch(FILENAME, "in-progress", False)
 
         self.assertEqual(git(self.elena, "rev-parse", "HEAD").stdout, head,
                          "nothing to record: it was already yours")
@@ -470,7 +470,7 @@ class ClaimsGateLaunches(Boards):
         a lock — a hand-written line stays decoration."""
         self.patch(SYNC=False, COMMIT_MOVES=False)
 
-        agents._claim_for_launch(FILENAME, "in-progress", False)
+        agents.claim_for_launch(FILENAME, "in-progress", False)
 
         self.assertIn("**Assignee:** ada", self.text(self.elena))
 

tests/test_phase_runs.py

@@ -0,0 +1,939 @@
+"""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 AnUnwritableLogHalts(PhaseCase):
+    """The log is the durable memory a restart reads; a launch or a merge
+    with no line behind it is exactly what a restarted board would repeat.
+    So a log that will not take the line halts before the action it was
+    meant to precede."""
+
+    def test_a_launch_never_outruns_its_record(self):
+        real = phases._write_log
+        self.addCleanup(setattr, phases, "_write_log", real)
+        # the one line that will not write is the member-started record
+        phases._write_log = lambda phase, entry: (
+            False if entry == "31 started" else real(phase, entry))
+
+        self.start()
+
+        self.assertEqual(
+            [r for r in state.AGENTS.values() if r["task"] == ONE], [],
+            "the member was never launched")
+        self.assertNotIn("31 started", self.log())
+        self.assertTrue(any("halted" in s for s in self.summaries()),
+                        "the phase halts instead of launching blind")
+
+
+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_a_board_with_no_git_identity_is_not_gated_out(self):
+        """Team mode with no local git name is the one case that does not
+        gate: `claim_for_launch` cannot write an assignee there and so
+        cannot refuse a launch, and the beat must match it — else a phase
+        starts (branch cut, run recorded) and then advances nowhere."""
+        self.addCleanup(setattr, taskfiles, "actor_name", taskfiles.actor_name)
+        taskfiles.actor_name = lambda: ""
+
+        self.assertTrue(phases._mine({"file": PHASE, "assignee": "elena"}))
+
+    def test_a_named_board_that_is_not_the_assignee_is_gated_out(self):
+        self.addCleanup(setattr, taskfiles, "actor_name", taskfiles.actor_name)
+        taskfiles.actor_name = lambda: "ronald"
+
+        self.assertFalse(phases._mine({"file": PHASE, "assignee": "elena"}))
+
+    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")
+
+    def test_a_member_pr_bases_on_a_phase_branch_only_the_remote_carries(self):
+        """A board that did not run the phase knows the phase branch only
+        through the remote — sync fetches origin/main and nothing else. That
+        branch is still the base a member's PR opens against, never main."""
+        git(self.repo, "push", "-q", "origin", f"main:refs/heads/{PHASE_BRANCH}")
+
+        self.assertFalse(self.branch_exists(PHASE_BRANCH), "no local phase branch")
+        self.assertEqual(github._pr_base({"phase": {"file": PHASE}}), PHASE_BRANCH)
+
+
+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()