Phase 16: the flagship tree — receipts-first reading

CHANGELOG.md

@@ -7,6 +7,25 @@ holds the full story-by-story evidence trail, and the version below is
 single-sourced from `dw_pmo.__version__` (test-asserted against
 `dw --version`, the plugin manifest, and this file).
 
+## Unreleased
+
+The flagship tree: receipts-first reading. The read layer meets a
+decade-shaped legacy roadmap where it is — story tables are
+recognized by their header cells (the Evidence column optional),
+decorated statuses (`**done** (2026-07-07 — …)`, `CLOSED ✅ (6/6)`)
+normalize to comparable tokens at word boundaries (`host-complete`
+never reads as complete), evidence pairs against the
+`evidence-story-NN.md` receipts on disk rather than table prose,
+struck-through rows are retired history with no demands, table-less
+phases read file-derived under one legible warning, the README's
+Current-phase pointer names `current_phase` in the state feed, and
+`dw next` never proposes work from a phase holding a final-summary.
+The write gate is untouched and exactly as strict. Proven against
+the flagship consumer's real 86-phase tree: 397 reported errors
+fall to 31, every survivor a real desync.
+Full detail:
+[phase 16](./pmo-roadmap/pm/roadmap/work-log-automation/phase-16-flagship-tree/current-phase-status.md).
+
 ## v1.12.0 — 2026-07-05
 
 The fifth window: the local `dw-workbench` browser gains a

pmo-roadmap/lib/dw_pmo/__init__.py

