From 8f94fe1a3dcfb41a4c9bc1a0e4e36ce8be85bc01 Mon Sep 17 00:00:00 2001
From: Jammy2211 <JNightingale2211@gmail.com>
Date: Thu, 3 Sep 2026 20:42:25 -0400
Subject: [PATCH] feat(cortex): the by-project check-in view and the two
 missing prompts
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

The original request asked for one thing the board could not answer: "where
is every project, what folders should I check out, and give me the prompt to
continue any one of them". Every section was keyed by phase STATE, the
Projects table omitted the three folders a project lives in, and each phase's
`## Where to look` bullets — the folder answer itself — were parsed to locate
artefacts and never rendered.

`## By project` is that view, and it is one tree rendered three ways: the
first section of `dashboard.md` / `dashboard.html`, the text
`cortex census --by-project` prints from the checkout alone, and the closing
summary of a real `checkin` (the only one carrying pull results and six-leg
health). Per project: its `local_path` / `mirror` / `ral_root`, its phase
counts, then every phase still open — awaiting, live, ready, gated, planned —
with its state, its `## Where to look` bullets verbatim, and a copy chip per
prompt its state carries. History (accepted / rerun / dropped) is counted,
not listed; a project with nothing open folds to one line.

Two of the five starting prompts were missing and are now here, beside the
rule prompt rather than instead of it:

- `_next_phase_payload` — "the results are good": the accept, then the next
  phase. It prefills from the tree — a planned or gated sibling numbered N+1
  is opened rather than duplicated; a taken number is named and the lowest
  free one offered — and spells out `new`'s own title rule, that it writes
  `# <Project> — phase N: ` in front of `--title` itself.
- `_rerun_payload` — "run it again": the rerun ruling first (the only verb
  that puts a phase back on the board), then ready, the project's own submit
  and the move that records the job id.

`phase_chips()` is the one table saying which prompts a state carries, so a
prompt cannot appear on the board and not in the door. The counts strip is
deliberately unchanged — `board/_board.py` reads it by regex.

Refs PyAutoCortex#12.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_013HsZA1ufn3msgPiDFxEXa6
---
 agents/conductors/cortex/_cortex.py | 473 +++++++++++++++++++++++-----
 tests/test_cortex_conductor.py      | 189 +++++++++++
 2 files changed, 578 insertions(+), 84 deletions(-)

