feat(flow): Stacks on — auto-stacked dependency chains in one run (#123)

A single `flow` over a merge-gated, file-overlapping chain couldn't complete in one
run: `Depends on (merged)` waits for a real merge, and a flow run only opens draft PRs,
so the chain advanced only across separate invocations interleaved with human merges.

Add an opt-in `Stacks on: <id>…` field: a dependent builds on its prerequisite's
*just-produced branch* within the SAME run and publishes a separate stacked PR
(`gh pr create --base <prereq-branch>`), so a planned refactor sequence (203 → 207 → 204)
completes as a reviewable PR stack in one `flow` invocation. Keep `Depends on (merged)`
(the conservative multi-run mode) intact.

- brief.stacks_on() parses the field (shared _id_list, like depends_on_merged).
- flow: _declared_deps includes stacks_on (DAG validation); _stacked_snapshot gates a
  dependent on its prereq being COMPLETE-with-a-published-branch (not merged), once per
  pass like _merged_snapshot; _deps_met honours it. A stack prerequisite publishes
  in-loop (the moment it's COMPLETE) — not the deferred end-publish — so a dependent's
  next pass can base on its branch; _publish_bundle dedups against the end-publish.
- worktree: a stacked dependent's Do worktree bases off origin/<prereq-branch> (resolved
  from the prereq's publish.json), not the target base, so Do builds on the prereq's diff.
- publish: a `Stacks on` bundle cuts its branch off the parent branch and targets the PR
  at it (`--base <parent-branch>`); publish.json records mode "stacked-pr" + stacks_on.
  Warns (best-effort, gh) if the repo disallows merge commits (a stack must merge-commit
  bottom-up, not squash).
- docs 09 + brief field document Stacks on vs Depends on (merged), the bottom-up
  merge-commit rule, and the cross-run rebuild after a parent's branch changes.

Scope note (settled with the maintainer): under publish-on-accept + eligibility waiting
for COMPLETE, a prereq finishes ALL its iterations before its dependent ever builds, so
the in-run "auto-rebuild the stack above on iterate" can't fire — the dependent always
builds on the prereq's final branch. The cross-run rebuild (e.g. after a squash-merge)
is the documented mechanism; no in-run cascade code is needed.

Tests: stacks_on parsing; stacked-PR publish targets the parent branch + errors before
the parent publishes; worktree bases off the parent branch (real git); a stacked chain
completes in one flow_ids run (parent built+published before the dependent). 200 OK;
lint_docs OK.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Signed-off-by: Eduard Ralph <15236434+eduralph@users.noreply.github.com>

Author: eduralph

template/PCDA/quality-cycle/09-parallel-lanes.md

@@ -58,6 +58,7 @@ Manual wave-splitting (run a prerequisite batch to COMPLETE, *then* the next) en
 
 - **`- **Depends on:** <id>[, <id>…]`** — a topological gate. The in-driver pool dispatches a bundle only once every declared prerequisite is **COMPLETE** (signed off, not merely built). Because a prereq reaches COMPLETE only after its sign-off in an earlier pass, a dependent waits across passes — exactly the manual wave plan, now machine-enforced.
 - **`- **Depends on (merged):** <id>[, <id>…]`** — a *stricter* gate for a dependent that **edits files a prerequisite also edits**. `Depends on` waits only for COMPLETE — which means "a draft PR was opened", **not merged**. A dependent's Do runs in a worktree off the target base (`origin/<base>`), so a prereq whose PR is still open is *absent* from that base: the dependent is built without the predecessor's diff and its PR conflicts at merge. This field holds the dependent until the prereq's PR is **merged into the base** (read from the prereq's recorded PR via `gh pr view`), so Do genuinely builds on the merged result. Because the flow only *opens* draft PRs (it never merges), a merge-gated dependent is **held across `pdca flow` runs**: its prereq is published in one run, a human merges that PR, and a later `pdca flow <dependent>` finds it merged and proceeds. Best-effort and fail-closed: anything not confirmable as merged keeps the dependent safely blocked.
+- **`- **Stacks on:** <id>[, <id>…]`** — auto-stacked chains (issue #123). Like `Depends on (merged)` it is for a dependent that **edits files a prerequisite also edits**, but instead of *waiting for the prereq to merge* (multi-run), it builds the dependent on the prereq's **just-produced branch** within the **same `flow` run** and publishes a **separate stacked PR** (`gh pr create --base <prereq-branch>`, one PR per item showing only that item's increment). A stacked dependent is eligible once its prereq is **COMPLETE with a published branch**; its Do worktree bases off `origin/<prereq-branch>` and its PR targets that branch — the base is **derived from the prereq's `publish.json`**, never written in the brief (it doesn't exist at Plan time). So a planned, file-overlapping refactor sequence (`203 → 207 → 204`) completes as a reviewable PR stack in **one invocation** — independents still run in parallel; a `Stacks on` chain is sequential within itself. **Merge the stack bottom-up with merge-commit / rebase-merge, not squash** (a squash drops the parent's commits, so a child retargeted to the base re-shows the parent's diff until rebased); publish warns if the target repo disallows merge commits. If a parent's branch later changes (e.g. a squash-merge between runs), rebuild the items stacked above it (`signoff --iterate-do`) on the new foundation. Use `Depends on (merged)` instead when you'd rather wait for each PR to merge before the next builds.
 - **`- **Conflicts with:** <id>[, <id>…]`** — a same-wave exclusion. Two bundles that touch a shared resource (e.g. both edit one `ci.yml`) are **never in flight in the same concurrent wave**; the pool serializes them across lanes while still parallelizing everything else.
 
 The fields are **additive and backwards-compatible**: with none declared, every bundle is always eligible and dispatch is byte-for-byte the prior **sort-by-name pool**. An unschedulable graph — a cycle, or a dependency that is neither in the batch nor an already-COMPLETE bundle — is a **hard error rejected before any build** (`pdca flow` aborts up front). `pdca status` shows a `[blocked-by: <ids>]` flag so the queue reads as a DAG, not a flat list. Declared ordering complements lane planning: planning *avoids* integration tangling by code locality; `depends_on` / `conflicts_with` *enforce* the residual ordering that locality cannot express.

template/src/pdca_harness/brief.py

@@ -118,6 +118,20 @@ def conflicts_with(brief_path: Path) -> list[str]:
     return _id_list(field(brief_path, "conflicts with", "conflicts_with"))
 
 
+def stacks_on(brief_path: Path) -> list[str]:
+    """Issue ids whose just-produced branch this bundle stacks on (issue #123).
+
+    The optional ``- **Stacks on:** <id>[, <id>…]`` field: build this bundle on top of a
+    prerequisite's *produced patch branch* within the SAME ``flow`` run — not waiting for
+    a merge (unlike ``Depends on (merged)``) — and publish it as a separate stacked PR
+    (``gh pr create --base <prereq-branch>``). Use for a planned, file-overlapping refactor
+    sequence so the whole chain completes in one run. Names the immediate parent(s); the
+    worktree + PR base derive from the parent's ``publish.json`` (never hand-written — the
+    branch doesn't exist at Plan time). Absent ⇒ ``[]``.
+    """
+    return _id_list(field(brief_path, "stacks on", "stacks_on"))
+
+
 def onto_branch(brief_path: Path) -> tuple[str, str] | None:
     """``(remote, branch)`` of an existing PR's head to stack a commit onto, or ``None``.
 

template/src/pdca_harness/flow.py

@@ -199,9 +199,33 @@ def flow(
 # byte-for-byte today's sort-by-name pool.
 # ----------------------------------------------------------------------------
 def _declared_deps(bp: Path) -> list[str]:
-    """All declared prerequisite ids — COMPLETE-gated (`Depends on`) and merge-gated
-    (`Depends on (merged)`, #107) — for DAG validation and dispatch."""
-    return brief.depends_on(bp) + brief.depends_on_merged(bp)
+    """All declared prerequisite ids — COMPLETE-gated (`Depends on`), merge-gated
+    (`Depends on (merged)`, #107) and stack-gated (`Stacks on`, #123) — for DAG
+    validation and dispatch."""
+    return brief.depends_on(bp) + brief.depends_on_merged(bp) + brief.stacks_on(bp)
+
+
+def _prereq_published(cfg: Config, dep_id: str) -> bool:
+    """True iff prereq ``dep_id`` is COMPLETE and has a published branch (issue #123) — the
+    foundation a ``Stacks on`` dependent builds + publishes on top of."""
+    d = cfg.bundle(dep_id)
+    if state.state(d) != state.COMPLETE:
+        return False
+    rec = publish._publish_record(d)
+    return bool(rec and rec.get("branch"))
+
+
+def _stacked_snapshot(cfg: Config, bundles: list[Path]) -> set[str]:
+    """Prereq ids whose ``Stacks on`` foundation is ready now — COMPLETE with a published
+    branch (issue #123). A stacked dependent is eligible once its parent has produced a
+    branch to build on (not waiting for a *merge*, unlike `Depends on (merged)`), so the
+    whole chain completes in one run. Computed once per pass, like :func:`_merged_snapshot`."""
+    wanted: set[str] = set()
+    for b in bundles:
+        bp = b / "brief.md"
+        if bp.exists():
+            wanted.update(brief.stacks_on(bp))
+    return {dep for dep in wanted if _prereq_published(cfg, dep)}
 
 
 def _merged_snapshot(cfg: Config, bundles: list[Path]) -> set[str]:
@@ -221,19 +245,21 @@ def _merged_snapshot(cfg: Config, bundles: list[Path]) -> set[str]:
     return {dep for dep in wanted if merged.is_merged(cfg, dep)}
 
 
-def _deps_met(cfg: Config, d: Path, merged_ids: set[str]) -> bool:
+def _deps_met(cfg: Config, d: Path, merged_ids: set[str], stacked_ids: set[str]) -> bool:
     """True iff every prerequisite ``d`` declares is satisfied.
 
     ``Depends on`` prereqs must be COMPLETE; ``Depends on (merged)`` prereqs must be in
-    ``merged_ids`` (this pass's merged set, #107). An unplanned/reopened bundle (no brief
-    yet) declares nothing, so it is eligible; its deps are honoured once it is re-planned.
+    ``merged_ids`` (this pass's merged set, #107); ``Stacks on`` prereqs must be in
+    ``stacked_ids`` (COMPLETE-with-a-published-branch, #123). An unplanned/reopened bundle
+    (no brief yet) declares nothing, so it is eligible; its deps are honoured on re-plan.
     """
     bp = d / "brief.md"
     if not bp.exists():
         return True
     return (all(state.state(cfg.bundle(dep)) == state.COMPLETE
                 for dep in brief.depends_on(bp))
-            and all(dep in merged_ids for dep in brief.depends_on_merged(bp)))
+            and all(dep in merged_ids for dep in brief.depends_on_merged(bp))
+            and all(dep in stacked_ids for dep in brief.stacks_on(bp)))
 
 
 def _conflict_map(cfg: Config, bundles: list[Path]) -> dict[str, set[str]]:
@@ -334,33 +360,38 @@ def _build_all(cfg: Config, bundles: list[Path]) -> None:
     Then advance one beat per still-running, deps-met bundle per round until nothing
     progresses: serial by default, or fanned across ``cfg.lanes`` workers when configured.
     """
-    # Merge-gate snapshot once per pass (#107) — keeps the gh merge check out of the beat
-    # sweep / lane-pool dispatch; threaded into _deps_met below.
+    # Eligibility snapshots once per pass — keep the gh merge check (#107) and the
+    # stacked-branch check (#123) out of the beat sweep / lane-pool dispatch; threaded into
+    # _deps_met below.
     merged_ids = _merged_snapshot(cfg, bundles)
+    stacked_ids = _stacked_snapshot(cfg, bundles)
     # Serial Plan pre-pass — the interactive Plan beat must never enter the sweep/pool.
     for d in bundles:
         _isolate(d, "plan", lambda d=d: _plan_if_unplanned(cfg, d, None))
     if cfg.lanes <= 1 or len(bundles) <= 1:
-        _beat_sweep_serial(cfg, bundles, merged_ids)
+        _beat_sweep_serial(cfg, bundles, merged_ids, stacked_ids)
     else:
-        _beat_sweep_pooled(cfg, bundles, merged_ids)
+        _beat_sweep_pooled(cfg, bundles, merged_ids, stacked_ids)
 
 
-def _beat_sweep_serial(cfg: Config, bundles: list[Path], merged_ids: set[str]) -> None:
+def _beat_sweep_serial(cfg: Config, bundles: list[Path], merged_ids: set[str],
+                       stacked_ids: set[str]) -> None:
     """Round-robin one beat per still-running, deps-met bundle (sort-by-name) until no
     bundle progresses — so the wave advances all Dos, then all Checks, then all assembles.
     A dep-blocked bundle simply isn't advanced (a later pass picks it up once its prereq
-    is COMPLETE / merged); a bundle whose beat raises drops out (isolated)."""
+    is COMPLETE / merged / published); a bundle whose beat raises drops out (isolated)."""
     while True:
         progressed = False
         for d in bundles:
-            if _running(d) and _deps_met(cfg, d, merged_ids) and _advance_one(cfg, d):
+            if (_running(d) and _deps_met(cfg, d, merged_ids, stacked_ids)
+                    and _advance_one(cfg, d)):
                 progressed = True
         if not progressed:
             return
 
 
-def _beat_sweep_pooled(cfg: Config, bundles: list[Path], merged_ids: set[str]) -> None:
+def _beat_sweep_pooled(cfg: Config, bundles: list[Path], merged_ids: set[str],
+                       stacked_ids: set[str]) -> None:
     """Pooled beat sweep: each round fans one beat across ``min(lanes, n)`` lane-pinned
     workers, conflict-aware, then **joins (a barrier per beat)** before the next round.
 
@@ -373,7 +404,8 @@ def _beat_sweep_pooled(cfg: Config, bundles: list[Path], merged_ids: set[str]) -
     n_lanes = min(cfg.lanes, len(bundles))
     slot_of: dict[str, int] = {}  # bundle name → its fixed lane slot (worktree affinity)
     while True:
-        eligible = [d for d in bundles if _running(d) and _deps_met(cfg, d, merged_ids)]
+        eligible = [d for d in bundles
+                    if _running(d) and _deps_met(cfg, d, merged_ids, stacked_ids)]
         if not eligible:
             return
         if not _run_beat_round_pooled(cfg, eligible, conflicts, slot_of, n_lanes):
@@ -422,6 +454,17 @@ def run_slot(slot: int, ds: list[Path]) -> None:
     return progressed[0]
 
 
+def _publish_bundle(cfg: Config, d: Path, *, by: str, today: str) -> None:
+    """Publish one COMPLETE bundle (Check's closing step), isolated so a single failure
+    can't abort the batch (testbed #3); a non-zero return is loud, never silent (#97)."""
+    rc = _isolate(d, "publish", lambda: publish.publish(
+        cfg, d.name.removeprefix("issue_"),
+        dry_run=cfg.publisher.mode == "stub", by=by, today=today, skip_if_no_target=True))
+    if rc not in (0, None):  # None ⇒ _isolate already logged an exception
+        print(f"flow: {d.name} is COMPLETE but publish did not complete (rc {rc}) — NOT "
+              f"published; run `pdca publish {d.name.removeprefix('issue_')}`.", file=sys.stderr)
+
+
 # ----------------------------------------------------------------------------
 # Shared multi-bundle driver: build all → cheap-first sign-off → publish → Act once.
 # ----------------------------------------------------------------------------
@@ -448,6 +491,12 @@ def _drive_and_act(
     # Reject an unschedulable declared-ordering graph (cycle / unresolved dep) before any
     # build touches a bundle (issue #36). No `Depends on` fields ⇒ no-op.
     _check_dep_graph(cfg, bundles)
+    # Stack prerequisites (#123): bundles some other brief `Stacks on`. Each must publish
+    # its branch DURING the loop — not the deferred end-publish — so a dependent's next pass
+    # can base its worktree + PR on it. `published` dedups against the end-publish below.
+    stack_prereqs = {cfg.bundle(dep).name
+                     for b in bundles for dep in brief.stacks_on(b / "brief.md")}
+    published: set[str] = set()
     for _ in range(max_passes):
         # Build-all (unattended): advance each bundle to AWAITING_SIGNOFF / COMPLETE.
         # Each bundle is isolated — one that raises (a leaf left it half-written) is
@@ -483,6 +532,15 @@ def _drive_and_act(
             for d in chunk:
                 _isolate(d, "sign-off", lambda d=d: _apply_decision(
                     cfg, d, by=by, today=today, apply_now=False))
+        # Publish a stack prerequisite the moment it's COMPLETE (#123) so its branch exists
+        # for a dependent's next-pass build/publish; `published` keeps the end-loop from
+        # re-publishing it. Independents / leaf dependents publish in the end-loop as before.
+        if do_publish:
+            for d in bundles:
+                if (d.name in stack_prereqs and d.name not in published
+                        and state.state(d) == state.COMPLETE):
+                    _publish_bundle(cfg, d, by=by, today=today)
+                    published.add(d.name)
         if all(state.state(d) == state.COMPLETE for d in bundles):
             break
 
@@ -491,17 +549,9 @@ def _drive_and_act(
         # Isolated like the other per-bundle loops — one bundle whose publish raises
         # must not abort the batch return / Act for the rest (testbed issue #3).
         for d in bundles:
-            if state.state(d) == state.COMPLETE:
-                rc = _isolate(d, "publish", lambda d=d: publish.publish(
-                    cfg, d.name.removeprefix("issue_"),
-                    dry_run=cfg.publisher.mode == "stub", by=by, today=today,
-                    skip_if_no_target=True))
-                # rc != 0 (and not None — None means _isolate already logged an exception):
-                # a publish that returned failure must not pass silently (#97).
-                if rc not in (0, None):
-                    print(f"flow: {d.name} is COMPLETE but publish did not complete "
-                          f"(rc {rc}) — NOT published; run `pdca publish "
-                          f"{d.name.removeprefix('issue_')}`.", file=sys.stderr)
+            if state.state(d) == state.COMPLETE and d.name not in published:
+                _publish_bundle(cfg, d, by=by, today=today)
+                published.add(d.name)
     if do_act:
         _maybe_run_act(cfg, today,
                        any_complete=any(s == state.COMPLETE for s in results.values()))