@@ -10,6 +10,7 @@
 from __future__ import annotations
 
 from .model import (
+    CUT_STATUSES,
     DONE_STATUSES,
     OPEN_STATUSES,
     PHASE_RE,
@@ -20,6 +21,7 @@
     Project,
     StoryRow,
     die,
+    normalize_status,
 )
 from .paths import (
     ensure_under,

pmo-roadmap/lib/dw_pmo/api.py

@@ -4,13 +4,14 @@
 
 from pathlib import Path
 
-from .model import DONE_STATUSES, OPEN_STATUSES, Phase, Project, StoryRow
+from .model import DONE_STATUSES, OPEN_STATUSES, Phase, Project, StoryRow, normalize_status
 from .parse import (
     discover_phases,
     get_phase,
     header_status,
     hook_snapshot,
     link_target,
+    parse_current_phase_dirname,
     parse_current_phase_target,
     parse_story_rows,
     story_num_from_file,
@@ -23,10 +24,21 @@
 
 def next_story(project: Project, root: Path) -> dict[str, object] | None:
     preferred = ("in-progress", "ready", "backlog")
+    # Nothing in a closed phase is actionable (WLA-16-03) — a
+    # final-summary is the phase's terminal receipt. Within each status
+    # tier, the phase the README pointer names is consulted first.
+    phases = [
+        phase
+        for phase in discover_phases(project)
+        if not (phase.path / "final-summary.md").exists()
+    ]
+    pointer_dir = parse_current_phase_dirname(project)
+    if pointer_dir:
+        phases.sort(key=lambda phase: phase.path.name != pointer_dir)
     for status in preferred:
-        for phase in discover_phases(project):
+        for phase in phases:
             for row in parse_story_rows(phase.path / "current-phase-status.md"):
-                if row.status == status:
+                if normalize_status(row.status) == status:
                     story_target = link_target(row.story_file)
                     return {
                         "story_id": row.story_id,
@@ -108,7 +120,7 @@ def project_context(
         all_rows = parse_story_rows(phase.path / "current-phase-status.md")
         rows = []
         for row in all_rows:
-            if status_filter and row.status != status_filter:
+            if status_filter and normalize_status(row.status) != normalize_status(status_filter):
                 continue
             rows.append(story_context(row, phase, project, root, include_trace))
         phase_items.append(
@@ -120,7 +132,7 @@ def project_context(
                 "status_file_exists": (phase.path / "current-phase-status.md").exists(),
                 "final_summary": rel(phase.path / "final-summary.md", root),
                 "final_summary_exists": (phase.path / "final-summary.md").exists(),
-                "active": any(row.status in OPEN_STATUSES for row in all_rows),
+                "active": any(normalize_status(row.status) in OPEN_STATUSES for row in all_rows),
                 "stories": rows,
             }
         )
@@ -212,10 +224,10 @@ def story_timeline(row: StoryRow, phase: Phase, project: Project, root: Path) ->
     events.sort(key=lambda e: str(e["sort_key"]), reverse=True)
     status = str(context["status"])
     evidence_exists = bool(context["evidence_exists"])
-    shipped = status in DONE_STATUSES and evidence_exists
+    shipped = normalize_status(status) in DONE_STATUSES and evidence_exists
     reason = ""
     if not shipped:
-        if status not in DONE_STATUSES:
+        if normalize_status(status) not in DONE_STATUSES:
             reason = f"story status is {status!r}, not done"
         elif not evidence_exists:
             reason = "story is marked done but its evidence file does not exist"

pmo-roadmap/lib/dw_pmo/model.py

@@ -18,8 +18,59 @@
 STORY_OPEN_STATUSES = {"backlog", "ready", "in-progress", "blocked"}
 STORY_STATUSES = DONE_STATUSES | STORY_OPEN_STATUSES
 # Phase-activity detection tolerates the looser phase-index vocabulary
-# (planned/not-started) that may appear in legacy story tables.
-OPEN_STATUSES = STORY_OPEN_STATUSES | {"planned", "not-started"}
+# (planned/not-started/planning/scaffolded/paused) that may appear in
+# legacy story tables and phase indexes.
+OPEN_STATUSES = STORY_OPEN_STATUSES | {"planned", "not-started", "planning", "scaffolded", "paused"}
+
+# Read-side terminal statuses for rows a legacy tree retired rather
+# than shipped (struck-through rows, cancelled work). Never accepted
+# by the write vocabulary above.
+CUT_STATUSES = {"cut", "cancelled", "superseded"}
+
+# ── read-side status normalization (WLA-16-01) ────────────────────────
+#
+# Legacy trees decorate statuses: "**done** (2026-07-07 — twelve new
+# tests)", "CLOSED ✅ (6/6)", "in-progress (3/6)". The read layer
+# resolves any such cell or header to one comparable token so that
+# membership tests and header↔table comparisons see through the
+# decoration. Matching happens at token boundaries, leftmost match
+# wins, longest keyword first at a given position — "host-complete"
+# must never read as "complete". The WRITE vocabulary stays
+# STORY_STATUSES, exact: `dw story status` rejects anything else.
+
+_STATUS_KEYWORDS = [
+    "in-progress", "in progress", "not-started", "not started",
+    "superseded", "cancelled", "scaffolded", "planning", "planned",
+    "complete", "shipped", "backlog", "blocked", "paused", "closed",
+    "ready", "done", "cut",
+]
+_STATUS_KEYWORD_RE = re.compile(
+    r"(?<![a-z0-9-])("
+    + "|".join(k.replace(" ", r"\s+") for k in sorted(_STATUS_KEYWORDS, key=len, reverse=True))
+    + r")(?![a-z0-9-])"
+)
+_STATUS_CANONICAL = {"in progress": "in-progress", "not started": "not-started"}
+
+
+def normalize_status(raw: str | None) -> str:
+    """Resolve a possibly-decorated status string to a comparable token.
+
+    Only the HEAD of the string — everything before the first
+    decoration delimiter — is consulted: a keyword found there wins
+    (canonicalized); otherwise the head's first token, lowercased.
+    Narrative tails are never searched — a cell ending "…the request
+    never shipped)" must not read as shipped. Read-side only.
+    """
+    if not raw:
+        return ""
+    s = re.sub(r"\*|`|_{2}|~~", "", raw).strip().lower()
+    head = re.split(r"[(—–:;,.!]", s, maxsplit=1)[0].strip()
+    m = _STATUS_KEYWORD_RE.search(head)
+    if m:
+        token = re.sub(r"\s+", " ", m.group(1))
+        return _STATUS_CANONICAL.get(token, token)
+    parts = head.split()
+    return parts[0] if parts else ""
 
 # The generator's stand-in body for evidence created without content.
 # dw check treats a done story whose evidence still carries this line
@@ -68,3 +119,14 @@ class StoryRow:
     status: str
     story_file: str
     evidence: str
+
+
+def row_is_retired(row: StoryRow) -> bool:
+    """A retired row is legacy history that will never ship: the ID is
+    struck through (`~~HS-1-01~~`) or the status normalizes to a cut
+    token. Read-side validators make no file or evidence demands of it
+    (WLA-16-02)."""
+    story_id = row.story_id.strip()
+    if story_id.startswith("~~") and story_id.endswith("~~") and len(story_id) > 4:
+        return True
+    return normalize_status(row.status) in CUT_STATUSES

pmo-roadmap/lib/dw_pmo/parse.py

@@ -5,7 +5,7 @@
 import re
 from pathlib import Path
 
-from .model import PHASE_RE, STORY_RE, Phase, Project, StoryRow, die
+from .model import PHASE_RE, STORY_ID_RE, STORY_RE, Phase, Project, StoryRow, die
 from .paths import read_text, rel, roadmap_dir, strip_code
 
 
@@ -72,25 +72,58 @@ def split_table_row(line: str) -> list[str]:
     return [cell.strip() for cell in stripped.strip("|").split("|")]
 
 
+def _story_table_columns(cells: list[str]) -> dict[str, int] | None:
+    """Map story-table columns from a header row, or None if the row is
+    not a story-table header. Requires ID/Story/Status/Story file
+    (case-insensitive); Evidence is optional (legacy 4-column dialect,
+    WLA-16-01). The canonical 5-column header maps to the same indices
+    as the historical fixed-position parse."""
+    lowered = [re.sub(r"\s+", " ", cell.strip().lower()) for cell in cells]
+    index = {name: i for i, name in enumerate(lowered)}
+    required = ("id", "story", "status", "story file")
+    if not all(name in index for name in required):
+        return None
+    return {
+        "story_id": index["id"],
+        "title": index["story"],
+        "status": index["status"],
+        "story_file": index["story file"],
+        "evidence": index.get("evidence", -1),
+    }
+
+
+def _is_separator_row(cells: list[str]) -> bool:
+    return bool(cells) and all(re.fullmatch(r":?-{2,}:?|-", cell) for cell in cells)
+
+
 def parse_story_rows(status_file: Path) -> list[StoryRow]:
     if not status_file.exists():
         return []
     rows: list[StoryRow] = []
-    in_table = False
+    columns: dict[str, int] | None = None
     for line in read_text(status_file).splitlines():
-        if line.startswith("| ID | Story | Status | Story file | Evidence |"):
-            in_table = True
-            continue
-        if not in_table:
-            continue
-        if line.startswith("|---"):
+        cells = split_table_row(line)
+        if columns is None:
+            if cells:
+                columns = _story_table_columns(cells)
             continue
-        if not line.startswith("|"):
+        if not line.strip().startswith("|"):
             break
-        cells = split_table_row(line)
-        if len(cells) != 5:
+        if not cells or _is_separator_row(cells):
+            continue
+        width = max(v for v in columns.values())
+        if len(cells) <= width:
             continue
-        rows.append(StoryRow(*cells))
+        evidence_idx = columns["evidence"]
+        rows.append(
+            StoryRow(
+                cells[columns["story_id"]],
+                cells[columns["title"]],
+                cells[columns["status"]],
+                cells[columns["story_file"]],
+                cells[evidence_idx] if 0 <= evidence_idx < len(cells) else "",
+            )
+        )
     return rows
 
 
@@ -115,6 +148,35 @@ def header_status(path: Path) -> str | None:
     return None
 
 
+def story_id_from_header(path: Path) -> str:
+    """The story ID from a story file's H1 (`# FX-85-01 - Title`), or ""
+    when the file or a well-formed ID is absent. A receipt-side
+    identity for stories no table row covers (WLA-16-02)."""
+    if not path.exists():
+        return ""
+    lines = read_text(path).splitlines()
+    if not lines:
+        return ""
+    first = lines[0].lstrip("#").strip()
+    for sep in (" — ", " - "):
+        if sep in first:
+            candidate = first.split(sep, 1)[0].strip()
+            if STORY_ID_RE.match(candidate):
+                return candidate
+    return ""
+
+
+def phase_story_files(phase_path: Path) -> dict[int, Path]:
+    """On-disk story files by number — the receipts, independent of any
+    table (WLA-16-02)."""
+    files: dict[int, Path] = {}
+    for path in sorted(phase_path.glob("story-*.md")):
+        m = STORY_RE.match(path.name)
+        if m:
+            files[int(m.group(1))] = path
+    return files
+
+
 def story_title(path: Path) -> str:
     if not path.exists():
         return path.stem
@@ -145,6 +207,19 @@ def parse_current_phase_target(project: Project) -> str:
     return ""
 
 
+def parse_current_phase_dirname(project: Project) -> str:
+    """The phase directory the README's Current-phase pointer names, or
+    "" when absent/unresolvable. The methodology's own current-phase
+    receipt (WLA-16-03)."""
+    target = parse_current_phase_target(project)
+    if not target:
+        return ""
+    for part in Path(target).parts:
+        if PHASE_RE.match(part):
+            return part
+    return ""
+
+
 def current_phase_status_path(project: Project) -> Path | None:
     target = parse_current_phase_target(project)
     if not target:

pmo-roadmap/lib/dw_pmo/sessions.py

@@ -24,6 +24,7 @@
 from datetime import datetime, timezone
 from pathlib import Path
 
+from .model import normalize_status
 from .parse import discover_phases, discover_projects, parse_story_rows
 
 SESSIONS_SCHEMA = 1
@@ -80,7 +81,7 @@ def _in_progress_stories(root: Path) -> list[dict] | None:
                 for row in parse_story_rows(
                     phase.path / "current-phase-status.md"
                 ):
-                    if row.status in _IN_PROGRESS:
+                    if normalize_status(row.status) in _IN_PROGRESS:
                         stories.append(
                             {
                                 "story_id": row.story_id,