ae44be31d5
Includes AirPlan design documents, AircOding-alpha1-plan, AirPlanV2, AirPlan-ParaV2, AirPlan-Para V1 reference docs, and all working code changes across packages. Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
1083 lines
42 KiB
Python
Executable File
1083 lines
42 KiB
Python
Executable File
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", "")),
|
|
}
|