External/bench
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

A phase runs itself, on a branch of its own

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

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

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

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

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

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

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