Merge pull request #38 from eduralph/feat/dag-gated-dispatch

feat(flow): DAG-gated dispatch from declared depends_on / conflicts_with (#36)

Author: eduralph

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

@@ -52,6 +52,15 @@ The first defense against integration tangling is to not create it. Assign work
 
 Partitioning by issue id alone is not enough — it isolates the *runs* but not the *changes*. The information needed is already produced at Plan: root-cause analysis names the files / area a fix will touch. Lane assignment is therefore a **Plan-beat judgment** — the same place the human decides scope and which issues to brief ([03 - Cycle Automation](03-cycle-automation.md)) — not a mechanical sharding step. When the touched areas genuinely cannot be predicted, prefer fewer, broader lanes and lean on the integration check below.
 
+### Declared ordering — `Depends on:` / `Conflicts with:` [built]
+
+Manual wave-splitting (run a prerequisite batch to COMPLETE, *then* the next) enforces ordering by hand; it does not scale to a batch with a real dependency graph, which is exactly when the lane pool is most useful. A brief may instead **declare** its ordering constraints and let the scheduler enforce them:
+
+- **`- **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.
+- **`- **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 batch` / `flow` abort 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.
+
 ## Integration validation — at the merge boundary
 
 Whatever planning misses, correctness-*under-combination* is established where the patches actually meet: the **merge boundary**, not the lane. The harness already has the primitive — the gates are **single-sourced** ([04 - Validation Tooling](04-validation-tooling.md) §Single-sourcing): the same `pdca gates` runs over a bundle (per-fix, in a lane) **and** over the working tree (repo-scoped — `gates.run_working_tree`, "the CI merge re-gate"). Run the repo-scoped re-gate over the **merged** tree and it sees the combination the per-lane gates could not.

template/src/pdca_harness/cli.py

@@ -13,7 +13,7 @@
 import sys
 from pathlib import Path
 
-from . import act, driver, flow, gates, publish, queue, signoff, state
+from . import act, brief, driver, flow, gates, publish, queue, signoff, state
 from .config import Config
 
 # Ordering for the cheap-first sign-off queue (docs 03 §sign-off queue).
@@ -205,10 +205,22 @@ def _status(cfg: Config, issue_id: str | None) -> int:
         if s == state.AWAITING_SIGNOFF:
             n = len(signoff.open_needs_human(d / "SUMMARY.md"))
             flag = "  [cheap: confirm]" if n == 0 else f"  [{n} NEEDS-HUMAN]"
+        blocked = _blocked_by(cfg, d) if s != state.COMPLETE else []
+        if blocked:
+            flag += f"  [blocked-by: {', '.join(blocked)}]"
         print(f"{s:18}{d.name}{flag}")
     return 0
 
 
+def _blocked_by(cfg: Config, d: Path) -> list[str]:
+    """Declared `Depends on` ids of bundle ``d`` that are not yet COMPLETE (issue #36)."""
+    bp = d / "brief.md"
+    if not bp.exists():
+        return []
+    return [dep for dep in brief.depends_on(bp)
+            if state.state(cfg.bundle(dep)) != state.COMPLETE]
+
+
 def _batch(cfg: Config, args: argparse.Namespace) -> int:
     """Drive specific already-briefed issues through the FULL cycle, ending at Act.
 

template/src/pdca_harness/flow.py

@@ -22,10 +22,9 @@
 import datetime
 import sys
 import threading
-from collections import deque
 from pathlib import Path
 
-from . import driver, lane, leaves, publish, queue, signoff, state
+from . import brief, driver, lane, leaves, publish, queue, signoff, state
 from .config import Config
 
 
@@ -168,6 +167,88 @@ def flow(
     return final
 
 
+# ----------------------------------------------------------------------------
+# Declared inter-bundle ordering (docs 09, issue #36). Bundles may declare
+# `Depends on:` / `Conflicts with:` in their brief; the scheduler gates dispatch on
+# them. With NO fields declared every bundle is always eligible, so dispatch is
+# byte-for-byte today's sort-by-name pool.
+# ----------------------------------------------------------------------------
+def _deps_met(cfg: Config, d: Path) -> bool:
+    """True iff every bundle ``d`` declares ``Depends on`` is COMPLETE.
+
+    An unplanned/reopened bundle (no brief yet) declares nothing, so it is eligible;
+    its deps, if any, are honoured once it is re-planned on a later pass.
+    """
+    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))
+
+
+def _conflict_map(cfg: Config, bundles: list[Path]) -> dict[str, set[str]]:
+    """Symmetric bundle-name → conflicting-bundle-names map, restricted to this wave.
+
+    A declared conflict naming a bundle outside the wave is moot (it cannot be
+    co-scheduled with something that is not running) and is dropped.
+    """
+    names = {b.name for b in bundles}
+    conflicts: dict[str, set[str]] = {b.name: set() for b in bundles}
+    for b in bundles:
+        bp = b / "brief.md"
+        if not bp.exists():
+            continue
+        for cid in brief.conflicts_with(bp):
+            other = cfg.bundle(cid).name
+            if other in names and other != b.name:
+                conflicts[b.name].add(other)
+                conflicts[other].add(b.name)
+    return conflicts
+
+
+def _check_dep_graph(cfg: Config, bundles: list[Path]) -> None:
+    """Validate the declared `Depends on` DAG before any build (issue #36).
+
+    A dependency that is neither in this wave nor an already-COMPLETE bundle on
+    disk is a misconfigured brief; a cycle is unschedulable. Both raise ``ValueError``
+    so the run aborts before touching any bundle. No deps declared ⇒ no-op.
+    """
+    names = {b.name for b in bundles}
+    graph: dict[str, list[str]] = {}
+    for b in bundles:
+        bp = b / "brief.md"
+        edges: list[str] = []
+        for dep in (brief.depends_on(bp) if bp.exists() else []):
+            dn = cfg.bundle(dep).name
+            if dn in names:
+                edges.append(dn)
+            elif state.state(cfg.bundle(dep)) != state.COMPLETE:
+                raise ValueError(
+                    f"{b.name}: declared dependency '{dep}' is neither in this batch "
+                    f"nor an existing COMPLETE bundle")
+        graph[b.name] = edges
+
+    WHITE, GRAY, BLACK = 0, 1, 2
+    color = dict.fromkeys(graph, WHITE)
+    path: list[str] = []
+
+    def visit(n: str) -> None:
+        color[n] = GRAY
+        path.append(n)
+        for m in graph[n]:
+            if color[m] == GRAY:
+                cyc = path[path.index(m):] + [m]
+                raise ValueError("dependency cycle: " + " → ".join(cyc))
+            if color[m] == WHITE:
+                visit(m)
+        path.pop()
+        color[n] = BLACK
+
+    for n in graph:
+        if color[n] == WHITE:
+            visit(n)
+
+
 # ----------------------------------------------------------------------------
 # The unattended band: advance every bundle through Do + Check (docs 09). Serial by
 # default; a worker pool of cfg.lanes lanes when configured (PDCA_LANES / [driver].lanes).
@@ -185,27 +266,57 @@ def _build_all(cfg: Config, bundles: list[Path]) -> None:
     """
     if cfg.lanes <= 1 or len(bundles) <= 1:
         for d in bundles:
+            if not _deps_met(cfg, d):
+                continue  # a declared prereq isn't COMPLETE yet — a later pass picks it up
             def _build(d=d):
                 _plan_if_unplanned(cfg, d, None)  # iterate-plan may have re-opened it
                 driver.run_issue(d, cfg)
             _isolate(d, "build/check", _build)
         return
 
-    # Serial Plan pre-pass — the interactive Plan beat stays out of the pool.
+    # Serial Plan pre-pass — the interactive Plan beat stays out of the pool. After it
+    # every bundle has a brief, so the declared-conflict map is complete.
     for d in bundles:
         _isolate(d, "plan", lambda d=d: _plan_if_unplanned(cfg, d, None))
+    conflicts = _conflict_map(cfg, bundles)
     # Pooled drive — fixed lane slot per worker; gates read it via lane.current().
-    work = deque(bundles)
-    lock = threading.Lock()
+    # A worker claims the first queued bundle whose declared deps are COMPLETE and which
+    # conflicts with nothing currently in flight; with no fields declared the first
+    # queued bundle is always eligible, so this is the same FIFO pool as before.
+    remaining = list(bundles)  # preserves the caller's sort-by-name order
+    inflight: set[str] = set()
+    cond = threading.Condition()
+
+    def _next_eligible() -> Path | None:
+        # caller holds `cond`. Pop+return the first eligible bundle, else None.
+        for i, d in enumerate(remaining):
+            if _deps_met(cfg, d) and conflicts[d.name].isdisjoint(inflight):
+                inflight.add(d.name)
+                return remaining.pop(i)
+        return None
 
     def worker(slot: int) -> None:
         lane.set_current(slot)
         while True:
-            with lock:
-                if not work:
-                    return
-                d = work.popleft()
+            with cond:
+                while True:
+                    if not remaining:
+                        return
+                    d = _next_eligible()
+                    if d is not None:
+                        break
+                    # Nothing eligible right now. If nothing is in flight to unblock the
+                    # rest, they are dep-blocked on prereqs that only go COMPLETE after a
+                    # later sign-off pass — leave them and exit. Otherwise wait for an
+                    # in-flight bundle to finish and re-check.
+                    if not inflight:
+                        cond.notify_all()
+                        return
+                    cond.wait()
             _isolate(d, "build/check", lambda d=d: driver.run_issue(d, cfg))
+            with cond:
+                inflight.discard(d.name)
+                cond.notify_all()
 
     threads = [threading.Thread(target=worker, args=(k,), name=f"pdca-lane{k}")
                for k in range(min(cfg.lanes, len(bundles)))]
@@ -238,6 +349,9 @@ def _drive_and_act(
     batch — the endpoint is Act, like any single cycle, just fanned over several bundles.
     """
     names = {b.name for b in bundles}
+    # 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)
     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

template/tests/test_flow_slice.py

@@ -427,5 +427,141 @@ def test_serial_path_sets_no_pdca_lane(self) -> None:
         self.assertEqual(val, "none")
 
 
+class DeclaredOrdering(unittest.TestCase):
+    """Declared inter-bundle ordering (docs 09 / issue #36): a brief may declare
+    `Depends on:` (topological gate — a dependent isn't driven until its prereq is
+    COMPLETE) and `Conflicts with:` (never co-scheduled in one concurrent wave). With
+    no fields declared, dispatch is exactly today's sort-by-name pool."""
+
+    def setUp(self) -> None:
+        self.tmp = Path(tempfile.mkdtemp())
+        self.cfg = _stub_config(self.tmp)
+
+    def tearDown(self) -> None:
+        shutil.rmtree(self.tmp, ignore_errors=True)
+
+    def _brief(self, iid: str, *, depends_on: str = "", conflicts_with: str = "") -> Path:
+        d = self.cfg.bundle(iid)
+        d.mkdir(parents=True)
+        body = _TOY_BRIEF.format(slug=iid.lower())
+        if depends_on:
+            body += f"- **Depends on:** {depends_on}\n"
+        if conflicts_with:
+            body += f"- **Conflicts with:** {conflicts_with}\n"
+        (d / "brief.md").write_text(body, encoding="utf-8")
+        return d
+
+    def test_dependent_not_driven_until_prereq_complete(self) -> None:
+        # AA depends on ZZ. Sort-by-name would build AA first; the gate must hold AA
+        # until ZZ is COMPLETE (a later pass), proving ordering is by deps, not name.
+        self._brief("AA", depends_on="ZZ")
+        self._brief("ZZ")
+        seen = {}
+        real = driver.run_issue
+
+        def spy(d: Path, cfg: Config):
+            if d.name == "issue_AA" and "zz_state" not in seen:
+                seen["zz_state"] = state.state(cfg.bundle("ZZ"))
+            return real(d, cfg)
+
+        driver.run_issue = spy
+        try:
+            results = flow.flow_ids(self.cfg, ["AA", "ZZ"], do_publish=False,
+                                    do_act=False, today="2026-06-04")
+        finally:
+            driver.run_issue = real
+        self.assertEqual(seen.get("zz_state"), state.COMPLETE)  # ZZ done before AA built
+        self.assertTrue(all(s == state.COMPLETE for s in results.values()))
+
+    def test_no_deps_keeps_sort_by_name_dispatch(self) -> None:
+        # No Depends-on fields → the serial build order is exactly sort-by-name, byte
+        # for byte today's behaviour.
+        ids = ["N3", "N1", "N2"]
+        for iid in ids:
+            self._brief(iid)
+        order: list[str] = []
+        real = driver.run_issue
+
+        def spy(d: Path, cfg: Config):
+            if d.name not in order:
+                order.append(d.name)
+            return real(d, cfg)
+
+        driver.run_issue = spy
+        try:
+            flow.flow_ids(self.cfg, ids, do_publish=False, do_act=False,
+                          today="2026-06-04")
+        finally:
+            driver.run_issue = real
+        self.assertEqual(order, ["issue_N1", "issue_N2", "issue_N3"])
+
+    def test_conflict_pair_never_co_scheduled(self) -> None:
+        # C conflicts with D; E/F are free. Under a 2-lane pool, C and D must never be
+        # in flight together, while the free bundles still prove the pool parallelises.
+        import threading
+        import time
+
+        self._brief("C", conflicts_with="D")
+        self._brief("D")
+        self._brief("E")
+        self._brief("F")
+        self.cfg.lanes = 2
+
+        active: set[str] = set()
+        together: set[tuple[str, str]] = set()
+        max_conc = [0]
+        lk = threading.Lock()
+        real = driver.run_issue
+
+        def spy(d: Path, cfg: Config):
+            with lk:
+                active.add(d.name)
+                max_conc[0] = max(max_conc[0], len(active))
+                for a in active:
+                    for b in active:
+                        if a < b:
+                            together.add((a, b))
+            time.sleep(0.05)
+            try:
+                return real(d, cfg)
+            finally:
+                with lk:
+                    active.discard(d.name)
+
+        driver.run_issue = spy
+        try:
+            results = flow.flow_ids(self.cfg, ["C", "D", "E", "F"], do_publish=False,
+                                    do_act=False, today="2026-06-04")
+        finally:
+            driver.run_issue = real
+        self.assertNotIn(("issue_C", "issue_D"), together)  # conflict respected
+        self.assertEqual(max_conc[0], 2)                     # pool genuinely concurrent
+        self.assertTrue(all(s == state.COMPLETE for s in results.values()))
+
+    def test_dependency_cycle_is_rejected_before_build(self) -> None:
+        # A↔B mutual dependency is unschedulable: reject up front, before any build.
+        self._brief("CYA", depends_on="CYB")
+        self._brief("CYB", depends_on="CYA")
+        real = driver.run_issue
+        built = {"n": 0}
+        driver.run_issue = lambda d, cfg: (built.__setitem__("n", built["n"] + 1)
+                                           or real(d, cfg))
+        try:
+            with self.assertRaises(ValueError):
+                flow.flow_ids(self.cfg, ["CYA", "CYB"], do_publish=False,
+                              do_act=False, today="2026-06-04")
+        finally:
+            driver.run_issue = real
+        self.assertEqual(built["n"], 0)  # rejected before touching any bundle
+
+    def test_unresolved_dependency_is_rejected(self) -> None:
+        # A dep that is neither in the wave nor an existing COMPLETE bundle is a
+        # misconfigured brief — a hard error.
+        self._brief("DEP1", depends_on="GHOST")
+        with self.assertRaises(ValueError):
+            flow.flow_ids(self.cfg, ["DEP1"], do_publish=False, do_act=False,
+                          today="2026-06-04")
+
+
 if __name__ == "__main__":
     unittest.main()