from __future__ import annotations import json from datetime import datetime, timezone from pathlib import Path from typing import Any, Dict, List, Tuple from .airxdb_runtime import ( ensure_xdb_sessions_for_result, merge_xdb_sessions_into_state, normalize_xdb_policy, ) from .contracts import ( DEFAULT_DEBUG_POLICY, DEFAULT_XDB_POLICY, ParallelReview, REQUIRED_PROJECT_ARTIFACTS, WorkerResult, now_iso, ) from .debug_runtime import ( ensure_debug_sessions_for_result, merge_debug_sessions_into_state, normalize_debug_policy, ) from .doc_sync import ( apply_document_updates, enforce_doc_sync_requirements, mark_tasks_dispatched, sync_engine_managed_docs, update_todo_after_merge, ) from .paths import airarc_root, aireng_root, todo_path as workflow_todo_path from .repair_runtime import ( ensure_repair_attempts_for_result, load_active_repair_attempt, mark_repair_attempts_active, merge_repair_attempts_into_state, normalize_repair_policy, write_repair_queue, ) from .review import build_parallel_review, render_review_markdown from .todo_parser import parse_tasks from .worker import enter_worker def _json_dump(path: Path, payload: Dict[str, object]) -> None: path.parent.mkdir(parents=True, exist_ok=True) path.write_text(json.dumps(payload, indent=2) + "\n", encoding="utf-8") def _json_load(path: Path) -> Dict[str, object]: if not path.exists(): return {} return json.loads(path.read_text(encoding="utf-8-sig")) def _default_autonomy_policy() -> Dict[str, object]: return { "mayRunUnattended": True, "mayEditFiles": True, "mayRunCommands": True, "defaultNoCodeExecution": True, "subagentFirst": True, "interventionScope": [ "dispatch-state-repair", "worker-redispatch", "repair-brief-refresh", "global-doc-convergence", "glue-layer-unblock", ], } def _default_monitoring_policy() -> Dict[str, object]: return { "checkIntervalSeconds": 300, "stallAfterSeconds": 1800, "maxInterventionAttemptsPerTask": 2, } def _default_active_worker() -> Dict[str, object]: return { "taskId": "", "dispatchGroup": "", "waveId": "", "briefPath": "", "handoffPath": "", "workerStatePath": "", "resultPath": "", "spawnedAt": "", "lastObservedAt": "", "lastHeartbeatAt": "", "status": "queued", "lastResultPath": "", "stallCount": 0, "interventionCount": 0, "repairDispatchPath": "", "notes": [], } def _normalize_active_worker(payload: Dict[str, object]) -> Dict[str, object]: worker = _default_active_worker() worker.update(payload) worker["notes"] = [str(item) for item in list(worker.get("notes", []))] worker["stallCount"] = int(worker.get("stallCount", 0) or 0) worker["interventionCount"] = int(worker.get("interventionCount", 0) or 0) return worker def _default_state(project_root: Path) -> Dict[str, object]: return { "enabled": False, "updatedAt": "", "projectRoot": str(project_root), "artifactHealth": artifact_health(project_root), "mergedResults": [], "pendingGlobalUpdates": [], "planningSource": "", "reviewSourcePath": "", "debugPolicy": normalize_debug_policy(DEFAULT_DEBUG_POLICY), "debugSessions": [], "taskDebugCounts": {}, "xdbPolicy": normalize_xdb_policy(DEFAULT_XDB_POLICY), "xdbSessions": [], "taskXdbCounts": {}, "repairPolicy": normalize_repair_policy(), "repairAttempts": [], "activeRepairCount": 0, "engineMode": "idle", "autonomyPolicy": _default_autonomy_policy(), "monitoringPolicy": _default_monitoring_policy(), "activeWaveId": "", "activeDispatchPath": "", "activeWorkers": [], "dispatchHistory": [], "dispatchedGroups": [], "nextAction": "plan-or-dispatch", "nextActionDetails": {"type": "plan-or-dispatch"}, "lastLoopAt": "", "lastInterventionAt": "", "interventionHistory": [], } def _ensure_state_defaults(project_root: Path, state: Dict[str, object]) -> Dict[str, object]: merged = _default_state(project_root) merged.update(state) merged["artifactHealth"] = artifact_health(project_root) merged["debugPolicy"] = normalize_debug_policy(dict(merged.get("debugPolicy", {}))) merged["xdbPolicy"] = normalize_xdb_policy(dict(merged.get("xdbPolicy", {}))) merged["repairPolicy"] = normalize_repair_policy(dict(merged.get("repairPolicy", {}))) merged["autonomyPolicy"] = { **_default_autonomy_policy(), **dict(merged.get("autonomyPolicy", {})), } merged["monitoringPolicy"] = { **_default_monitoring_policy(), **dict(merged.get("monitoringPolicy", {})), } merged.setdefault("debugSessions", []) merged.setdefault("taskDebugCounts", {}) merged.setdefault("xdbSessions", []) merged.setdefault("taskXdbCounts", {}) merged.setdefault("mergedResults", []) merged.setdefault("pendingGlobalUpdates", []) merged.setdefault("repairAttempts", []) merged.setdefault("activeRepairCount", 0) merged.setdefault("dispatchHistory", []) merged.setdefault("dispatchedGroups", []) merged.setdefault("interventionHistory", []) merged["activeWorkers"] = [ _normalize_active_worker(item) for item in list(merged.get("activeWorkers", [])) if isinstance(item, dict) ] merged.setdefault("nextAction", "plan-or-dispatch") merged.setdefault("nextActionDetails", {"type": str(merged.get("nextAction", "plan-or-dispatch"))}) return merged def artifact_health(project_root: Path) -> Dict[str, bool]: health: Dict[str, bool] = {} for relative in REQUIRED_PROJECT_ARTIFACTS: health[relative] = (project_root / relative).exists() return health def _engine_paths(project_root: Path) -> Dict[str, Path]: engine_root = aireng_root(project_root) return { "root": engine_root, "state": engine_root / "state.json", "results": engine_root / "results", "reviews": engine_root / "reviews", "checkpoints": engine_root / "checkpoints", "dispatch": engine_root / "dispatch", "plan": engine_root / "plan.json", "plan_md": engine_root / "plan.md", "doc_queue_md": engine_root / "doc-update-queue.md", } def _ensure_layout(project_root: Path) -> Dict[str, Path]: paths = _engine_paths(project_root) paths["results"].mkdir(parents=True, exist_ok=True) paths["reviews"].mkdir(parents=True, exist_ok=True) paths["checkpoints"].mkdir(parents=True, exist_ok=True) paths["dispatch"].mkdir(parents=True, exist_ok=True) return paths def _set_next_action(state: Dict[str, object], action: str, **details: object) -> Dict[str, object]: state["nextAction"] = action state["nextActionDetails"] = {"type": action, **details} return state def _mtime_iso(path: Path) -> str: if not path.exists(): return "" return datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc).isoformat() def _seconds_since_mtime(path: Path) -> float | None: if not path.exists(): return None return max(datetime.now(timezone.utc).timestamp() - path.stat().st_mtime, 0.0) def _sanitize_stamp(value: str) -> str: return value.replace(":", "-").replace("+", "_") def _next_dispatch_group_name(state: Dict[str, object], plan_payload: Dict[str, object]) -> str: dispatched_groups = {str(item).strip() for item in list(state.get("dispatchedGroups", [])) if str(item).strip()} for group in list(plan_payload.get("parallelGroups", [])): if not isinstance(group, dict): continue group_name = str(group.get("name", "")).strip() if group_name and group_name not in dispatched_groups: return group_name return "" def _refresh_engine_progress(project_root: Path, state: Dict[str, object]) -> Dict[str, object]: plan_payload = _json_load(_engine_paths(project_root)["plan"]) active_workers = [ _normalize_active_worker(item) for item in list(state.get("activeWorkers", [])) if isinstance(item, dict) ] state["activeWorkers"] = active_workers blocked_workers = [item for item in active_workers if str(item.get("status", "")) == "blocked"] if blocked_workers: state["engineMode"] = "blocked" return _set_next_action( state, "user-decision-required", blockedTaskIds=[item.get("taskId", "") for item in blocked_workers], ) if active_workers: state["engineMode"] = "monitoring" return _set_next_action( state, "monitor-workers", activeWorkerCount=len(active_workers), checkIntervalSeconds=int(state.get("monitoringPolicy", {}).get("checkIntervalSeconds", 300) or 300), ) next_group_name = _next_dispatch_group_name(state, plan_payload) if next_group_name: state["engineMode"] = "idle" return _set_next_action(state, "dispatch-next-wave", groupName=next_group_name) state["engineMode"] = "completed" state["activeWaveId"] = "" state["activeDispatchPath"] = "" return _set_next_action(state, "completed", mergedResultCount=len(list(state.get("mergedResults", [])))) def enter_engine(project_root: Path) -> Tuple[Path, Dict[str, bool]]: paths = _ensure_layout(project_root) state = _ensure_state_defaults(project_root, {}) state["enabled"] = True state["updatedAt"] = now_iso() _set_next_action(state, "plan-or-dispatch") _json_dump(paths["state"], state) _write_doc_queue(paths["doc_queue_md"], []) return paths["state"], state["artifactHealth"] def status_engine(project_root: Path) -> Dict[str, object]: paths = _ensure_layout(project_root) state = _json_load(paths["state"]) return _ensure_state_defaults(project_root, state) def _arc_review_paths(project_root: Path) -> Dict[str, Path]: arc_root = airarc_root(project_root) / "reviews" return { "review_json": arc_root / "parallel-review.json", "execution_plan_json": arc_root / "execution-plan.json", } def _load_parallel_review(path: Path) -> ParallelReview: payload = json.loads(path.read_text(encoding="utf-8-sig")) if "parallelReview" in payload: payload = payload["parallelReview"] return ParallelReview.from_dict(payload) def _resolve_review_source(project_root: Path, todo_path: Path) -> Tuple[ParallelReview, str, str]: arc_paths = _arc_review_paths(project_root) if arc_paths["execution_plan_json"].exists(): review = _load_parallel_review(arc_paths["execution_plan_json"]) return review, "airarc-execution-plan", str(arc_paths["execution_plan_json"]) if arc_paths["review_json"].exists(): review = _load_parallel_review(arc_paths["review_json"]) return review, "airarc-review", str(arc_paths["review_json"]) review = build_parallel_review(todo_path) return review, "engine-fallback-analysis", "" def _render_engine_plan_markdown(plan_payload: Dict[str, object]) -> str: lines = [ "# AirEng Execution Plan", "", f"- Generated At: `{plan_payload['generatedAt']}`", f"- Project Root: `{plan_payload['projectRoot']}`", f"- Todo Path: `{plan_payload['todoPath']}`", f"- Planning Source: `{plan_payload['planningSource']}`", f"- Review Source Path: `{plan_payload['reviewSourcePath'] or '(generated by engine fallback)'}`", "", "## Selected Tasks", ] selected_tasks = plan_payload.get("selectedTasks", []) if selected_tasks: for task_id in selected_tasks: lines.append(f"- `{task_id}`") else: lines.append("- No selected tasks.") lines.extend(["", "## Parallel Groups"]) parallel_groups = plan_payload.get("parallelGroups", []) if parallel_groups: for group in parallel_groups: lines.append(f"- `{group['name']}`: {', '.join(group['taskIds'])}") lines.append(f" Reason: {group['reason']}") else: lines.append("- No parallel groups.") lines.extend(["", "## Serialization Points"]) serialization_points = plan_payload.get("serializationPoints", []) if serialization_points: for item in serialization_points: lines.append(f"- `{item['taskId']}`: {'; '.join(item['reasons'])}") else: lines.append("- No serialization points.") return "\n".join(lines) + "\n" def _write_doc_queue(path: Path, pending_updates: list[Dict[str, object]]) -> None: lines = [ "# AirEng Doc Update Queue", "", "This file tracks global document updates that workers recommended but did not apply directly.", "", ] if pending_updates: for item in pending_updates: lines.append(f"- Task: `{item['taskId']}`") recommended = item.get("recommendedUpdates", []) global_paths = item.get("globalDocPaths", []) result_path = item.get("resultPath", "") lines.append( " Recommended Updates: " + (", ".join(f"`{value}`" for value in recommended) if recommended else "`none`") ) lines.append( " Global Doc Paths: " + (", ".join(f"`{value}`" for value in global_paths) if global_paths else "`none`") ) lines.append(f" Result Path: `{result_path}`") else: lines.append("- No pending global document updates.") path.write_text("\n".join(lines) + "\n", encoding="utf-8") def build_engine_plan(project_root: Path, todo_path: Path) -> Dict[str, object]: paths = _ensure_layout(project_root) review, planning_source, review_source_path = _resolve_review_source(project_root, todo_path) review_json_path = paths["reviews"] / "parallel-review.json" review_md_path = paths["reviews"] / "parallel-review.md" _json_dump(review_json_path, review.to_dict()) review_md_path.write_text(render_review_markdown(review), encoding="utf-8") selected_tasks = review.parallel_groups[0].task_ids if review.parallel_groups else [] plan_payload = { "generatedAt": now_iso(), "projectRoot": str(project_root), "todoPath": str(todo_path), "planningSource": planning_source, "reviewSourcePath": review_source_path, "selectedTasks": selected_tasks, "parallelGroups": [group.to_dict() for group in review.parallel_groups], "conflicts": [conflict.to_dict() for conflict in review.conflicts], "serializationPoints": review.serialization_points, "reviewJsonPath": str(review_json_path), "reviewMarkdownPath": str(review_md_path), } _json_dump(paths["plan"], plan_payload) paths["plan_md"].write_text(_render_engine_plan_markdown(plan_payload), encoding="utf-8") state = status_engine(project_root) state["enabled"] = True state["updatedAt"] = now_iso() state["engineMode"] = "planning" state["lastPlanPath"] = str(paths["plan"]) state["lastPlanMarkdownPath"] = str(paths["plan_md"]) state["lastReviewJsonPath"] = str(review_json_path) state["lastReviewMarkdownPath"] = str(review_md_path) state["planningSource"] = planning_source state["reviewSourcePath"] = review_source_path _refresh_engine_progress(project_root, state) if not list(state.get("activeWorkers", [])): next_group_name = _next_dispatch_group_name(state, plan_payload) if next_group_name: state["engineMode"] = "planning" _set_next_action(state, "dispatch-next-wave", groupName=next_group_name) _json_dump(paths["state"], state) _write_doc_queue(paths["doc_queue_md"], list(state.get("pendingGlobalUpdates", []))) return { "planPath": str(paths["plan"]), "planMarkdownPath": str(paths["plan_md"]), "reviewJsonPath": str(review_json_path), "reviewMarkdownPath": str(review_md_path), "planningSource": planning_source, "reviewSourcePath": review_source_path, "selectedTasks": selected_tasks, "parallelGroupCount": len(review.parallel_groups), "conflictCount": len(review.conflicts), } def _build_active_worker_records( dispatched_tasks: List[Dict[str, object]], group_name: str, wave_id: str ) -> List[Dict[str, object]]: records: List[Dict[str, object]] = [] for task in dispatched_tasks: record = _default_active_worker() record.update( { "taskId": str(task.get("taskId", "")).strip(), "dispatchGroup": group_name, "waveId": wave_id, "briefPath": str(task.get("briefPath", "")).strip(), "handoffPath": str(task.get("handoffPath", "")).strip(), "workerStatePath": str(task.get("workerStatePath", "")).strip(), "resultPath": str(task.get("resultPath", "")).strip(), "spawnedAt": now_iso(), "lastObservedAt": "", "lastHeartbeatAt": _mtime_iso(Path(str(task.get("workerStatePath", "")).strip())) if str(task.get("workerStatePath", "")).strip() else "", "status": "dispatched", "lastResultPath": "", "stallCount": 0, "interventionCount": 0, "repairDispatchPath": "", "notes": [], } ) records.append(record) return records def dispatch_worker_group(project_root: Path, group_name: str = "") -> Dict[str, object]: paths = _ensure_layout(project_root) plan_payload = _json_load(paths["plan"]) if not plan_payload: raise ValueError("engine plan is missing; run plan mode before dispatch") parallel_groups = list(plan_payload.get("parallelGroups", [])) if not parallel_groups: raise ValueError("no parallel groups available for dispatch") selected_group = None if group_name: for group in parallel_groups: if group.get("name") == group_name: selected_group = group break if selected_group is None: raise ValueError(f"parallel group not found: {group_name}") else: next_group_name = _next_dispatch_group_name(status_engine(project_root), plan_payload) for group in parallel_groups: if group.get("name") == next_group_name: selected_group = group break if selected_group is None: selected_group = parallel_groups[0] current_todo_path = workflow_todo_path(project_root) task_map = {} if current_todo_path.exists(): try: task_map = {task.task_id: task for task in parse_tasks(current_todo_path)} except ValueError: task_map = {} task_ids = [str(task_id) for task_id in selected_group.get("taskIds", [])] recommended_concurrency = min(max(len(task_ids), 1), 3) dispatched_tasks = [] for task_id in task_ids: prepared = enter_worker(project_root, task_id) task_record = task_map.get(task_id) dispatched_tasks.append( { "taskId": task_id, "module": task_record.module if task_record else "", "task": task_record.task if task_record else "", "writePaths": list(task_record.write_paths) if task_record else [], "globalDocPaths": list(task_record.global_doc_paths) if task_record else [], "dependencies": list(task_record.dependencies) if task_record else [], "briefPath": prepared["briefPath"], "handoffPath": prepared["handoffPath"], "resultPath": prepared["resultPath"], "workerStatePath": prepared["workerStatePath"], "subagentCommand": "/airdo", "requiresIsolatedContext": True, "recommendedAgentType": "worker", } ) timestamp = now_iso() selected_group_name = str(selected_group.get("name", "")).strip() or "group" wave_id = f"{selected_group_name}-{_sanitize_stamp(timestamp)}" manifest = { "generatedAt": timestamp, "projectRoot": str(project_root), "groupName": selected_group_name, "waveId": wave_id, "taskIds": task_ids, "planningSource": plan_payload.get("planningSource", ""), "reviewSourcePath": plan_payload.get("reviewSourcePath", ""), "dispatchMode": "isolated-airdo-subagents", "workerCommand": "/airdo", "requiresIsolatedContext": True, "recommendedConcurrency": recommended_concurrency, "dispatchedTasks": dispatched_tasks, } manifest_path = paths["dispatch"] / f"{selected_group_name}.json" _json_dump(manifest_path, manifest) dispatch_doc_paths = mark_tasks_dispatched( project_root, selected_group_name, dispatched_tasks, recommended_concurrency, ) state = status_engine(project_root) state["enabled"] = True state["updatedAt"] = now_iso() state["lastLoopAt"] = state["updatedAt"] state["engineMode"] = "dispatching" state["activeWaveId"] = wave_id state["activeDispatchPath"] = str(manifest_path) state["activeWorkers"] = _build_active_worker_records(dispatched_tasks, selected_group_name, wave_id) state["lastDispatchPath"] = str(manifest_path) state["lastDispatchGroup"] = selected_group_name state["lastDispatchDocPaths"] = dispatch_doc_paths dispatched_groups = [str(item) for item in list(state.get("dispatchedGroups", [])) if str(item).strip()] if selected_group_name not in dispatched_groups: dispatched_groups.append(selected_group_name) state["dispatchedGroups"] = dispatched_groups history = [item for item in list(state.get("dispatchHistory", [])) if isinstance(item, dict)] history.append( { "waveId": wave_id, "groupName": selected_group_name, "dispatchPath": str(manifest_path), "taskIds": task_ids, "dispatchedAt": state["updatedAt"], } ) state["dispatchHistory"] = history[-20:] _set_next_action( state, "monitor-workers", activeWorkerCount=len(state["activeWorkers"]), checkIntervalSeconds=int(state.get("monitoringPolicy", {}).get("checkIntervalSeconds", 300) or 300), dispatchPath=str(manifest_path), waveId=wave_id, ) _json_dump(paths["state"], state) return { "dispatchPath": str(manifest_path), "groupName": selected_group_name, "waveId": wave_id, "taskIds": task_ids, "recommendedConcurrency": recommended_concurrency, } def _prepare_worker_for_repair(project_root: Path, task_id: str) -> Dict[str, object]: prepared = enter_worker(project_root, str(task_id)) return { "taskId": str(task_id), "prepared": prepared, "stdout": json.dumps(prepared, ensure_ascii=False), } def _write_repair_dispatch_manifest( paths: Dict[str, Path], project_root: Path, task_id: str, prepared_stdout: str ) -> Path: active_attempt = load_active_repair_attempt(project_root, task_id) manifest_path = paths["dispatch"] / f"repair-{task_id}.json" manifest = { "generatedAt": now_iso(), "projectRoot": str(project_root), "taskId": task_id, "repairAttemptId": active_attempt.repair_id if active_attempt else "", "repairBriefPath": active_attempt.repair_brief_path if active_attempt else "", "workerBriefPath": active_attempt.worker_brief_path if active_attempt else "", "command": f"/airdo handoff {task_id}", "prepareStdout": prepared_stdout, "nextAction": "continue-repair", } _json_dump(manifest_path, manifest) return manifest_path def _inspect_worker_record( project_root: Path, worker: Dict[str, object], monitoring_policy: Dict[str, object] ) -> Dict[str, object]: record = _normalize_active_worker(worker) record["lastObservedAt"] = now_iso() task_id = str(record.get("taskId", "")).strip() worker_state_path = Path(str(record.get("workerStatePath", "")).strip()) if str(record.get("workerStatePath", "")).strip() else None result_path = Path(str(record.get("resultPath", "")).strip()) if str(record.get("resultPath", "")).strip() else None if worker_state_path and worker_state_path.exists(): record["lastHeartbeatAt"] = _mtime_iso(worker_state_path) payload = _json_load(worker_state_path) payload_status = str(payload.get("status", "")).strip() payload_result_path = str(payload.get("resultPath", "")).strip() if payload_result_path: record["resultPath"] = payload_result_path result_path = Path(payload_result_path) if payload_status: record["status"] = payload_status if payload_status == "completed" and result_path and result_path.exists(): record["status"] = "ready_to_merge" record["lastResultPath"] = str(result_path) return { "taskId": task_id, "classification": "ready_to_merge", "worker": record, "resultPath": str(result_path), } active_attempt = load_active_repair_attempt(project_root, task_id) if task_id else None if active_attempt is not None: record["status"] = f"repair-{active_attempt.status}" return { "taskId": task_id, "classification": "repairing", "worker": record, "repairAttemptId": active_attempt.repair_id, } if worker_state_path and worker_state_path.exists(): stall_after_seconds = int(monitoring_policy.get("stallAfterSeconds", 1800) or 1800) age_seconds = _seconds_since_mtime(worker_state_path) if age_seconds is not None and age_seconds >= stall_after_seconds: record["status"] = "stalled" record["stallCount"] = int(record.get("stallCount", 0) or 0) + 1 return { "taskId": task_id, "classification": "stalled", "worker": record, "ageSeconds": int(age_seconds), } if worker_state_path and not worker_state_path.exists(): record["status"] = "stalled" record["stallCount"] = int(record.get("stallCount", 0) or 0) + 1 return { "taskId": task_id, "classification": "stalled", "worker": record, "ageSeconds": None, } record["status"] = str(record.get("status", "active") or "active") return { "taskId": task_id, "classification": "active", "worker": record, } def _record_intervention(task_id: str, reason: str, action: str, outcome: str) -> Dict[str, object]: return { "taskId": task_id, "reason": reason, "action": action, "outcome": outcome, "at": now_iso(), } def _handle_stalled_workers( project_root: Path, state: Dict[str, object], stalled_workers: List[Dict[str, object]], ) -> Tuple[Dict[str, object], List[Dict[str, object]], List[str]]: if not stalled_workers: return state, [], [] monitoring_policy = dict(state.get("monitoringPolicy", {})) max_interventions = int(monitoring_policy.get("maxInterventionAttemptsPerTask", 2) or 2) stalled_by_task = {str(item.get("taskId", "")): item for item in stalled_workers} intervention_history = [item for item in list(state.get("interventionHistory", [])) if isinstance(item, dict)] interventions: List[Dict[str, object]] = [] blocked_tasks: List[str] = [] updated_workers: List[Dict[str, object]] = [] for worker in [ _normalize_active_worker(item) for item in list(state.get("activeWorkers", [])) if isinstance(item, dict) ]: task_id = str(worker.get("taskId", "")).strip() stalled = stalled_by_task.get(task_id) if stalled is None: updated_workers.append(worker) continue intervention_count = int(worker.get("interventionCount", 0) or 0) if intervention_count >= max_interventions: worker["status"] = "blocked" blocked_tasks.append(task_id) interventions.append( _record_intervention( task_id, "worker stalled beyond intervention budget", "escalate-to-user", "blocked", ) ) updated_workers.append(worker) continue prepared = enter_worker(project_root, task_id) worker.update( { "briefPath": prepared["briefPath"], "handoffPath": prepared["handoffPath"], "workerStatePath": prepared["workerStatePath"], "resultPath": prepared["resultPath"], "status": "redispatched", "stallCount": 0, "interventionCount": intervention_count + 1, "lastObservedAt": now_iso(), "lastHeartbeatAt": _mtime_iso(Path(prepared["workerStatePath"])), "notes": list(worker.get("notes", [])) + ["redispatched after stall detection"], } ) interventions.append( _record_intervention( task_id, "worker stalled or stopped updating state", "redispatch-worker", "monitor-again", ) ) updated_workers.append(worker) if interventions: intervention_history.extend(interventions) state["interventionHistory"] = intervention_history[-50:] state["lastInterventionAt"] = interventions[-1]["at"] state["activeWorkers"] = updated_workers return state, interventions, blocked_tasks def merge_worker_result(project_root: Path, result_path: Path) -> Dict[str, object]: paths = _ensure_layout(project_root) result = WorkerResult.from_dict(json.loads(result_path.read_text(encoding="utf-8-sig"))) result.validate() ensure_xdb_sessions_for_result(project_root, result, "engine-merge-fallback") ensure_debug_sessions_for_result(project_root, result, "engine-merge-fallback") ensure_repair_attempts_for_result(project_root, result, str(result_path)) result.validate() enforce_doc_sync_requirements(project_root, result) timestamp = now_iso().replace(":", "-") archived_path = paths["results"] / f"{result.task_id}-{timestamp}.json" if not result.finalized_at: result.finalized_at = now_iso() _json_dump(archived_path, result.to_dict()) state = status_engine(project_root) state["enabled"] = True state["updatedAt"] = now_iso() state["lastLoopAt"] = state["updatedAt"] state["engineMode"] = "merging" merged_results = list(state.get("mergedResults", [])) merged_results.append( { "taskId": result.task_id, "status": result.status, "summary": result.summary, "archivedResultPath": str(archived_path), } ) state["mergedResults"] = merged_results merge_xdb_sessions_into_state(state, result.xdb_sessions) merge_debug_sessions_into_state(state, result.debug_sessions) merge_repair_attempts_into_state(state, result.repair_attempts) repair_prepared = False repair_prepare_stdout = "" repair_dispatch_path = "" repair_worker_payload: Dict[str, str] = {} if result.status != "done" and result.repair_attempts: repair_policy = normalize_repair_policy(dict(state.get("repairPolicy", {}))) if bool(repair_policy.get("autoPrepareWorker", True)): prepared = _prepare_worker_for_repair(project_root, result.task_id) repair_prepared = True repair_prepare_stdout = str(prepared["stdout"]) repair_worker_payload = dict(prepared.get("prepared", {})) result.repair_attempts = ( mark_repair_attempts_active(project_root, result.task_id) or result.repair_attempts ) merge_repair_attempts_into_state(state, result.repair_attempts) repair_dispatch = _write_repair_dispatch_manifest( paths, project_root, result.task_id, repair_prepare_stdout ) repair_dispatch_path = str(repair_dispatch) state["lastRepairDispatchPath"] = repair_dispatch_path pending_updates = list(state.get("pendingGlobalUpdates", [])) applied_doc_paths = apply_document_updates(project_root, result) sync_paths = sync_engine_managed_docs(project_root, result, applied_doc_paths) merged_todo_path = update_todo_after_merge(project_root, result, applied_doc_paths, sync_paths) if result.recommend_global_doc_updates and not result.document_updates: pending_updates.append( { "taskId": result.task_id, "recommendedUpdates": result.recommend_global_doc_updates, "globalDocPaths": result.global_doc_paths, "resultPath": str(archived_path), } ) state["pendingGlobalUpdates"] = pending_updates remaining_workers: List[Dict[str, object]] = [] for item in [ _normalize_active_worker(worker) for worker in list(state.get("activeWorkers", [])) if isinstance(worker, dict) ]: if str(item.get("taskId", "")).strip() != result.task_id: remaining_workers.append(item) continue if repair_prepared: item.update( { "briefPath": repair_worker_payload.get("briefPath", item.get("briefPath", "")), "handoffPath": repair_worker_payload.get("handoffPath", item.get("handoffPath", "")), "workerStatePath": repair_worker_payload.get("workerStatePath", item.get("workerStatePath", "")), "resultPath": repair_worker_payload.get("resultPath", item.get("resultPath", "")), "status": "repair-dispatched", "lastResultPath": str(archived_path), "repairDispatchPath": repair_dispatch_path, "stallCount": 0, "lastObservedAt": now_iso(), "lastHeartbeatAt": _mtime_iso(Path(repair_worker_payload.get("workerStatePath", ""))) if repair_worker_payload.get("workerStatePath") else item.get("lastHeartbeatAt", ""), } ) remaining_workers.append(item) state["activeWorkers"] = remaining_workers _refresh_engine_progress(project_root, state) if repair_prepared: state["engineMode"] = "monitoring" _set_next_action( state, "continue-repair", taskId=result.task_id, repairDispatchPath=repair_dispatch_path, ) _json_dump(paths["state"], state) _write_doc_queue(paths["doc_queue_md"], pending_updates) repair_queue_path = write_repair_queue(project_root, state) return { "taskId": result.task_id, "status": result.status, "archivedResultPath": str(archived_path), "pendingGlobalUpdateCount": len(pending_updates), "docQueuePath": str(paths["doc_queue_md"]), "repairQueuePath": str(repair_queue_path), "todoPath": str(merged_todo_path), "appliedDocPathCount": len(applied_doc_paths) + len(sync_paths), "xdbSessionCount": len(result.xdb_sessions), "debugSessionCount": len(result.debug_sessions), "repairAttemptCount": len(result.repair_attempts), "repairPrepared": repair_prepared, "repairPrepareStdout": repair_prepare_stdout, "repairDispatchPath": repair_dispatch_path, "repairWorkerStatePath": repair_worker_payload.get("workerStatePath", ""), "repairHandoffPath": repair_worker_payload.get("handoffPath", ""), "repairResultPath": repair_worker_payload.get("resultPath", ""), "nextAction": str(state.get("nextAction", "")), } def monitor_engine(project_root: Path) -> Dict[str, object]: paths = _ensure_layout(project_root) state = status_engine(project_root) state["enabled"] = True state["updatedAt"] = now_iso() state["lastLoopAt"] = state["updatedAt"] state["engineMode"] = "monitoring" monitoring_policy = dict(state.get("monitoringPolicy", {})) inspected = [ _inspect_worker_record(project_root, worker, monitoring_policy) for worker in list(state.get("activeWorkers", [])) if isinstance(worker, dict) ] state["activeWorkers"] = [item["worker"] for item in inspected] _json_dump(paths["state"], state) ready_to_merge = [item for item in inspected if item.get("classification") == "ready_to_merge"] merged: List[Dict[str, object]] = [] for item in ready_to_merge: result_path = Path(str(item.get("resultPath", "")).strip()) if result_path.exists(): merged.append(merge_worker_result(project_root, result_path)) state = status_engine(project_root) monitoring_policy = dict(state.get("monitoringPolicy", {})) reinspected = [ _inspect_worker_record(project_root, worker, monitoring_policy) for worker in list(state.get("activeWorkers", [])) if isinstance(worker, dict) ] state["activeWorkers"] = [item["worker"] for item in reinspected] stalled_workers = [item for item in reinspected if item.get("classification") == "stalled"] state, interventions, blocked_tasks = _handle_stalled_workers(project_root, state, stalled_workers) state["updatedAt"] = now_iso() state["lastLoopAt"] = state["updatedAt"] _refresh_engine_progress(project_root, state) if blocked_tasks: state["engineMode"] = "blocked" _set_next_action(state, "user-decision-required", blockedTaskIds=blocked_tasks) _json_dump(paths["state"], state) _write_doc_queue(paths["doc_queue_md"], list(state.get("pendingGlobalUpdates", []))) repair_queue_path = write_repair_queue(project_root, state) return { "engineMode": str(state.get("engineMode", "")), "activeWorkerCount": len(list(state.get("activeWorkers", []))), "readyToMergeCount": len(ready_to_merge), "mergedCount": len(merged), "stalledCount": len(stalled_workers), "interventionCount": len(interventions), "blockedTaskCount": len(blocked_tasks), "repairQueuePath": str(repair_queue_path), "nextAction": str(state.get("nextAction", "")), } def intervene_engine(project_root: Path) -> Dict[str, object]: paths = _ensure_layout(project_root) state = status_engine(project_root) monitoring_policy = dict(state.get("monitoringPolicy", {})) inspected = [ _inspect_worker_record(project_root, worker, monitoring_policy) for worker in list(state.get("activeWorkers", [])) if isinstance(worker, dict) ] state["activeWorkers"] = [item["worker"] for item in inspected] stalled_workers = [item for item in inspected if item.get("classification") == "stalled"] state["engineMode"] = "intervening" state["updatedAt"] = now_iso() state["lastLoopAt"] = state["updatedAt"] state, interventions, blocked_tasks = _handle_stalled_workers(project_root, state, stalled_workers) _refresh_engine_progress(project_root, state) if blocked_tasks: state["engineMode"] = "blocked" _set_next_action(state, "user-decision-required", blockedTaskIds=blocked_tasks) _json_dump(paths["state"], state) return { "engineMode": str(state.get("engineMode", "")), "stalledCount": len(stalled_workers), "interventionCount": len(interventions), "blockedTaskCount": len(blocked_tasks), "nextAction": str(state.get("nextAction", "")), } def run_engine_once(project_root: Path, todo_path: Path | None = None) -> Dict[str, object]: paths = _ensure_layout(project_root) steps: List[str] = [] plan_payload = _json_load(paths["plan"]) if not plan_payload: plan_result = build_engine_plan(project_root, todo_path or workflow_todo_path(project_root)) steps.append("plan") plan_payload = _json_load(paths["plan"]) else: plan_result = { "planPath": str(paths["plan"]), "planMarkdownPath": str(paths["plan_md"]), } state = status_engine(project_root) if list(state.get("activeWorkers", [])): monitor_result = monitor_engine(project_root) steps.append("monitor") return { "action": "monitor", "steps": steps, "planPath": str(plan_result.get("planPath", paths["plan"])), "activeWorkerCount": monitor_result["activeWorkerCount"], "nextAction": monitor_result["nextAction"], "engineMode": monitor_result["engineMode"], } next_group_name = _next_dispatch_group_name(state, plan_payload) if next_group_name: dispatch_result = dispatch_worker_group(project_root, next_group_name) steps.append("dispatch") refreshed_state = status_engine(project_root) return { "action": "dispatch", "steps": steps, "planPath": str(plan_result.get("planPath", paths["plan"])), "dispatchPath": dispatch_result["dispatchPath"], "waveId": dispatch_result["waveId"], "taskIds": dispatch_result["taskIds"], "nextAction": str(refreshed_state.get("nextAction", "")), "engineMode": str(refreshed_state.get("engineMode", "")), } state["enabled"] = True state["updatedAt"] = now_iso() state["lastLoopAt"] = state["updatedAt"] _refresh_engine_progress(project_root, state) _json_dump(paths["state"], state) steps.append("complete") return { "action": "complete", "steps": steps, "planPath": str(plan_result.get("planPath", paths["plan"])), "nextAction": str(state.get("nextAction", "")), "engineMode": str(state.get("engineMode", "")), }