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>
374 lines
14 KiB
Python
Executable File
374 lines
14 KiB
Python
Executable File
from __future__ import annotations
|
|
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Dict, Iterable, List
|
|
|
|
from .contracts import DEFAULT_REPAIR_POLICY, RepairAttempt, WorkerResult, now_iso
|
|
from .paths import (
|
|
airdo_root,
|
|
aireng_root,
|
|
debug_log_path,
|
|
gui_debug_log_path,
|
|
plan_path,
|
|
todo_path,
|
|
)
|
|
|
|
|
|
def _json_load(path: Path) -> Dict[str, object]:
|
|
if not path.exists():
|
|
return {}
|
|
return json.loads(path.read_text(encoding="utf-8-sig"))
|
|
|
|
|
|
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 _ordered_unique(items: Iterable[str]) -> List[str]:
|
|
seen = set()
|
|
ordered: List[str] = []
|
|
for item in items:
|
|
cleaned = str(item).strip()
|
|
if not cleaned or cleaned in seen:
|
|
continue
|
|
seen.add(cleaned)
|
|
ordered.append(cleaned)
|
|
return ordered
|
|
|
|
|
|
def normalize_repair_policy(policy: Dict[str, object] | None = None) -> Dict[str, object]:
|
|
merged = dict(DEFAULT_REPAIR_POLICY)
|
|
if policy:
|
|
merged.update(policy)
|
|
|
|
merged["enabled"] = bool(merged.get("enabled", True))
|
|
merged["triggerOnBlocked"] = bool(merged.get("triggerOnBlocked", True))
|
|
merged["autoPrepareWorker"] = bool(merged.get("autoPrepareWorker", True))
|
|
merged["triggerValidationStatuses"] = _ordered_unique(
|
|
str(item).lower() for item in list(merged.get("triggerValidationStatuses", []))
|
|
)
|
|
try:
|
|
max_attempts = int(merged.get("maxAttemptsPerTask", 2) or 2)
|
|
except (TypeError, ValueError):
|
|
max_attempts = 2
|
|
merged["maxAttemptsPerTask"] = max(1, max_attempts)
|
|
|
|
requeue_status = str(merged.get("requeueTodoStatus", "DOING")).strip().upper()
|
|
if requeue_status not in {"TODO", "DOING"}:
|
|
requeue_status = "DOING"
|
|
merged["requeueTodoStatus"] = requeue_status
|
|
return merged
|
|
|
|
|
|
def load_repair_policy(project_root: Path) -> Dict[str, object]:
|
|
state_path = aireng_root(project_root) / "state.json"
|
|
state = _json_load(state_path)
|
|
return normalize_repair_policy(state.get("repairPolicy", {}))
|
|
|
|
|
|
def _repair_root(project_root: Path) -> Path:
|
|
return aireng_root(project_root) / "repairs"
|
|
|
|
|
|
def _repair_queue_path(project_root: Path) -> Path:
|
|
return aireng_root(project_root) / "repair-queue.md"
|
|
|
|
|
|
def _repair_attempt_path(project_root: Path, task_id: str, repair_id: str) -> Path:
|
|
return _repair_root(project_root) / task_id / repair_id / "attempt.json"
|
|
|
|
|
|
def _repair_brief_path(project_root: Path, task_id: str, repair_id: str) -> Path:
|
|
return _repair_root(project_root) / task_id / repair_id / "repair-brief.md"
|
|
|
|
|
|
def _worker_repair_brief_path(project_root: Path, task_id: str) -> Path:
|
|
return airdo_root(project_root) / "tasks" / task_id / "repair-brief.md"
|
|
|
|
|
|
def _repair_reasons(result: WorkerResult, policy: Dict[str, object]) -> List[str]:
|
|
reasons: List[str] = []
|
|
if result.status == "blocked" and bool(policy.get("triggerOnBlocked", True)):
|
|
reasons.append("blocked worker result requires automatic repair")
|
|
tracked_statuses = {
|
|
str(item).lower() for item in list(policy.get("triggerValidationStatuses", []))
|
|
}
|
|
failed_validations = [
|
|
item for item in result.validations if item.status.lower() in tracked_statuses
|
|
]
|
|
if failed_validations:
|
|
reasons.append(
|
|
"validation failures: "
|
|
+ ", ".join(f"{item.kind}:{item.status}" for item in failed_validations)
|
|
)
|
|
return reasons
|
|
|
|
|
|
def _build_repair_brief(project_root: Path, result: WorkerResult, attempt: RepairAttempt) -> str:
|
|
debug_session_ids = ", ".join(f"`{item}`" for item in attempt.debug_session_ids) or "`none`"
|
|
xdb_session_ids = ", ".join(f"`{item.session_id}`" for item in result.xdb_sessions) or "`none`"
|
|
validations = (
|
|
"; ".join(
|
|
f"{item.kind}:{item.status}" + (f" ({item.command})" if item.command else "")
|
|
for item in result.validations
|
|
)
|
|
or "none"
|
|
)
|
|
evidence = ", ".join(f"`{item}`" for item in result.evidence_paths) or "`none`"
|
|
blockers = ", ".join(result.blockers) or "none"
|
|
risks = ", ".join(result.risks) or "none"
|
|
return "\n".join(
|
|
[
|
|
f"# Auto Repair Brief: {attempt.repair_id}",
|
|
"",
|
|
f"- Task ID: `{result.task_id}`",
|
|
f"- Attempt Index: `{attempt.attempt_index}`",
|
|
f"- Source Status: `{result.status}`",
|
|
f"- Reason: {attempt.reason}",
|
|
f"- Debug Sessions: {debug_session_ids}",
|
|
f"- Source Result Path: `{attempt.source_result_path}`",
|
|
"",
|
|
"## Required Context",
|
|
"",
|
|
f"- Load `{project_root / 'AirPlan' / 'AGENTS.md'}`, `{project_root / 'AirPlan' / 'docs' / 'architecture' / 'adr'}`, `{project_root / 'AirPlan' / 'docs' / 'architecture' / 'c4' / 'module.md'}`, `{plan_path(project_root)}`, `{todo_path(project_root)}`, `{debug_log_path(project_root)}`, and `{gui_debug_log_path(project_root)}` when present.",
|
|
"- Read the referenced AirDbg request/session artifacts before changing code.",
|
|
"- If this task has AirXDB evidence, read the GUI screenshots and report before changing code.",
|
|
"- Continue repairing automatically. Do not stop at the existence of a debug request.",
|
|
"- Only return `blocked` again when the repair budget is exhausted or a real user decision is required.",
|
|
"",
|
|
"## Failure Snapshot",
|
|
"",
|
|
f"- Summary: {result.summary}",
|
|
f"- Validations: {validations}",
|
|
f"- Blockers: {blockers}",
|
|
f"- Evidence: {evidence}",
|
|
f"- XDB Sessions: {xdb_session_ids}",
|
|
f"- Risks: {risks}",
|
|
"",
|
|
"## Repair Goal",
|
|
"",
|
|
"- Modify the scoped code until the original validation path passes again.",
|
|
"- Keep the same task id and reuse the existing debug session id in the final repaired result.",
|
|
"- Update document sync content if the final fix changes architecture or execution rules.",
|
|
"",
|
|
f"- Project Root: `{project_root}`",
|
|
]
|
|
) + "\n"
|
|
|
|
|
|
def _write_repair_attempt_artifacts(
|
|
project_root: Path, result: WorkerResult, attempt: RepairAttempt
|
|
) -> RepairAttempt:
|
|
attempt_path = Path(attempt.state_path)
|
|
brief_path = Path(attempt.repair_brief_path)
|
|
worker_brief_path = Path(attempt.worker_brief_path)
|
|
|
|
brief_content = _build_repair_brief(project_root, result, attempt)
|
|
brief_path.parent.mkdir(parents=True, exist_ok=True)
|
|
brief_path.write_text(brief_content, encoding="utf-8")
|
|
|
|
worker_brief_path.parent.mkdir(parents=True, exist_ok=True)
|
|
worker_brief_path.write_text(brief_content, encoding="utf-8")
|
|
|
|
_json_dump(
|
|
attempt_path,
|
|
{
|
|
"repairAttempt": attempt.to_dict(),
|
|
"sourceSummary": result.summary,
|
|
"sourceBlockers": result.blockers,
|
|
"sourceValidations": [item.to_dict() for item in result.validations],
|
|
"sourceEvidencePaths": result.evidence_paths,
|
|
},
|
|
)
|
|
return attempt
|
|
|
|
|
|
def list_repair_attempts(project_root: Path, task_id: str = "") -> List[RepairAttempt]:
|
|
repair_root = _repair_root(project_root)
|
|
if not repair_root.exists():
|
|
return []
|
|
|
|
attempts: List[RepairAttempt] = []
|
|
search_root = repair_root / task_id if task_id else repair_root
|
|
if not search_root.exists():
|
|
return []
|
|
|
|
for path in sorted(search_root.rglob("attempt.json")):
|
|
try:
|
|
payload = json.loads(path.read_text(encoding="utf-8-sig"))
|
|
attempt_payload = payload.get("repairAttempt", payload)
|
|
attempt = RepairAttempt.from_dict(attempt_payload)
|
|
attempt.validate()
|
|
except (json.JSONDecodeError, TypeError, ValueError):
|
|
continue
|
|
if task_id and attempt.task_id != task_id:
|
|
continue
|
|
attempts.append(attempt)
|
|
|
|
attempts.sort(key=lambda item: (item.created_at, item.repair_id))
|
|
return attempts
|
|
|
|
|
|
def load_active_repair_attempt(project_root: Path, task_id: str) -> RepairAttempt | None:
|
|
for attempt in reversed(list_repair_attempts(project_root, task_id)):
|
|
if attempt.status in {"queued", "active"}:
|
|
return attempt
|
|
return None
|
|
|
|
|
|
def _set_attempt_status(attempt: RepairAttempt, status: str) -> RepairAttempt:
|
|
attempt.status = status
|
|
attempt_path = Path(attempt.state_path)
|
|
if attempt_path.exists():
|
|
payload = json.loads(attempt_path.read_text(encoding="utf-8-sig"))
|
|
payload["repairAttempt"] = attempt.to_dict()
|
|
_json_dump(attempt_path, payload)
|
|
return attempt
|
|
|
|
|
|
def ensure_repair_attempts_for_result(
|
|
project_root: Path, result: WorkerResult, source_result_path: str = ""
|
|
) -> List[RepairAttempt]:
|
|
policy = load_repair_policy(project_root)
|
|
if not bool(policy.get("enabled", True)):
|
|
return result.repair_attempts
|
|
|
|
if result.repair_attempts:
|
|
return result.repair_attempts
|
|
|
|
existing_attempts = list_repair_attempts(project_root, result.task_id)
|
|
active_attempts = [
|
|
item for item in existing_attempts if item.status in {"queued", "active"}
|
|
]
|
|
|
|
if result.status == "done":
|
|
if active_attempts:
|
|
resolved_attempts = []
|
|
for attempt in active_attempts:
|
|
resolved_attempts.append(_set_attempt_status(attempt, "resolved"))
|
|
result.repair_attempts = resolved_attempts
|
|
return result.repair_attempts
|
|
|
|
reasons = _repair_reasons(result, policy)
|
|
if not reasons:
|
|
return result.repair_attempts
|
|
|
|
if active_attempts:
|
|
result.repair_attempts = active_attempts
|
|
return result.repair_attempts
|
|
|
|
if len(existing_attempts) >= int(policy.get("maxAttemptsPerTask", 2) or 2):
|
|
result.notes.append(
|
|
f"Automatic repair budget exhausted for {result.task_id}; user decision may be required."
|
|
)
|
|
return result.repair_attempts
|
|
|
|
attempt_index = len(existing_attempts) + 1
|
|
repair_id = f"{result.task_id}-repair-{attempt_index:03d}"
|
|
created_at = now_iso()
|
|
attempt = RepairAttempt(
|
|
repair_id=repair_id,
|
|
task_id=result.task_id,
|
|
attempt_index=attempt_index,
|
|
source_status=result.status,
|
|
source_result_path=source_result_path or "",
|
|
reason="; ".join(reasons),
|
|
repair_brief_path=str(_repair_brief_path(project_root, result.task_id, repair_id)),
|
|
state_path=str(_repair_attempt_path(project_root, result.task_id, repair_id)),
|
|
worker_brief_path=str(_worker_repair_brief_path(project_root, result.task_id)),
|
|
debug_session_ids=_ordered_unique(
|
|
item.session_id for item in result.debug_sessions if item.session_id
|
|
),
|
|
status="queued",
|
|
created_at=created_at,
|
|
)
|
|
_write_repair_attempt_artifacts(project_root, result, attempt)
|
|
result.repair_attempts = [attempt]
|
|
result.notes.append(
|
|
f"Automatic repair attempt queued for {result.task_id}: {attempt.repair_id}"
|
|
)
|
|
return result.repair_attempts
|
|
|
|
|
|
def mark_repair_attempts_active(project_root: Path, task_id: str) -> List[RepairAttempt]:
|
|
updated: List[RepairAttempt] = []
|
|
for attempt in list_repair_attempts(project_root, task_id):
|
|
if attempt.status == "queued":
|
|
updated.append(_set_attempt_status(attempt, "active"))
|
|
return updated
|
|
|
|
|
|
def merge_repair_attempts_into_state(
|
|
state: Dict[str, object], repair_attempts: List[RepairAttempt]
|
|
) -> Dict[str, object]:
|
|
state["repairPolicy"] = normalize_repair_policy(dict(state.get("repairPolicy", {})))
|
|
|
|
existing_items = list(state.get("repairAttempts", []))
|
|
items_by_id = {}
|
|
for item in existing_items:
|
|
if not isinstance(item, dict):
|
|
continue
|
|
repair_id = str(item.get("repairId", "")).strip()
|
|
if repair_id:
|
|
items_by_id[repair_id] = item
|
|
|
|
for attempt in repair_attempts:
|
|
items_by_id[attempt.repair_id] = attempt.to_dict()
|
|
|
|
merged_items = [items_by_id[key] for key in sorted(items_by_id)]
|
|
state["repairAttempts"] = merged_items
|
|
state["activeRepairCount"] = sum(
|
|
1 for item in merged_items if str(item.get("status", "")).strip() in {"queued", "active"}
|
|
)
|
|
if repair_attempts:
|
|
latest = repair_attempts[-1]
|
|
state["lastRepairAttemptId"] = latest.repair_id
|
|
state["lastRepairTaskId"] = latest.task_id
|
|
state["lastRepairAt"] = latest.created_at
|
|
return state
|
|
|
|
|
|
def summarize_active_repairs(state: Dict[str, object]) -> List[Dict[str, str]]:
|
|
summary: List[Dict[str, str]] = []
|
|
for item in list(state.get("repairAttempts", [])):
|
|
if not isinstance(item, dict):
|
|
continue
|
|
status = str(item.get("status", "")).strip()
|
|
if status not in {"queued", "active"}:
|
|
continue
|
|
summary.append(
|
|
{
|
|
"repairId": str(item.get("repairId", "")).strip(),
|
|
"taskId": str(item.get("taskId", "")).strip(),
|
|
"status": status,
|
|
"reason": str(item.get("reason", "")).strip(),
|
|
"repairBriefPath": str(item.get("repairBriefPath", "")).strip(),
|
|
}
|
|
)
|
|
return summary
|
|
|
|
|
|
def write_repair_queue(project_root: Path, state: Dict[str, object]) -> Path:
|
|
queue_path = _repair_queue_path(project_root)
|
|
active_items = summarize_active_repairs(state)
|
|
lines = [
|
|
"# Air Engine Auto Repair Queue",
|
|
"",
|
|
"This file lists repair attempts that were automatically queued from blocked or failed worker results.",
|
|
"",
|
|
]
|
|
if active_items:
|
|
for item in active_items:
|
|
lines.append(f"- Repair: `{item['repairId']}`")
|
|
lines.append(f" Task: `{item['taskId']}`")
|
|
lines.append(f" Status: `{item['status']}`")
|
|
lines.append(f" Reason: {item['reason'] or 'n/a'}")
|
|
lines.append(f" Brief: `{item['repairBriefPath']}`")
|
|
else:
|
|
lines.append("- No active auto repair attempts.")
|
|
|
|
queue_path.write_text("\n".join(lines) + "\n", encoding="utf-8")
|
|
return queue_path
|