From f091e78efe3696b1b8ec125260a3d510d078c197 Mon Sep 17 00:00:00 2001 From: pbean Date: Thu, 9 Jul 2026 19:32:16 -0700 Subject: [PATCH] fix(sweep): re-drive in-flight bundles by identity on resume MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit SweepEngine._loop never had the base engine's _finish_inflight pass. Its only recovery arm lived inside _run_bundle, reachable only once a cycle re-derived the bundle key from the *current* triage plan — so a bundle re-armed by `bmad-loop resolve` survived a resume solely because the cached triage.json reloaded and re-emitted the same bundle name. Lose that cache and a fresh triage partitions the ids under new names, silently orphaning the resolution. _finish_inflight_bundles now runs at the top of the loop, before the ledger is read, and re-drives every non-terminal dw* task under its own persisted story_key. Because its ids leave the open set first, validate_triage rejects any fresh plan that references them — a double-drive is unrepresentable. A still-escalated bundle is terminal and stays untouched. The corrupt-triage.json leg of the issue was unreachable: _read_json is a bare json.loads, so a truncated cache crashed the whole run out of _ensure_triage. It now degrades to a fresh triage, as does a cache holding a non-object. Also: _run_bundle's recovery arm is extracted verbatim to _recover_inflight_bundle and kept as its fallback (the plan-matched path is bit-for-bit unchanged); a missing bundle intent file is regenerated from the task; and a bundle surviving to a cycle is journaled + notified, never dropped. Closes #94 --- CHANGELOG.md | 14 +++ src/bmad_loop/sweep.py | 205 +++++++++++++++++++++++++++++------- tests/test_sweep.py | 232 ++++++++++++++++++++++++++++++++++++++++- 3 files changed, 410 insertions(+), 41 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 8e53e836..92bef222 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -46,6 +46,20 @@ breaking changes may land in a minor release. ### Fixed +- **A resumed sweep re-drives its in-flight bundles by identity, not by bundle name.** `SweepEngine` + recovered a bundle only from inside `_run_bundle`, which a cycle reaches after re-deriving the key + from the _current_ triage plan — so a bundle re-armed by `bmad-loop resolve` survived a resume only + because the cached `triage.json` reloaded and re-emitted the same name. Lose that cache and a fresh + triage partitioned the ids under new names, silently orphaning the human's resolution. The sweep + loop now opens with `_finish_inflight_bundles`, mirroring the base engine: every non-terminal `dw*` + task is re-driven under its own persisted `story_key`, before the ledger is read, so its ids leave + the open set and no fresh plan can re-bundle them. A still-escalated bundle stays terminal and + untouched. A missing bundle intent file is regenerated from the task (the verbatim ledger entries + become the contract; the triage prose is the only unrecoverable piece), and a bundle that survives + to a cycle anyway is journaled + notified rather than dropped. Relatedly, a truncated or + wrong-shaped `triage.json` now degrades to a fresh triage instead of crashing the whole run — the + corrupt leg of this bug was previously unreachable. New journal events: `sweep-inflight-redrive` / + `-stranded`, `sweep-intent-regenerated`. (#94) - **Run ids are validated, so a run ref can no longer escape the runs directory.** A positional ref (`delete`, `stop`, `archive`, `resume`, `status`) was recomposed into a path raw, so `bmad-loop delete ../../x` deleted any outside directory holding a `state.json`; the hidden diff --git a/src/bmad_loop/sweep.py b/src/bmad_loop/sweep.py index 7b69597f..02b74c32 100644 --- a/src/bmad_loop/sweep.py +++ b/src/bmad_loop/sweep.py @@ -36,6 +36,10 @@ def _read_json(path: Path) -> Any: MIGRATE_KEY = "sweep-migrate" MIGRATE_WORKFLOW = "deferred-sweep-migrate" BUNDLE_NAME_RE = re.compile(r"^[a-z0-9][a-z0-9-]{1,39}$") +# the inverse of SweepEngine._bundle_key: "dw-" (cycle 1) / "dw-". +# A cycle-1 key always has "-" straight after "dw", so the cycle group matches +# empty and the split stays unambiguous even for a bundle named "2fix". +BUNDLE_KEY_RE = re.compile(r"^dw(\d*)-(.+)$") DECISION_EFFECTS = ("build", "close", "keep-open") @@ -408,6 +412,11 @@ def _today(self) -> str: def _loop(self) -> None: ledger = self.workspace.paths.deferred_work cycle = max(1, self.state.sweep_cycle) + if self._finish_inflight_bundles(): + # a recovered bundle's ledger restore can leave the tree dirty, and + # triage plus the first bundle baseline need a clean one. Guarded on + # a non-empty pass so a fresh sweep never commits the user's dirt. + self._commit_ledger("chore(sweep): commit ledger after recovering in-flight bundles") while True: self.state.sweep_cycle = cycle self._save() @@ -451,6 +460,62 @@ def _loop(self) -> None: self._commit_ledger("chore(sweep): commit ledger before next sweep cycle") cycle += 1 + def _finish_inflight_bundles(self) -> int: + """Re-drive every bundle this run left in flight, keyed on the task's own + persisted story_key. Returns how many were recovered. + + The base Engine._loop opens with _finish_inflight for exactly this reason; + the sweep loop used to recover a bundle only from inside _run_bundle, which + a cycle reaches only after re-deriving the bundle's key from the *current* + triage plan. A re-armed bundle therefore survived a resume solely because + the cached triage.json reloaded and re-emitted the same bundle name — lose + that cache and a fresh triage partitions the ids under new names, silently + orphaning the human's resolution (#94). + + Runs before the ledger is read, so a bundle it closes leaves the open set + and no fresh triage can re-bundle those ids (validate_triage rejects a plan + whose open_ids disagree with the ledger). A recovered bundle that defers or + escalates keeps its ids open, where the existing failed_ids filter drops the + fresh plan's overlapping bundle.""" + recovered = 0 + for task in list(self.state.tasks.values()): + if task.terminal or not BUNDLE_KEY_RE.match(task.story_key): + continue + recovered += 1 + self.journal.append( + "sweep-inflight-redrive", + story_key=task.story_key, + phase=str(task.phase), + rearmed=task.rearmed, # read before the recovery clears the latch + ) + if self._recover_inflight_bundle(task): + continue + self._ensure_bundle_intent(task) + self._save() + self._emit("pre_bundle", task) + self._run_story(task) + self._emit("post_bundle", task) + return recovered + + def _warn_stranded_bundles(self) -> None: + """Invariant: _finish_inflight_bundles has driven every persisted bundle to + a terminal phase before a cycle picks new work. A survivor means a bundle + would be silently dropped — say so loudly rather than sweep past it.""" + stranded = [ + t.story_key + for t in self.state.tasks.values() + if BUNDLE_KEY_RE.match(t.story_key) and not t.terminal + ] + if not stranded: + return + self.journal.append("sweep-inflight-stranded", story_keys=stranded) + gates.notify( + self.policy, + self.run_dir, + f"{len(stranded)} sweep bundle(s) left in flight", + "not re-driven by this cycle: " + ", ".join(stranded), + ) + def _cycle(self, cycle: int, open_now: set[str]) -> bool: """One triage -> close -> decide -> bundle pass. Returns whether the cycle completed any addressable work — the repeat loop's progress @@ -458,6 +523,7 @@ def _cycle(self, cycle: int, open_now: set[str]) -> bool: already-resolved closes, the replayed (idempotent) closes report 0 and the run stops with no-progress; errs toward stopping, never loops.""" self._emit("pre_sweep_cycle", phase=str(cycle)) + self._warn_stranded_bundles() plan = self._ensure_triage(open_now, cycle) closed = self._close_resolved(plan) answers, decisions_closed = self._decisions_phase(plan) @@ -515,41 +581,8 @@ def _run_bundle(self, bundle: Bundle, cycle: int) -> None: task = StoryTask(story_key=key, epic=0, dw_ids=list(bundle.dw_ids)) self.state.tasks[key] = task self.journal.append("bundle-start", story_key=key, dw_ids=list(bundle.dw_ids)) - else: - # interrupted mid-bundle (or a re-armed escalation resolved via - # `bmad-loop resolve`): same recovery as Engine._finish_inflight, - # including the re-drive latch so a human-resolved escalation is - # protected through every reset (mirrors engine.py:1151-1157). - self.journal.append("resume-restart", story_key=key, phase=str(task.phase)) - isolated = self._isolated and task.worktree_path - if task.phase == Phase.DEV_VERIFY and task.spec_file: - self._save() - if isolated: - unit = self._reopen_unit(task) - prev = self.workspace - self.workspace = unit.workspace - try: - self._review_and_commit(task) - finally: - self.workspace = prev - self._integrate_unit(task, unit) - else: - self._review_and_commit(task) - return - if isolated: - # drop the half-built worktree; _run_story mounts a fresh one - discard_worktree(self.paths.repo_root, task.worktree_path, task.branch) - task.worktree_path = "" - task.branch = "" - elif task.baseline_commit: - # latch resolved_redrive so the corrected spec + restored diff stay - # protected through every reset of this re-drive, not just this - # first one; cause="resolved" keeps a human-initiated re-arm - # pause-free regardless of scm.rollback_on_failure - task.resolved_redrive = task.resolved_redrive or task.rearmed - self._rollback_or_pause(task, cause="resolved" if task.rearmed else "stopped") - task.rearmed = False # past rollback (only reached when not paused) - task.phase = Phase.PENDING # deliberate reset, not a normal transition + elif self._recover_inflight_bundle(task): + return dirname = bundle.name if cycle == 1 else f"c{cycle}-{bundle.name}" task.bundle_file = str(self._write_intent(bundle, dirname)) self._save() @@ -557,6 +590,53 @@ def _run_bundle(self, bundle: Bundle, cycle: int) -> None: self._run_story(task) self._emit("post_bundle", task) + def _recover_inflight_bundle(self, task: StoryTask) -> bool: + """Recover a bundle task interrupted mid-flight (or re-armed after a + human resolved its escalation via `bmad-loop resolve`): the same recovery + as Engine._finish_inflight, including the re-drive latch so a + human-resolved escalation is protected through every reset (mirrors + engine.py:1163-1169). + + Returns True when the dev-verified arm carried the bundle all the way + through review and commit — the caller is done. Returns False once the + task has been reset to PENDING, leaving the caller to dispatch it. + + Deliberately narrower than the base _finish_inflight: no + `_resumable_session` arm, so a bundle whose host died in the + post-session window still restarts rather than replaying its recorded + result. Lifting that is a resume-fidelity change of its own.""" + self.journal.append("resume-restart", story_key=task.story_key, phase=str(task.phase)) + isolated = self._isolated and task.worktree_path + if task.phase == Phase.DEV_VERIFY and task.spec_file: + self._save() + if isolated: + unit = self._reopen_unit(task) + prev = self.workspace + self.workspace = unit.workspace + try: + self._review_and_commit(task) + finally: + self.workspace = prev + self._integrate_unit(task, unit) + else: + self._review_and_commit(task) + return True + if isolated: + # drop the half-built worktree; _run_story mounts a fresh one + discard_worktree(self.paths.repo_root, task.worktree_path, task.branch) + task.worktree_path = "" + task.branch = "" + elif task.baseline_commit: + # latch resolved_redrive so the corrected spec + restored diff stay + # protected through every reset of this re-drive, not just this + # first one; cause="resolved" keeps a human-initiated re-arm + # pause-free regardless of scm.rollback_on_failure + task.resolved_redrive = task.resolved_redrive or task.rearmed + self._rollback_or_pause(task, cause="resolved" if task.rearmed else "stopped") + task.rearmed = False # past rollback (only reached when not paused) + task.phase = Phase.PENDING # deliberate reset, not a normal transition + return False + # ------------------------------------------------------------ migration def _ensure_migration(self, text: str) -> None: @@ -678,11 +758,21 @@ def _ensure_triage(self, open_now: set[str], cycle: int = 1) -> TriagePlan: triage_key = TRIAGE_KEY + suffix if triage_path.is_file(): # already validated this run; the ledger has moved since (closes, - # decisions), so skip the open-set equality re-check - plan, errors = validate_triage(_read_json(triage_path), None) - if plan is not None: - return plan - self.journal.append("sweep-triage-reload-failed", errors=errors) + # decisions), so skip the open-set equality re-check. A cache we + # cannot read or that is not a JSON object degrades to a fresh + # triage — a truncated file must not crash the whole run. + try: + cached = _read_json(triage_path) + except (json.JSONDecodeError, OSError, UnicodeDecodeError) as exc: + self.journal.append("sweep-triage-reload-failed", errors=[f"unreadable: {exc}"]) + else: + if isinstance(cached, dict): + plan, errors = validate_triage(cached, None) + else: + plan, errors = None, [f"not a JSON object: {type(cached).__name__}"] + if plan is not None: + return plan + self.journal.append("sweep-triage-reload-failed", errors=errors) task = self.state.tasks.get(triage_key) if task is None: @@ -979,6 +1069,41 @@ def _write_intent(self, bundle: Bundle, dirname: str) -> Path: path.write_text("\n".join(lines), encoding="utf-8") return path + def _ensure_bundle_intent(self, task: StoryTask) -> None: + """Guarantee a recovered bundle has the intent file its dev prompt points + at. The rendered intent.md persists in the run dir and the prompt consumes + nothing else from the plan, so the normal case is to reuse it untouched. + + Only when it is gone do we rebuild a degraded one from the task itself. + The triage session's authored intent prose is the single unrecoverable + piece; the verbatim ledger entries _write_intent re-attaches carry the + actual work, so say plainly that they are now the contract.""" + if task.bundle_file and Path(task.bundle_file).is_file(): + return + match = BUNDLE_KEY_RE.match(task.story_key) + if match is None: # pragma: no cover - callers filter on BUNDLE_KEY_RE + return + cycle = int(match.group(1)) if match.group(1) else 1 + name = match.group(2) + bundle = Bundle( + name=name, + dw_ids=tuple(task.dw_ids), + intent=( + "Resolve the deferred-work entries reproduced below. This bundle's " + "original triage intent did not survive the run it was written in, " + "so the verbatim ledger entries are the authoritative statement of " + "the work." + ), + ) + dirname = name if cycle == 1 else f"c{cycle}-{name}" + task.bundle_file = str(self._write_intent(bundle, dirname)) + self.journal.append( + "sweep-intent-regenerated", + story_key=task.story_key, + dw_ids=list(task.dw_ids), + path=task.bundle_file, + ) + # ------------------------------------------------------ override seams def _dev_prompt(self, task: StoryTask, feedback: Path | None) -> str: diff --git a/tests/test_sweep.py b/tests/test_sweep.py index 2d37eafd..93c62872 100644 --- a/tests/test_sweep.py +++ b/tests/test_sweep.py @@ -2,7 +2,9 @@ import json import re +from pathlib import Path +import pytest from conftest import ( bundle_dev_effect, bundle_dev_escalates, @@ -18,7 +20,7 @@ from bmad_loop import deferredwork, runs, verify from bmad_loop.adapters.base import SessionResult from bmad_loop.adapters.mock import MockAdapter -from bmad_loop.journal import Journal, load_state +from bmad_loop.journal import Journal, load_state, save_state from bmad_loop.model import Phase, RunState, StoryTask, TokenUsage from bmad_loop.policy import ( DevPolicy, @@ -1719,3 +1721,231 @@ def test_sweep_bundle_budget_followup_not_refiled_twice(project): assert open_followups == [] # no second follow-up entry created capped = [e for e in engine.journal.entries() if e["kind"] == "review-budget-committed"] assert len(capped) == 1 and capped[0]["re_review_capped"] is True + + +# ------------------------------ in-flight bundle recovery on resume (#94) + + +def _lose_triage(run_dir, corruption="missing"): + """Make the cached triage plan unusable the three ways a real run can: the + file vanished, it was truncated mid-write, or it holds something that is not + a triage result.""" + path = run_dir / "triage.json" + if corruption == "missing": + path.unlink() + elif corruption == "invalid-json": + path.write_text("{{{", encoding="utf-8") + else: + path.write_text("{}", encoding="utf-8") + + +def _run_two_bundle_dev_escalation(project): + """Drive a two-bundle sweep until the first bundle's dev session escalates. + Returns the paused engine; `dw-fix` is ESCALATED, `dw-other` never started, + both dw ids still open.""" + write_ledger(project, {"DW-1": "open", "DW-2": "open"}) + plan = triage_result( + ["DW-1", "DW-2"], + bundles=[ + {"name": "fix", "dw_ids": ["DW-1"], "intent": "resolve DW-1"}, + {"name": "other", "dw_ids": ["DW-2"], "intent": "resolve DW-2"}, + ], + ) + engine, _ = make_sweep( + project, [triage_effect(plan), bundle_dev_escalates(project, "fix", ["DW-1"])] + ) + assert engine.run().paused + assert engine.state.tasks["dw-fix"].phase == Phase.ESCALATED + assert "dw-other" not in engine.state.tasks + return engine + + +def _redrive_script(project): + return [bundle_dev_effect(project, "fix", ["DW-1"]), bundle_review_effect(project, "fix")] + + +def test_rearmed_bundle_redrives_when_triage_json_lost(project): + # The regression: a human-resolved bundle used to re-drive only because the + # cached triage plan reloaded and re-emitted its name. Recovery now keys on + # the persisted task, so losing the cache changes nothing. + engine = _run_to_dev_escalation(project) + runs.rearm_escalation(engine.run_dir, "dw-fix") + _lose_triage(engine.run_dir) + + resumed, adapter = resume_sweep(project, engine, _redrive_script(project)) + summary = resumed.run() + + assert not summary.paused + assert resumed.state.tasks["dw-fix"].phase == Phase.DONE + # no triage session: the re-drive never consulted a plan + assert [s.role for s in adapter.sessions] == ["dev", "review"] + journal = journal_text(resumed) + assert "sweep-inflight-redrive" in journal + assert "sweep-nothing-open" in journal # recovery closed the only open id + assert ledger_entries(project)["DW-1"].status.startswith("done") + + +@pytest.mark.parametrize("corruption", ["missing", "invalid-json", "wrong-shape"]) +def test_fresh_triage_different_bundle_name_no_double_drive(project, corruption): + # The fresh triage renames the surviving bundle, so a name-matched recovery + # would orphan the re-armed one. It must re-drive by identity, and its ids + # must have left the open set before the fresh triage sees them. + engine = _run_two_bundle_dev_escalation(project) + runs.rearm_escalation(engine.run_dir, "dw-fix") + _lose_triage(engine.run_dir, corruption) + + fresh = triage_result( + ["DW-2"], bundles=[{"name": "renamed-fix", "dw_ids": ["DW-2"], "intent": "resolve DW-2"}] + ) + resumed, adapter = resume_sweep( + project, + engine, + [ + *_redrive_script(project), + triage_effect(fresh), + bundle_dev_effect(project, "renamed-fix", ["DW-2"]), + bundle_review_effect(project, "renamed-fix"), + ], + ) + summary = resumed.run() + + assert not summary.paused + assert [s.role for s in adapter.sessions] == ["dev", "review", "triage", "dev", "review"] + assert resumed.state.tasks["dw-fix"].phase == Phase.DONE + assert resumed.state.tasks["dw-renamed-fix"].phase == Phase.DONE + # each id was closed exactly once, by the bundle that owned it + closed = [e for e in resumed.journal.entries() if e["kind"] == "sweep-bundle-closed"] + owners = [(i, e["story_key"]) for e in closed for i in e["dw_ids"]] + assert sorted(owners) == [("DW-1", "dw-fix"), ("DW-2", "dw-renamed-fix")] + if corruption != "missing": + # a truncated / wrong-shape cache degrades to a fresh triage, never a crash + assert "sweep-triage-reload-failed" in journal_text(resumed) + + +def test_restore_patch_latch_honored_when_triage_json_lost(project, monkeypatch): + monkeypatch.setattr(verify, "apply_patch", lambda repo, patch: None) + engine = _run_to_dev_escalation(project) + patch = project.implementation_artifacts / "attempt-dw-fix.patch" + patch.parent.mkdir(parents=True, exist_ok=True) + patch.write_text("dummy\n") + runs.rearm_escalation(engine.run_dir, "dw-fix", restore_patch=str(patch)) + _lose_triage(engine.run_dir) + + resumed, adapter = resume_sweep(project, engine, _redrive_script(project)) + summary = resumed.run() + + assert not summary.paused + task = resumed.state.tasks["dw-fix"] + assert task.phase == Phase.DONE + # the recovery pass preserved the restore semantics _run_bundle used to own + assert "Resume review of the in-review spec" in adapter.sessions[0].prompt + assert "attempt-restored" in journal_text(resumed) + assert task.restore_patch is None and task.resolved_redrive is False + assert ledger_entries(project)["DW-1"].status.startswith("done") + + +def test_escalated_unresolved_still_skipped_when_triage_json_lost(project): + # An escalation nobody resolved is terminal: recovery must not touch it, and + # the fresh triage's overlapping bundle is still dropped by the failed-ids filter. + engine = _run_two_bundle_dev_escalation(project) + _lose_triage(engine.run_dir) + + fresh = triage_result( + ["DW-1", "DW-2"], + bundles=[ + {"name": "retry-fix", "dw_ids": ["DW-1"], "intent": "a"}, + {"name": "other", "dw_ids": ["DW-2"], "intent": "b"}, + ], + ) + resumed, adapter = resume_sweep( + project, + engine, + [ + triage_effect(fresh), + bundle_dev_effect(project, "other", ["DW-2"]), + bundle_review_effect(project, "other"), + ], + ) + summary = resumed.run() + + assert not summary.paused + assert "sweep-inflight-redrive" not in journal_text(resumed) + assert [s.role for s in adapter.sessions] == ["triage", "dev", "review"] + assert resumed.state.tasks["dw-fix"].phase == Phase.ESCALATED + assert "dw-retry-fix" not in resumed.state.tasks + assert "sweep-bundle-skipped" in journal_text(resumed) + entries = ledger_entries(project) + assert entries["DW-1"].open # escalated bundle untouched + assert entries["DW-2"].status.startswith("done") + + +def test_interrupted_bundle_redrives_by_identity_after_triage_loss(project): + # Not a re-arm: the host just died mid-dev. The restart arm rolls the attempt + # back against its own baseline (cause="stopped") and re-runs the bundle. + engine = _run_to_dev_escalation(project) + state = load_state(engine.run_dir) + task = state.tasks["dw-fix"] + assert task.baseline_commit + task.phase = Phase.DEV_RUNNING + save_state(engine.run_dir, state) + _lose_triage(engine.run_dir) + + resumed, adapter = resume_sweep(project, engine, _redrive_script(project)) + summary = resumed.run() + + assert not summary.paused + assert resumed.state.tasks["dw-fix"].phase == Phase.DONE + assert [s.role for s in adapter.sessions] == ["dev", "review"] + redrive = [e for e in resumed.journal.entries() if e["kind"] == "sweep-inflight-redrive"] + assert len(redrive) == 1 + assert redrive[0]["phase"] == "dev-running" and redrive[0]["rearmed"] is False + journal = journal_text(resumed) + assert "rollback-auto" in journal # a stopped attempt, rolled back to baseline + assert "attempt-restored" not in journal # not a resolved re-drive + assert ledger_entries(project)["DW-1"].status.startswith("done") + + +def test_regenerated_intent_when_bundle_file_missing(project): + # The triage session's authored prose is the one unrecoverable piece; the + # verbatim ledger entries are re-attached and become the contract. + engine = _run_to_dev_escalation(project) + runs.rearm_escalation(engine.run_dir, "dw-fix") + _lose_triage(engine.run_dir) + intent = Path(engine.state.tasks["dw-fix"].bundle_file) + intent.unlink() + + resumed, adapter = resume_sweep(project, engine, _redrive_script(project)) + summary = resumed.run() + + assert not summary.paused + assert resumed.state.tasks["dw-fix"].phase == Phase.DONE + regen = [e for e in resumed.journal.entries() if e["kind"] == "sweep-intent-regenerated"] + assert len(regen) == 1 and regen[0]["dw_ids"] == ["DW-1"] + assert regen[0]["path"] == str(intent) + text = intent.read_text(encoding="utf-8") + assert "bundle_name: fix" in text + assert "### DW-1" in text and "reason: test entry." in text # verbatim ledger entry + assert "authoritative" in text + assert str(intent) in adapter.sessions[0].prompt # the dev session got the rebuilt file + + +def test_stranded_bundle_task_warns_loudly(project): + write_ledger(project, {"DW-1": "open"}) + engine, _ = make_sweep(project, []) + engine.state.tasks["dw-ghost"] = StoryTask( + story_key="dw-ghost", epic=0, dw_ids=["DW-1"], phase=Phase.DEV_RUNNING + ) + + engine._warn_stranded_bundles() + + stranded = [e for e in engine.journal.entries() if e["kind"] == "sweep-inflight-stranded"] + assert len(stranded) == 1 and stranded[0]["story_keys"] == ["dw-ghost"] + assert "dw-ghost" in (engine.run_dir / "ATTENTION").read_text(encoding="utf-8") + + # a terminal bundle and a non-bundle task are not stranded + engine.state.tasks["dw-ghost"].phase = Phase.DONE + engine.state.tasks["sweep-triage"] = StoryTask( + story_key="sweep-triage", epic=0, phase=Phase.TRIAGE_RUNNING + ) + engine._warn_stranded_bundles() + assert len([e for e in engine.journal.entries() if e["kind"] == "sweep-inflight-stranded"]) == 1