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