diff --git a/agents/conductors/cortex/_cortex.py b/agents/conductors/cortex/_cortex.py
index 541d4ea5..da4202e0 100644
--- a/agents/conductors/cortex/_cortex.py
+++ b/agents/conductors/cortex/_cortex.py
@@ -227,6 +227,12 @@ def _home_from_git(root: Path) -> str:
 
 # One section: (key, title, source path in the repo, blurb).
 SECTIONS = (
+    ("by_project", "By project", "projects.yaml",
+     "Where every project is: the folders it lives in, every phase of it that "
+     "is still open, the locations its `## Where to look` names, and the "
+     "prompt for each thing that phase could become. This is the check-in "
+     "view — `pyauto-brain cortex checkin` and `census --by-project` print "
+     "the same tree."),
     ("awaiting", "Awaiting ruling", "phases/",
      "Results are in and nothing is running — the human's verdict is the only "
      "thing outstanding. Ordered failures first, then the phases a ruling is "
@@ -269,6 +275,24 @@ def _int(value: str) -> int | None:
     return int(v) if v.isdigit() else None
 
 
+def _where_bullets(mod, ph) -> list[str]:
+    """The phase's own `## Where to look` bullets, `- ` stripped, verbatim.
+
+    A phase names where its results are. The by-project view renders this
+    text as the "which folder do I open" answer and `where_paths` resolves
+    the same bullets to paths — one reading, two uses.
+    """
+    getter = getattr(mod, "_where_to_look", None)
+    if getter is not None:
+        bullets = getter(ph)
+    else:  # a checkout whose script predates the helper
+        span = mod.sections(ph.text).get("Where to look")
+        bullets = ([ln for ln in ph.text.split("\n")[span[0]:span[1]]
+                    if ln.startswith("- ") and ln.strip() != "-"]
+                   if span else [])
+    return [b[2:].strip() for b in bullets if b[2:].strip()]
+
+
 def _phase_row(mod, ph) -> dict:
     refs, bad_refs = mod.gate_refs(ph.get("Gates"))
     runs = [{"ident": r.ident, "state": r.state, "partition": r.partition,
@@ -286,6 +310,7 @@ def _phase_row(mod, ph) -> dict:
         "gates": refs,
         "bad_gates": bad_refs,
         "witness": ph.get("Witness"),
+        "where": _where_bullets(mod, ph),
         "budget": budget,
         "budget_minutes": _mins(budget),
         "runs": runs,
@@ -375,6 +400,10 @@ def census(root: Path) -> dict:
                                   r["phase"] or 0, r["rel"]))
     gated = sorted([r for r in rows if r["state"] == "gated"],
                    key=lambda r: (r["project"], r["phase"] or 0, r["rel"]))
+    # Not a board section — `planned` is the tail of the by-project tree, the
+    # "and then what" a check-in reads after everything open.
+    planned = sorted([r for r in rows if r["state"] == "planned"],
+                     key=lambda r: (r["project"], r["phase"] or 0, r["rel"]))
 
     home = _home(root)
     return {
@@ -387,6 +416,7 @@ def census(root: Path) -> dict:
         "live": live,
         "ready": ready,
         "gated": gated,
+        "planned": planned,
         "rulings": ruling_rows,
         "projects": projects,
         "epics": parse_epics(root / "epics.md"),
@@ -518,21 +548,24 @@ def _live_payload(r: dict, projects: dict) -> str:
             f"({', '.join(x['ident'] for x in r['runs']) or 'no runs'}) stand.")
 
 
+def _submit_line(project: str, projects: dict) -> str:
+    """The project's own submit verb, as a human would type it — or the note
+    that says the row has no such verb. Every path comes from the row."""
+    row = projects.get(project, {})
+    cli, path = (row.get("sync_cli") or ""), (row.get("local_path") or "")
+    if cli and path and "submit" in (row.get("sync_verbs") or []):
+        return f"cd {path} && {cli} submit <script>"
+    return (f"# project {project}: no `submit` verb in projects.yaml — "
+            "submit by hand")
+
+
 def launch_payload(r: dict, projects: dict) -> list[str]:
     """The launch lines for a ready phase: the phase, the project's own
     submit verb, and the move that records the job id. No decision rides
     here — everything decided was decided when the plan was approved."""
-    row = projects.get(r["project"], {})
-    cli, path = row.get("sync_cli", ""), row.get("local_path", "")
-    lines = [r["rel"]]
-    if cli and path and "submit" in (row.get("sync_verbs") or []):
-        lines.append(f"cd {path} && {cli} submit <script>")
-    else:
-        lines.append(f"# project {r['project']}: no `submit` verb in "
-                     "projects.yaml — submit by hand")
-    lines.append(f"python3 scripts/cortex.py move {r['rel']} submitted "
-                 "--run <jobid>")
-    return lines
+    return [r["rel"], _submit_line(r["project"], projects),
+            f"python3 scripts/cortex.py move {r['rel']} submitted "
+            "--run <jobid>"]
 
 
 def _ready_payload(r: dict, projects: dict) -> str:
@@ -544,6 +577,172 @@ def _gate_payload(r: dict) -> str:
             f"move {r['rel']} ready\n# gates: " + ", ".join(r["gates"]))
 
 
+#: The states a phase already in the tree can still be opened from — the
+#: two the `move` table lets a human take to `ready`.
+OPENABLE_STATES = ("planned", "gated")
+
+
+def _next_phase(r: dict, c: dict) -> dict | None:
+    """The phase numbered N+1 of this project, when the tree already holds it.
+
+    A planned sibling IS the ledger naming the next phase — `epics.md` points
+    at `phases/<project>/` for exactly this reason — so the prompt opens the
+    phase that exists rather than writing a second one beside it.
+    """
+    if r["phase"] is None:
+        return None
+    for row in c["phases"]:
+        if row["project"] == r["project"] and row["phase"] == r["phase"] + 1:
+            return row
+    return None
+
+
+def _next_free_phase(project: str, c: dict) -> int:
+    """The lowest phase number this project has not used. Phase numbers are
+    unique per project, so a `new` that reuses one is drift."""
+    used = [r["phase"] for r in c["phases"]
+            if r["project"] == project and r["phase"] is not None]
+    return max(used) + 1 if used else 1
+
+
+def _next_phase_payload(r: dict, c: dict) -> str:
+    """"The results are good" — file the ruling, then open phase N+1.
+
+    Two commands in one prompt because they are one decision: an accept that
+    opens nothing leaves the programme where it was. `cortex.py new` writes
+    `# <Project> — phase N: ` in front of `--title` itself, so the prompt
+    asks for the tail alone.
+    """
+    nxt = _next_phase(r, c)
+    L = [f"The results for the PyAutoCortex phase {r['rel']} are good. Read "
+         "its `## Witness` and the evidence under its `## Where to look`, "
+         "draft the accept body for my approval, then file it and open the "
+         "next phase:",
+         f"python3 scripts/cortex.py rule {r['rel']} accept --body <file>"]
+    if nxt is not None and nxt["state"] in OPENABLE_STATES:
+        L += [f"# phase {nxt['phase']} is already in the tree ({nxt['state']})"
+              " — open that one, do not write a second:",
+              f"python3 scripts/cortex.py move {nxt['rel']} ready",
+              *launch_payload(nxt, c["projects"])[1:]]
+    else:
+        num = _next_free_phase(r["project"], c)
+        epic = f" --epic {r['epic']}" if r.get("epic") else ""
+        if nxt is not None:
+            L.append(f"# phase {nxt['phase']} is taken ({nxt['rel']}, "
+                     f"{nxt['state']}) — the next free number is {num}:")
+        L += [f"python3 scripts/cortex.py new {r['project']} <slug> --phase "
+              f"{num}{epic} --title \"<the tail only — `new` writes "
+              f"'<Project> — phase {num}: ' itself>\"",
+              f"python3 scripts/cortex.py move phases/{r['project']}/"
+              "<slug>.md ready",
+              _submit_line(r["project"], c["projects"])]
+    return "\n".join(L)
+
+
+def _rerun_payload(r: dict, c: dict) -> str:
+    """"Run it again" — the ruling that says so, then the same launch.
+
+    A rerun is a ruling first: the verdict on what came back is what makes
+    the next submission a rerun rather than a repeat, and `rule … rerun`
+    is the only edge that puts a phase back on the board.
+    """
+    return "\n".join([
+        f"The PyAutoCortex phase {r['rel']} needs running again. Draft the "
+        "rerun body — what came back, and what changes — for my approval, "
+        "then file it and relaunch:",
+        f"python3 scripts/cortex.py rule {r['rel']} rerun --body <file>",
+        f"python3 scripts/cortex.py move {r['rel']} ready",
+        _submit_line(r["project"], c["projects"]),
+        f"python3 scripts/cortex.py move {r['rel']} submitted --run <jobid>"
+        "   # one run per call; --after <run> chains the next",
+    ])
+
+
+def _planned_payload(r: dict) -> str:
+    return (f"python3 scripts/cortex.py move {r['rel']} ready   # when the "
+            "`Ready when:` clause in its `## Question` is met")
+
+
+def phase_chips(r: dict, c: dict) -> list[tuple[str, str]]:
+    """`(label, payload)` for every prompt this phase's state carries.
+
+    The one place the board, `census --by-project` and the check-in door
+    agree on what a human can do with a phase today — a state that grows a
+    prompt grows it here and appears in all three.
+    """
+    state, chips = r["state"], []
+    if state in AWAITING_STATES:
+        chips.append(("rule on it", _ruling_payload(r)))
+        nth = "" if r["phase"] is None else f" {r['phase'] + 1}"
+        chips.append((f"the results are good — accept and open phase{nth}",
+                      _next_phase_payload(r, c)))
+    if state in LIVE_STATES:
+        chips.append(("where the jobs stand", _live_payload(r, c["projects"])))
+    if state == "ready":
+        chips.append(("submit it", _ready_payload(r, c["projects"])))
+    if state == "gated":
+        chips.append(("its gates", _gate_payload(r)))
+    if state == "planned":
+        chips.append(("open it", _planned_payload(r)))
+    if state in AWAITING_STATES or state == "running":
+        chips.append(("run it again", _rerun_payload(r, c)))
+    return chips
+
+
+#: The by-project reading order — the state order a check-in works down.
+#: `accepted`, `rerun` and `dropped` are history and fold into the counts.
+PROJECT_GROUPS = (("awaiting", "Awaiting your ruling"),
+                  ("live", "Still out there"),
+                  ("ready", "Ready to submit"),
+                  ("gated", "Gated"),
+                  ("planned", "Planned"))
+
+#: The buckets that make a dormant project worth a block of its own: a job
+#: still out there, or a verdict still owed, is still the human's business.
+OPEN_BUCKETS = ("awaiting", "live", "ready")
+
+
+def project_groups(key: str, c: dict) -> list[tuple[str, list[dict]]]:
+    """`(title, rows)` for one project, in the order a check-in reads them.
+    Empty groups are dropped — a project shows only what it actually has."""
+    out = []
+    for bucket, title in PROJECT_GROUPS:
+        rows = [r for r in c[bucket] if r["project"] == key]
+        if rows:
+            out.append((title, rows))
+    return out
+
+
+def by_project_keys(c: dict) -> tuple[list[str], list[str]]:
+    """`(shown, folded)` — a block each for every `status: active` project
+    plus any project still holding an open phase; everything else folds to
+    one line, because a dormant project with nothing open is a fact, not a
+    thing to read."""
+    shown = {key for key, row in c["projects"].items()
+             if (row.get("status") or "").strip() == "active"}
+    for bucket in OPEN_BUCKETS:
+        shown |= {r["project"] for r in c[bucket] if r["project"] in c["projects"]}
+    return sorted(shown), sorted(set(c["projects"]) - shown)
+
+
+def project_paths(row: dict) -> list[str]:
+    """The three folders a project lives in — the answer to "where is it"."""
+    where = [f"local `{row.get('local_path') or '?'}`"]
+    if (row.get("mirror") or "none") != "none":
+        where.append(f"mirror `{row['mirror']}`")
+    where.append(f"RAL `{row.get('ral_root') or '?'}`")
+    return where
+
+
+def project_counts(key: str, c: dict) -> str:
+    counts: dict[str, int] = {}
+    for r in c["phases"]:
+        if r["project"] == key:
+            counts[r["state"]] = counts.get(r["state"], 0) + 1
+    return " · ".join(f"{state} {n}" for state, n in sorted(counts.items())) \
+        or "none"
+
+
 def _project_payload(key: str, row: dict) -> str:
     path = row.get("mirror") if row.get("mirror") not in ("", "none") else row.get("local_path", "")
     return f"Show me what is under {path} for the science project {key}."
@@ -584,6 +783,32 @@ def _gate_note(r: dict) -> str:
     return ", ".join(r["gates"]) or "no refs"
 
 
+def _project_head(r: dict) -> str:
+    """A by-project phase line: the phase head, its state, and the warning a
+    failed run earns. State is on the line because this section is not keyed
+    by it."""
+    head = _phase_head(r) + f" — **{r['state']}**"
+    if r["failed_runs"]:
+        head += " — ⚠️ " + _summary_label(
+            "failed runs: " + ", ".join(r["failed_runs"]))
+    return head
+
+
+def _project_phase_md(r: dict, c: dict) -> list[str]:
+    """One phase inside `## By project`: the line, the folders it names
+    verbatim, then a copy chip per prompt its state carries."""
+    L = [_project_head(r), ""]
+    note = _live_note(r) if r["state"] in LIVE_STATES else ""
+    if note:
+        L += [f"- {_summary_label(note)}", ""]
+    if r["where"]:
+        L += [f"- where to look: {b}" for b in r["where"]] + [""]
+    L += _items([_task_row(label, payload)
+                 for label, payload in phase_chips(r, c)]) or \
+        ["- _(no prompt for this state)_"]
+    return L + [""]
+
+
 def render_dashboard(c: dict) -> str:
     """The Cortex board as `dashboard.md`.
 
@@ -637,20 +862,57 @@ def h2(key: str) -> list:
         L += [""]
 
 
+    # First section, above the state lists: on a phone the state lists are
+    # cross-project and every row has to be read to find out whose it is,
+    # while this one answers "where is everything" a screen per project and
+    # opens with the folders. The counts table sits above it, so the urgency
+    # sections stay one tap away.
+    L += h2("by_project")
+    shown, folded = by_project_keys(c)
+    for key in shown:
+        L += [f"### {_summary_label(key)}", "",
+              "- " + " · ".join(project_paths(c["projects"][key])),
+              f"- phases: {project_counts(key, c)}", ""]
+        groups = project_groups(key, c)
+        if not groups:
+            L += ["_Nothing open — every phase of this project is history._",
+                  ""]
+        for title, rows in groups:
+            L += [f"**{title}**", ""]
+            for r in rows:
+                L += _project_phase_md(r, c)
+    if folded:
+        L += [f"<details><summary>{len(folded)} dormant project(s) with "
+              "nothing open</summary>", "",
+              " · ".join(f"`{_cell(k)}` ({project_counts(k, c)})"
+                         for k in folded),
+              "", "</details>", ""]
+
     L += h2("awaiting")
-    L += _items([_task_row(_phase_head(r)
-                           + (" — ⚠️ " + _summary_label(
-                               "failed runs: " + ", ".join(r["failed_runs"]))
-                              if r["failed_runs"] else ""),
-                           _ruling_payload(r))
-                 for r in c["awaiting"]]) or ["- _(nothing awaiting a ruling)_"]
+    awaiting_rows = []
+    for r in c["awaiting"]:
+        head = _phase_head(r) + (" — ⚠️ " + _summary_label(
+            "failed runs: " + ", ".join(r["failed_runs"]))
+            if r["failed_runs"] else "")
+        # chip 0 IS the rule prompt this row has always carried; the rest —
+        # accept-and-open-the-next, run-it-again — ride beside it.
+        chips = phase_chips(r, c)
+        awaiting_rows.append(_task_row(head, chips[0][1]))
+        awaiting_rows += [_task_row("↳ " + label, payload)
+                          for label, payload in chips[1:]]
+    L += _items(awaiting_rows) or ["- _(nothing awaiting a ruling)_"]
     L += [""]
 
     L += h2("live")
-    L += _items([_task_row(_phase_head(r) + (" — " + _summary_label(note)
-                                             if (note := _live_note(r)) else ""),
-                           _live_payload(r, c["projects"]))
-                 for r in c["live"]]) or ["- _(nothing on the queue)_"]
+    live_rows = []
+    for r in c["live"]:
+        head = _phase_head(r) + (" — " + _summary_label(note)
+                                 if (note := _live_note(r)) else "")
+        chips = phase_chips(r, c)
+        live_rows.append(_task_row(head, chips[0][1]))
+        live_rows += [_task_row("↳ " + label, payload)
+                      for label, payload in chips[1:]]
+    L += _items(live_rows) or ["- _(nothing on the queue)_"]
     L += [""]
 
     L += h2("ready")
@@ -787,10 +1049,48 @@ def phase_head(r, tone_pills=()):
         H += ["</ul>"]
 
 
+    H.append(h2("by_project"))
+    shown, folded = by_project_keys(c)
+    for key in shown:
+        H += [f"<h3>{_summary_label(key)}</h3>",
+              '<p class="muted">'
+              + _md_inline(" · ".join(project_paths(c["projects"][key])))
+              + f' — phases: {_summary_label(project_counts(key, c))}</p>']
+        groups = project_groups(key, c)
+        if not groups:
+            H.append('<p class="muted">(nothing open — every phase of this '
+                     "project is history)</p>")
+        for title, rows in groups:
+            H.append(f"<p><b>{title}</b></p>")
+            for r in rows:
+                text = phase_head(r, ((r["state"], "n"),))
+                if r["failed_runs"]:
+                    text += ('<span class="facets"> — ⚠️ failed runs: '
+                             + _summary_label(", ".join(r["failed_runs"]))
+                             + "</span>")
+                note = _live_note(r) if r["state"] in LIVE_STATES else ""
+                if note:
+                    text += f'<span class="facets"> — {_summary_label(note)}</span>'
+                H.append(f"<p>{text}</p>")
+                if r["where"]:
+                    H += ["<ul>"] + [f"<li>{_md_inline(_summary_label(b))}</li>"
+                                     for b in r["where"]] + ["</ul>"]
+                for label, payload in phase_chips(r, c):
+                    H.append(_html_task(_summary_label(label), payload))
+    if folded:
+        H.append('<p class="muted">'
+                 + f"{len(folded)} dormant project(s) with nothing open: "
+                 + " · ".join(f"<code>{_summary_label(k)}</code> "
+                              f"({_summary_label(project_counts(k, c))})"
+                              for k in folded) + "</p>")
+
     H.append(h2("awaiting"))
     for r in c["awaiting"]:
         tone = ("failures", "r") if r["failed_runs"] else (r["state"], "y")
-        H.append(_html_task(phase_head(r, (tone,)), _ruling_payload(r)))
+        chips = phase_chips(r, c)
+        H.append(_html_task(phase_head(r, (tone,)), chips[0][1]))
+        H += [_html_task(f"↳ {_summary_label(label)}", payload)
+              for label, payload in chips[1:]]
     if not c["awaiting"]:
         H.append('<p class="muted">(nothing awaiting a ruling)</p>')
 
@@ -800,7 +1100,10 @@ def phase_head(r, tone_pills=()):
         text = phase_head(r, ((r["state"], "n"),))
         if note:
             text += f'<span class="facets"> — {_summary_label(note)}</span>'
-        H.append(_html_task(text, _live_payload(r, c["projects"])))
+        chips = phase_chips(r, c)
+        H.append(_html_task(text, chips[0][1]))
+        H += [_html_task(f"↳ {_summary_label(label)}", payload)
+              for label, payload in chips[1:]]
     if not c["live"]:
         H.append('<p class="muted">(nothing on the queue)</p>')
 
@@ -909,6 +1212,23 @@ def emit_census(c: dict) -> None:
               "`python3 scripts/cortex.py check`")
 
 
+def emit_by_project(c: dict) -> None:
+    """`census --by-project` — the check-in tree from the checkout alone: no
+    pull, no scoring, no cluster. The same text the door prints at the end of
+    a real check-in, minus the evidence a pull would have added."""
+    shown, folded = by_project_keys(c)
+    print(f"# The Cortex by project — {c['generated']}")
+    print()
+    print(f"{len(shown)} project(s) · {len(c['awaiting'])} awaiting a ruling "
+          f"· {len(c['live'])} live · {len(c['ready'])} ready")
+    print()
+    for key in shown:
+        print("\n".join(project_digest(key, c["projects"][key], c, {})))
+    if folded:
+        print(f"Dormant, nothing open: "
+              + " · ".join(f"{k} ({project_counts(k, c)})" for k in folded))
+
+
 # ---------------------------------------------------------------- collect ---
 # What a pull brought back, scored against the phase's own pre-registered
 # witness. Two facts shape every rule below.
@@ -1071,17 +1391,9 @@ def where_paths(mod, ph, roots: list[Path]) -> list[Path]:
     bullet pointing outside every root is not this project's tree (a RAL path,
     say) and is dropped rather than followed.
     """
-    getter = getattr(mod, "_where_to_look", None)
-    if getter is not None:
-        bullets = getter(ph)
-    else:  # a checkout whose script predates the helper
-        span = mod.sections(ph.text).get("Where to look")
-        bullets = ([ln for ln in ph.text.split("\n")[span[0]:span[1]]
-                    if ln.startswith("- ") and ln.strip() != "-"]
-                   if span else [])
     out = []
-    for bullet in bullets:
-        token = bullet[2:].strip().split()[0].strip("`,.") if bullet[2:].strip() else ""
+    for bullet in _where_bullets(mod, ph):
+        token = bullet.split()[0].strip("`,.")
         if not token:
             continue
         p = Path(token)
@@ -1884,68 +2196,52 @@ def _payload_block(payload: str) -> list[str]:
 
 
 def project_digest(key: str, row: dict, c: dict, scored_by_rel: dict,
-                   pull_line: str) -> list[str]:
-    """One project's block of the check-in summary — where it lives, what came
-    of its pull, and every phase of it a human could act on today, each with
-    the prompt that already exists for that state.
-
-    Phase 3 enriches this (the two missing prompts, the folders); it is keyed
-    by project here because that is the axis the human checks in along — they
-    ask about a project, never about a phase id.
+                   pull_line: str = "") -> list[str]:
+    """One project's block of the by-project tree — where it lives, what came
+    of its pull, and every phase of it a human could act on today: its health
+    from the last scoring, the folders its `## Where to look` names verbatim,
+    and a prompt for each thing that phase could become.
+
+    Keyed by project because that is the axis the human checks in along —
+    they ask about a project, never about a phase id. The same tree is what
+    `census --by-project` prints (no pull line, nothing scored) and what the
+    board renders with copy buttons; this is the text rendering of it.
     """
-    rows = [r for r in c["phases"] if r["project"] == key]
-    counts: dict[str, int] = {}
-    for r in rows:
-        counts[r["state"]] = counts.get(r["state"], 0) + 1
-    L = [f"### {key}", ""]
-    where = [f"local `{row.get('local_path', '?')}`"]
-    if (row.get("mirror") or "none") != "none":
-        where.append(f"mirror `{row['mirror']}`")
-    where.append(f"RAL `{row.get('ral_root', '?')}`")
-    L += ["- " + " · ".join(where),
-          f"- {pull_line}",
-          "- phases: " + (" · ".join(f"{state} {n}" for state, n
-                                     in sorted(counts.items()))
-                          or "none")]
-
-    def _phase_lines(title: str, phase_rows: list, payload) -> list[str]:
-        if not phase_rows:
-            return []
-        out = ["", f"**{title}**", ""]
+    L = [f"### {key}", "", "- " + " · ".join(project_paths(row))]
+    if pull_line:
+        L.append(f"- {pull_line}")
+    L.append(f"- phases: {project_counts(key, c)}")
+    groups = project_groups(key, c)
+    if not groups:
+        L += ["", "_Nothing open — every phase of this project is history._"]
+    for title, phase_rows in groups:
+        L += ["", f"**{title}**", ""]
         for r in phase_rows:
             s = scored_by_rel.get(r["rel"])
-            head = f"- `{r['rel']}` — {r['title']}"
+            head = f"- `{r['rel']}` — {r['title']} — {r['state']}"
             if s:
                 head += f" — {s['health']}"
-            out.append(head)
-            note = _live_note(r)
+            if r["failed_runs"]:
+                head += " — failed runs: " + ", ".join(r["failed_runs"])
+            L.append(head)
+            note = _live_note(r) if r["state"] in LIVE_STATES else ""
             if note:
-                out.append(f"  - {note}")
+                L.append(f"  - {note}")
             if s:
                 legs = ", ".join(f"{k} {s['legs'][k][0]}" for k in LEGS)
-                out.append(f"  - legs: {legs}")
+                L.append(f"  - legs: {legs}")
                 for label, value in (("run dir", s["run_dir"]),
                                      ("zip", s["zip"])):
                     if value is not None:
-                        out.append(f"  - {label}: `{value}`")
+                        L.append(f"  - {label}: `{value}`")
                 for kind in ("out", "err"):
                     for path in s["logs"][kind][:1]:
-                        out.append(f"  - `.{kind}`: `{path}`")
-            out += _payload_block(payload(r))
-        return out
-
-    L += _phase_lines("Awaiting your ruling",
-                      [r for r in c["awaiting"] if r["project"] == key],
-                      _ruling_payload)
-    L += _phase_lines("Still out there",
-                      [r for r in c["live"] if r["project"] == key],
-                      lambda r: _live_payload(r, c["projects"]))
-    L += _phase_lines("Ready to submit",
-                      [r for r in c["ready"] if r["project"] == key],
-                      lambda r: _ready_payload(r, c["projects"]))
-    L += _phase_lines("Gated",
-                      [r for r in c["gated"] if r["project"] == key],
-                      _gate_payload)
+                        L.append(f"  - `.{kind}`: `{path}`")
+            for bullet in r["where"]:
+                L.append(f"  - where to look: {bullet}")
+            for label, payload in phase_chips(r, c):
+                L.append(f"  - **{label}**")
+                L += _payload_block(payload)
     L += [""]
     return L
 
@@ -2117,6 +2413,10 @@ def common(p):
 
     c = common(sub.add_parser("census", help="what the Cortex is holding"))
     c.add_argument("--json", dest="as_json", action="store_true")
+    c.add_argument("--by-project", dest="by_project", action="store_true",
+                   help="the check-in tree instead of the counts: every "
+                        "project's folders, its open phases, the locations "
+                        "they name and the prompt for each")
 
     d = common(sub.add_parser("dashboard", help="render dashboard.md/.html"))
     d.add_argument("--check", action="store_true",
@@ -2230,8 +2530,13 @@ def main(argv=None) -> int:
         return RC_UNREADABLE
 
     if verb == "census":
-        print(json.dumps({k: v for k, v in c.items() if k != "phases"},
-                         indent=2)) if a.as_json else emit_census(c)
+        if a.as_json:
+            print(json.dumps({k: v for k, v in c.items() if k != "phases"},
+                             indent=2))
+        elif a.by_project:
+            emit_by_project(c)
+        else:
+            emit_census(c)
         return RC_OK
 
     if verb == "dashboard":
diff --git a/tests/test_cortex_conductor.py b/tests/test_cortex_conductor.py
index 96142866..0cd14d41 100644
--- a/tests/test_cortex_conductor.py
+++ b/tests/test_cortex_conductor.py
@@ -15,6 +15,7 @@
 
 import importlib.util
 import json
+import re
 import os
 import shutil
 import subprocess
@@ -272,6 +273,191 @@ def _score(board_) -> dict:
             ("11_healthy", "12_resumed", "01_partial")}
 
 
+# --- by project ------------------------------------------------------------
+# The check-in view: one tree, rendered three ways. These assert the tree's
+# promises (the folders, the `## Where to look` bullets, the prompts) rather
+# than its exact prose.
+def test_the_by_project_section_leads_the_board(skeleton):
+    """It is the first section, above the state lists: those are
+    cross-project, this one answers "where is everything" per project."""
+    page = _cortex.render_dashboard(_cortex.census(skeleton))
+    assert page.index("## By project") < page.index("## Awaiting ruling")
+    assert page.index("| Where | Count |") < page.index("## By project")
+
+
+def test_a_project_block_names_its_three_folders_and_its_counts(skeleton):
+    c = _cortex.census(skeleton)
+    page = _cortex.render_dashboard(c)
+    row = c["projects"]["example"]
+    assert "### example" in page
+    for path in (row["local_path"], row["mirror"], row["ral_root"]):
+        assert f"`{path}`" in page
+    # every state of the fixture is counted, history included
+    assert "- phases: accepted 1 · awaiting-ruling 1 · dropped 1" in page
+
+
+def test_the_where_to_look_bullets_render_verbatim(skeleton):
+    """Parsed since phase 1 of the check-in epic, rendered since phase 3 —
+    this is the "which folder do I open" answer."""
+    c = _cortex.census(skeleton)
+    awaiting = [r for r in c["phases"] if r["rel"].endswith("07_awaiting_ruling.md")][0]
+    assert awaiting["where"] == ["`/mnt/c/Users/Jammy/Science/example/output/phase_07/`"]
+    assert ("- where to look: `/mnt/c/Users/Jammy/Science/example/output/phase_07/`"
+            in _cortex.render_dashboard(c))
+
+
+def test_the_history_states_are_counted_not_listed(skeleton):
+    """`accepted` / `rerun` / `dropped` are history: they belong to the
+    counts line and to the ruling ledger, not to a list of things to do."""
+    c = _cortex.census(skeleton)
+    groups = dict(_cortex.project_groups("example", c))
+    listed = {r["state"] for rows in groups.values() for r in rows}
+    assert listed == {"awaiting-ruling", "pulled", "submitted", "running",
+                      "ready", "gated", "planned"}
+
+
+def test_census_by_project_prints_the_same_tree(skeleton):
+    r = _run(["census", "--cortex", str(skeleton), "--by-project"])
+    assert r.returncode == _cortex.RC_OK, r.stdout + r.stderr
+    assert "# The Cortex by project" in r.stdout
+    assert "### example" in r.stdout
+    assert "- where to look: `/mnt/c/Users/Jammy/Science/example/output/phase_07/`" in r.stdout
+    assert "**rule on it**" in r.stdout
+    # the counts census is still the default
+    assert "== Cortex census ==" in _run(["census", "--cortex", str(skeleton)]).stdout
+
+
+def test_a_dormant_project_with_nothing_open_folds_to_one_line(tmp_skeleton):
+    (tmp_skeleton / "projects.yaml").write_text(
+        (tmp_skeleton / "projects.yaml").read_text(encoding="utf-8")
+        + "\ndormant_one:\n  remote: none\n"
+          "  local_path: /mnt/c/Users/Jammy/Science/dormant\n"
+          "  ral_root: /mnt/ral/jnightin/dormant\n  mirror: none\n"
+          "  sync_cli: hpc/sync\n  sync_verbs: [pull]\n"
+          "  ledger: wiki/state.md\n  witness_file: out/**/*.json\n"
+          "  partition: gpu\n  status: dormant\n", encoding="utf-8")
+    c = _cortex.census(tmp_skeleton)
+    shown, folded = _cortex.by_project_keys(c)
+    assert shown == ["example"] and folded == ["dormant_one"]
+    page = _cortex.render_dashboard(c)
+    assert "### dormant_one" not in page
+    assert "1 dormant project(s) with nothing open" in page
+    assert "`dormant_one` (none)" in page
+
+
+def test_a_phase_that_names_nowhere_still_renders(tmp_skeleton):
+    """A `planned` phase may still be carrying the template placeholder —
+    `check` exempts it — so every rendering has to survive an empty list."""
+    scope = tmp_skeleton / "phases" / "example" / "01_scope.md"
+    text = scope.read_text(encoding="utf-8")
+    scope.write_text(re.sub(r"## Where to look\n\n(?:- .*\n)+",
+                            "## Where to look\n\n", text, count=1),
+                     encoding="utf-8")
+    mod = _cortex.load_cortex(tmp_skeleton)
+    assert mod.check_problems(tmp_skeleton) == []
+    c = _cortex.census(tmp_skeleton)
+    row = [r for r in c["phases"] if r["rel"].endswith("01_scope.md")][0]
+    assert row["where"] == []
+    page = _cortex.render_dashboard(c)
+    assert "### example" in page and "where to look: \n" not in page
+    _cortex.render_dashboard_html(c)          # must not raise either
+    text = "\n".join(_cortex.project_digest("example", c["projects"]["example"],
+                                            c, {}))
+    assert "01_scope.md" in text
+
+
+# --- the two missing prompts ----------------------------------------------
+def _synthetic(*rows) -> dict:
+    """A census-shaped dict — the payloads read only `phases` + `projects`."""
+    return {"phases": list(rows),
+            "projects": {"proj": {"local_path": "/s/proj", "sync_cli": "hpc/sync",
+                                  "sync_verbs": ["pull", "submit"]}}}
+
+
+def _row(number, state, slug, epic=None):
+    return {"rel": f"phases/proj/{slug}.md", "slug": slug, "title": slug,
+            "project": "proj", "phase": number, "state": state, "epic": epic,
+            "runs": [], "budget": None, "budget_minutes": None,
+            "wall_minutes": 0, "review_minutes": None, "gates": [],
+            "failed_runs": [], "where": []}
+
+
+def test_accept_and_open_opens_the_planned_sibling(skeleton):
+    """A planned phase N+1 IS the ledger naming the next phase — the prompt
+    opens it rather than writing a second one beside it."""
+    c = _synthetic(_row(4, "awaiting-ruling", "four"), _row(5, "planned", "five"))
+    payload = _cortex._next_phase_payload(c["phases"][0], c)
+    assert "rule phases/proj/four.md accept --body <file>" in payload
+    assert "move phases/proj/five.md ready" in payload
+    assert "cd /s/proj && hpc/sync submit <script>" in payload
+    assert "cortex.py new" not in payload
+
+
+def test_accept_and_open_writes_a_new_phase_when_there_is_none(skeleton):
+    c = _synthetic(_row(4, "awaiting-ruling", "four", epic="an-epic"))
+    payload = _cortex._next_phase_payload(c["phases"][0], c)
+    assert "new proj <slug> --phase 5 --epic an-epic" in payload
+    # the title-prefix convention, stated where it is used: `new` writes
+    # `# <Project> — phase N: ` itself, so `--title` takes the tail alone.
+    assert "the tail only" in payload and "phase 5: ' itself" in payload
+
+
+def test_accept_and_open_skips_a_number_that_is_already_taken(skeleton):
+    """Phase numbers are unique per project, and a phase that has already run
+    cannot be moved back to `ready` — so N+1 being taken means a new number."""
+    c = _synthetic(_row(4, "awaiting-ruling", "four"), _row(5, "accepted", "five"),
+                   _row(9, "dropped", "nine"))
+    payload = _cortex._next_phase_payload(c["phases"][0], c)
+    assert "phase 5 is taken (phases/proj/five.md, accepted)" in payload
+    assert "--phase 10" in payload
+
+
+def test_run_it_again_is_a_ruling_then_the_same_launch(skeleton):
+    c = _synthetic(_row(4, "awaiting-ruling", "four"))
+    payload = _cortex._rerun_payload(c["phases"][0], c)
+    order = [payload.index(x) for x in (
+        "rule phases/proj/four.md rerun --body <file>",
+        "move phases/proj/four.md ready",
+        "cd /s/proj && hpc/sync submit <script>",
+        "move phases/proj/four.md submitted --run <jobid>")]
+    assert order == sorted(order), payload
+    assert "one run per call" in payload
+
+
+def test_a_project_with_no_submit_verb_says_so_rather_than_inventing_one(skeleton):
+    c = _synthetic(_row(4, "awaiting-ruling", "four"))
+    c["projects"]["proj"]["sync_verbs"] = ["pull"]
+    assert "no `submit` verb in projects.yaml" in _cortex._rerun_payload(
+        c["phases"][0], c)
+
+
+def test_the_chips_a_state_carries_are_the_same_everywhere(skeleton):
+    """One table, three renderings — the board, `--by-project` and the door
+    all ask `phase_chips`, so a prompt cannot appear in one and not another."""
+    c = _cortex.census(skeleton)
+    by_state = {r["state"]: [label for label, _p in _cortex.phase_chips(r, c)]
+                for r in c["phases"]}
+    assert by_state["awaiting-ruling"] == [
+        "rule on it", "the results are good — accept and open phase 8",
+        "run it again"]
+    assert by_state["pulled"][0] == "rule on it"
+    assert by_state["running"] == ["where the jobs stand", "run it again"]
+    assert by_state["submitted"] == ["where the jobs stand"]
+    assert by_state["ready"] == ["submit it"]
+    assert by_state["gated"] == ["its gates"]
+    assert by_state["planned"] == ["open it"]
+
+
+def test_the_new_prompts_ride_beside_the_rule_prompt_on_the_board(skeleton):
+    """"Beside", not "instead of": the awaiting row keeps the rule payload it
+    has always had and the two new ones follow it as their own chips."""
+    page = _cortex.render_dashboard(_cortex.census(skeleton))
+    assert "📋 ↳ the results are good — accept and open phase 8" in page
+    assert "📋 ↳ run it again" in page
+    html = _cortex.render_dashboard_html(_cortex.census(skeleton))
+    assert "↳ run it again" in html
+
+
 def test_the_board_the_collect_tests_score_is_a_tree_that_checks(board):
     """Every fixture below asserts about scoring, not about schema — so the
     tree they score must be one `cortex.py check` accepts."""
@@ -609,6 +795,9 @@ def test_the_summary_is_keyed_by_project_and_is_the_last_thing_printed(board):
     assert "Awaiting your ruling" in out
     # the prompt each state already has, ready to paste
     assert "help me rule on it" in out
+    # ... and the two phase 3 added, in the same tree
+    assert "accept and open phase" in out and "needs running again" in out
+    assert "where to look:" in out
     assert f"local `{board['sub']}`" in out