From ddf72ea75dc75422e58d3791130d8420f542c89a Mon Sep 17 00:00:00 2001 From: mohamed zwanski Date: Thu, 22 Oct 2026 04:32:30 +0100 Subject: [PATCH] Explain why no features can start instead of waiting silently When no feature is ready and no agent is running, the orchestrator used to print a vague 'may be blocked by dependencies' line every 10s forever. It now diagnoses the root cause (circular dependencies, dependencies on nonexistent features, features that exhausted their retries, features waiting for human input) and prints an actionable explanation once, repeating it only when it changes. Also fixes _detect_cycles reporting features that merely depend on a cycle as part of it, and logs resume failures instead of discarding them. Closes #27 Co-Authored-By: Claude Opus 5.5 --- api/dependency_resolver.py | 70 ++++++++++++++++++++++++++++++++++++ parallel_orchestrator.py | 72 ++++++++++++++++++++++++++++++++++--- test_dependency_resolver.py | 50 ++++++++++++++++++++++++++ 3 files changed, 188 insertions(+), 4 deletions(-) diff --git a/api/dependency_resolver.py b/api/dependency_resolver.py index 9cc8082d3..680a6aa8a 100644 --- a/api/dependency_resolver.py +++ b/api/dependency_resolver.py @@ -24,6 +24,16 @@ class DependencyResult(TypedDict): missing_dependencies: dict[int, list[int]] # feature_id -> [missing_ids] +class StallDiagnosis(TypedDict): + """Why the remaining (non-passing) features cannot be started.""" + + needs_human_input: list[int] # features waiting for a human response + failed: list[int] # features that exhausted their retries + missing_dependencies: dict[int, list[int]] # feature_id -> [nonexistent dep ids] + circular_dependencies: list[list[int]] # cycles among non-passing features + waiting_on: dict[int, list[int]] # feature_id -> [unmet dep ids] + + def resolve_dependencies(features: list[dict]) -> DependencyResult: """Topological sort using Kahn's algorithm with priority-aware ordering. @@ -142,6 +152,62 @@ def get_blocking_dependencies( return [dep_id for dep_id in deps if dep_id not in passing_ids] +def diagnose_stalled_features( + features: list[dict], + failed_ids: set[int] | None = None, +) -> StallDiagnosis: + """Explain why non-passing features cannot make progress. + + Used when no feature is ready to start and nothing is running, so the + user can be told the actual root cause instead of waiting indefinitely. + + Args: + features: List of all feature dicts + failed_ids: IDs of features that exhausted their retries + + Returns: + StallDiagnosis grouping the blocked features by root cause + """ + failed_ids = failed_ids or set() + feature_map = {f["id"]: f for f in features} + pending = [f for f in features if not f.get("passes")] + pending_ids = {f["id"] for f in pending} + + needs_human_input: list[int] = [] + failed: list[int] = [] + missing: dict[int, list[int]] = {} + waiting_on: dict[int, list[int]] = {} + + for f in pending: + fid = f["id"] + deps = f.get("dependencies") or [] + if f.get("needs_human_input"): + needs_human_input.append(fid) + if fid in failed_ids: + failed.append(fid) + missing_deps = [d for d in deps if d not in feature_map] + if missing_deps: + missing[fid] = missing_deps + unmet = [d for d in deps if d in pending_ids] + if unmet: + waiting_on[fid] = unmet + + # Cycles only matter among features that have not passed yet + subgraph = [ + {**f, "dependencies": [d for d in (f.get("dependencies") or []) if d in pending_ids]} + for f in pending + ] + cycles = resolve_dependencies(subgraph)["circular_dependencies"] if subgraph else [] + + return { + "needs_human_input": sorted(needs_human_input), + "failed": sorted(failed), + "missing_dependencies": missing, + "circular_dependencies": cycles, + "waiting_on": waiting_on, + } + + def would_create_circular_dependency( features: list[dict], source_id: int, target_id: int ) -> bool: @@ -266,6 +332,10 @@ def dfs(fid: int) -> bool: for f in features: if f["id"] not in visited: + # dfs() returns early once it finds a cycle, so reset the stack state + # to avoid reporting features that merely depend on a cycle as cyclic + rec_stack.clear() + path.clear() dfs(f["id"]) return cycles diff --git a/parallel_orchestrator.py b/parallel_orchestrator.py index e39ef7995..9c1a089ee 100644 --- a/parallel_orchestrator.py +++ b/parallel_orchestrator.py @@ -34,7 +34,12 @@ from sqlalchemy import text from api.database import Feature, create_database -from api.dependency_resolver import are_dependencies_satisfied, compute_scheduling_scores +from api.dependency_resolver import ( + StallDiagnosis, + are_dependencies_satisfied, + compute_scheduling_scores, + diagnose_stalled_features, +) from progress import has_features from server.utils.process_utils import kill_process_tree @@ -199,6 +204,8 @@ def __init__( # Track feature failures to prevent infinite retry loops self._failure_counts: dict[int, int] = {} + # Last stall explanation printed, so it is only repeated when it changes + self._last_stall_message: str | None = None # Track recently tested feature IDs to avoid redundant re-testing. # Cleared when all passing features have been covered at least once. @@ -1575,7 +1582,9 @@ async def run_loop(self): slots = self.max_concurrency - current for feature in resumable[:slots]: print(f"Resuming feature #{feature['id']}: {feature['name']}", flush=True) - self.start_feature(feature["id"], resume=True) + success, msg = self.start_feature(feature["id"], resume=True) + if not success: + print(f"Could not resume feature #{feature['id']}: {msg}", flush=True) await asyncio.sleep(0.5) # Brief delay for subprocess to claim feature before re-querying continue @@ -1602,11 +1611,13 @@ async def run_loop(self): print("\nAll features complete!", flush=True) break - # Still have pending features but all are blocked by dependencies - print("No ready features available. All remaining features may be blocked by dependencies.", flush=True) + # Still have pending features but none can start: explain why + self._report_stall(fresh_dicts) await self._wait_for_agent_completion(timeout=POLL_INTERVAL * 2) continue + self._last_stall_message = None + # Build dependency-aware batches from ready features slots = self.max_concurrency - current batches = self.build_feature_batches(ready, feature_dicts, scheduling_scores) @@ -1658,6 +1669,18 @@ async def run_loop(self): print("Orchestrator finished.", flush=True) + def _report_stall(self, feature_dicts: list[dict]) -> None: + """Print why no feature can start, only when the explanation changes.""" + failed_ids = { + fid for fid, count in self._failure_counts.items() if count >= MAX_FEATURE_RETRIES + } + diagnosis = diagnose_stalled_features(feature_dicts, failed_ids) + message = format_stall_diagnosis(diagnosis) + if message != self._last_stall_message: + debug_log.log("STALL", "No ready features and nothing running", **diagnosis) + print(message, flush=True) + self._last_stall_message = message + def get_status(self) -> dict: """Get current orchestrator status.""" with self._lock: @@ -1713,6 +1736,47 @@ def cleanup(self) -> None: debug_log.log("CLEANUP", f"Engine dispose failed: {e}") +def _format_ids(ids: list[int], limit: int = 10) -> str: + shown = ", ".join(f"#{i}" for i in ids[:limit]) + return shown + (f" (+{len(ids) - limit} more)" if len(ids) > limit else "") + + +def format_stall_diagnosis(diagnosis: StallDiagnosis) -> str: + """Build a human-readable explanation of why no feature can start.""" + lines = ["No features can be started right now. Reason(s):"] + if diagnosis["circular_dependencies"]: + cycles = "; ".join(" -> ".join(f"#{i}" for i in [*c, c[0]]) for c in diagnosis["circular_dependencies"]) + lines.append(f" - Circular dependencies: {cycles}. Remove one dependency in each cycle.") + if diagnosis["missing_dependencies"]: + missing = "; ".join( + f"#{fid} needs {_format_ids(deps)}" for fid, deps in diagnosis["missing_dependencies"].items() + ) + lines.append(f" - Dependencies on features that do not exist: {missing}. Remove these dependencies.") + if diagnosis["failed"]: + lines.append( + f" - Features that failed {MAX_FEATURE_RETRIES} times and will not be retried this run: " + f"{_format_ids(diagnosis['failed'])}. Restart the agent to retry them." + ) + if diagnosis["needs_human_input"]: + lines.append( + f" - Features waiting for your input: {_format_ids(diagnosis['needs_human_input'])}. " + "Answer them in the UI to continue." + ) + root_causes = ( + set(diagnosis["failed"]) + | set(diagnosis["needs_human_input"]) + | set(diagnosis["missing_dependencies"]) + | {fid for cycle in diagnosis["circular_dependencies"] for fid in cycle} + ) + waiting = sorted(fid for fid in diagnosis["waiting_on"] if fid not in root_causes) + if waiting: + lines.append(f" - {len(waiting)} other feature(s) are waiting on the features above: {_format_ids(waiting)}.") + if len(lines) == 1: + lines.append(" - Unknown (see orchestrator_debug.log for the full feature state).") + lines.append("The agent will keep waiting and start work as soon as a feature becomes ready.") + return "\n".join(lines) + + async def run_parallel_orchestrator( project_dir: Path, max_concurrency: int = DEFAULT_CONCURRENCY, diff --git a/test_dependency_resolver.py b/test_dependency_resolver.py index 21550232e..d3591fe9b 100644 --- a/test_dependency_resolver.py +++ b/test_dependency_resolver.py @@ -15,6 +15,7 @@ from api.dependency_resolver import ( are_dependencies_satisfied, compute_scheduling_scores, + diagnose_stalled_features, get_blocked_features, get_blocking_dependencies, get_ready_features, @@ -380,6 +381,53 @@ def test_get_blocked_features(): return False +def test_diagnose_stalled_features(): + """Test that stalled features are grouped by their root cause.""" + print("\nTesting diagnose_stalled_features:") + + features = [ + {"id": 1, "priority": 1, "dependencies": [], "passes": True}, + {"id": 2, "priority": 2, "dependencies": [3], "passes": False}, + {"id": 3, "priority": 3, "dependencies": [2], "passes": False}, + {"id": 4, "priority": 4, "dependencies": [99], "passes": False}, + {"id": 5, "priority": 5, "dependencies": [1], "passes": False}, + {"id": 6, "priority": 6, "dependencies": [5], "passes": False}, + {"id": 7, "priority": 7, "dependencies": [], "passes": False, "needs_human_input": True}, + {"id": 8, "priority": 8, "dependencies": [2, 7], "passes": False}, + ] + + diagnosis = diagnose_stalled_features(features, failed_ids={5}) + + assert diagnosis["failed"] == [5] + assert diagnosis["needs_human_input"] == [7] + assert diagnosis["missing_dependencies"] == {4: [99]} + assert len(diagnosis["circular_dependencies"]) == 1 + assert set(diagnosis["circular_dependencies"][0]) == {2, 3} + # Passing dependency (1) is not reported as unmet + assert 5 not in diagnosis["waiting_on"] + assert diagnosis["waiting_on"][6] == [5] + assert diagnosis["waiting_on"][8] == [2, 7] + print(" PASS: Root causes identified") + return True + + +def test_diagnose_stalled_features_ignores_passing_cycles(): + """A cycle involving passing features does not block anything.""" + print("\nTesting diagnose_stalled_features ignores cycles through passing features:") + + features = [ + {"id": 1, "priority": 1, "dependencies": [2], "passes": True}, + {"id": 2, "priority": 2, "dependencies": [1], "passes": False}, + ] + + diagnosis = diagnose_stalled_features(features) + + assert diagnosis["circular_dependencies"] == [] + assert diagnosis["waiting_on"] == {} + print(" PASS: No cycle reported") + return True + + def run_all_tests(): """Run all tests and report results.""" print("=" * 60) @@ -399,6 +447,8 @@ def run_all_tests(): test_get_blocking_dependencies, test_get_ready_features, test_get_blocked_features, + test_diagnose_stalled_features, + test_diagnose_stalled_features_ignores_passing_cycles, ] passed = 0