|
|
@@ -0,0 +1,1153 @@
|
|
|
+"""Application service and sole writer of the explicit task ledger."""
|
|
|
+
|
|
|
+from __future__ import annotations
|
|
|
+
|
|
|
+import asyncio
|
|
|
+from dataclasses import asdict
|
|
|
+from typing import Any, Callable, Dict, Iterable, List, Optional, Sequence, Tuple
|
|
|
+from uuid import uuid4
|
|
|
+
|
|
|
+from agent.trace.goal_models import GoalTree
|
|
|
+
|
|
|
+from .config import OrchestrationConfig
|
|
|
+from .models import (
|
|
|
+ AcceptanceCriterion,
|
|
|
+ AgentRole,
|
|
|
+ ArtifactRef,
|
|
|
+ AttemptStatus,
|
|
|
+ AttemptSubmission,
|
|
|
+ CriterionResult,
|
|
|
+ DecisionAction,
|
|
|
+ PlannerDecision,
|
|
|
+ TaskAttempt,
|
|
|
+ TaskCycleResult,
|
|
|
+ TaskLedger,
|
|
|
+ TaskRecord,
|
|
|
+ TaskSpec,
|
|
|
+ TaskStatus,
|
|
|
+ ValidationReport,
|
|
|
+ ValidationRunStatus,
|
|
|
+ ValidationVerdict,
|
|
|
+ new_id,
|
|
|
+ utc_now,
|
|
|
+)
|
|
|
+from .protocols import (
|
|
|
+ AgentExecutor,
|
|
|
+ ArtifactStore,
|
|
|
+ EventSink,
|
|
|
+ TaskStore,
|
|
|
+ ValidatorRunResult,
|
|
|
+ WorkerRunResult,
|
|
|
+)
|
|
|
+from .state_machine import transition
|
|
|
+from .store import TaskStoreNotFound
|
|
|
+
|
|
|
+
|
|
|
+TERMINAL_TASK_STATUSES = {
|
|
|
+ TaskStatus.COMPLETED,
|
|
|
+ TaskStatus.CANCELLED,
|
|
|
+ TaskStatus.SUPERSEDED,
|
|
|
+}
|
|
|
+
|
|
|
+
|
|
|
+class OrchestrationError(RuntimeError):
|
|
|
+ pass
|
|
|
+
|
|
|
+
|
|
|
+class TaskConflict(OrchestrationError):
|
|
|
+ pass
|
|
|
+
|
|
|
+
|
|
|
+class TaskCoordinator:
|
|
|
+ """Coordinates Task -> Attempt -> Validation -> Planner Decision.
|
|
|
+
|
|
|
+ Agent executors and stores are ports. This class owns all state transition
|
|
|
+ rules and is the only component allowed to commit TaskLedger changes.
|
|
|
+ """
|
|
|
+
|
|
|
+ _locks: Dict[str, asyncio.Lock] = {}
|
|
|
+
|
|
|
+ def __init__(
|
|
|
+ self,
|
|
|
+ task_store: TaskStore,
|
|
|
+ artifact_store: ArtifactStore,
|
|
|
+ trace_store: Any,
|
|
|
+ config: Optional[OrchestrationConfig] = None,
|
|
|
+ event_sink: Optional[EventSink] = None,
|
|
|
+ executor: Optional[AgentExecutor] = None,
|
|
|
+ ) -> None:
|
|
|
+ self.task_store = task_store
|
|
|
+ self.artifact_store = artifact_store
|
|
|
+ self.trace_store = trace_store
|
|
|
+ self.config = config or OrchestrationConfig()
|
|
|
+ self.event_sink = event_sink
|
|
|
+ self.executor = executor
|
|
|
+
|
|
|
+ def set_executor(self, executor: AgentExecutor) -> None:
|
|
|
+ self.executor = executor
|
|
|
+
|
|
|
+ def _lock(self, root_trace_id: str) -> asyncio.Lock:
|
|
|
+ return self._locks.setdefault(root_trace_id, asyncio.Lock())
|
|
|
+
|
|
|
+ def _artifact_store(self, root_trace_id: str) -> ArtifactStore:
|
|
|
+ binder = getattr(self.artifact_store, "for_root", None)
|
|
|
+ return binder(root_trace_id) if binder else self.artifact_store
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _idem(root_trace_id: str, key: Optional[str]) -> Optional[str]:
|
|
|
+ return f"{root_trace_id}:{key}" if key else None
|
|
|
+
|
|
|
+ async def ensure_ledger(self, root_trace_id: str, mission: str) -> TaskLedger:
|
|
|
+ async with self._lock(root_trace_id):
|
|
|
+ try:
|
|
|
+ return await self.task_store.load(root_trace_id)
|
|
|
+ except TaskStoreNotFound:
|
|
|
+ ledger = TaskLedger(root_trace_id=root_trace_id, mission=mission)
|
|
|
+ committed = await self.task_store.commit(ledger, expected_revision=-1)
|
|
|
+ await self._emit(root_trace_id, "ledger_created", {"mission": mission})
|
|
|
+ return committed.ledger
|
|
|
+
|
|
|
+ async def _mutate(
|
|
|
+ self,
|
|
|
+ root_trace_id: str,
|
|
|
+ event_type: str,
|
|
|
+ mutator: Callable[[TaskLedger], Dict[str, Any]],
|
|
|
+ idempotency_key: Optional[str] = None,
|
|
|
+ ) -> Dict[str, Any]:
|
|
|
+ async with self._lock(root_trace_id):
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ idem = self._idem(root_trace_id, idempotency_key)
|
|
|
+ if idem and idem in ledger.idempotency_results:
|
|
|
+ return dict(ledger.idempotency_results[idem])
|
|
|
+ result = mutator(ledger)
|
|
|
+ if idem:
|
|
|
+ ledger.idempotency_results[idem] = _plain(result)
|
|
|
+ await self.task_store.commit(
|
|
|
+ ledger,
|
|
|
+ expected_revision=ledger.revision,
|
|
|
+ idempotency_key=idem,
|
|
|
+ )
|
|
|
+ await self._emit(root_trace_id, event_type, _plain(result))
|
|
|
+ return result
|
|
|
+
|
|
|
+ async def _emit(self, root_trace_id: str, event_type: str, payload: Dict[str, Any]) -> None:
|
|
|
+ if self.event_sink:
|
|
|
+ await self.event_sink.emit(root_trace_id, event_type, payload)
|
|
|
+
|
|
|
+ async def create_tasks(
|
|
|
+ self,
|
|
|
+ root_trace_id: str,
|
|
|
+ drafts: Sequence[Dict[str, Any]],
|
|
|
+ parent_task_id: Optional[str] = None,
|
|
|
+ placement: Optional[Dict[str, Any]] = None,
|
|
|
+ idempotency_key: Optional[str] = None,
|
|
|
+ ) -> Dict[str, Any]:
|
|
|
+ if not drafts:
|
|
|
+ raise ValueError("At least one task draft is required")
|
|
|
+ placement = dict(placement or {})
|
|
|
+
|
|
|
+ def mutate(ledger: TaskLedger) -> Dict[str, Any]:
|
|
|
+ after_task_id = placement.get("after_task_id")
|
|
|
+ if after_task_id:
|
|
|
+ target = _task(ledger, after_task_id)
|
|
|
+ if parent_task_id is not None and target.parent_task_id != parent_task_id:
|
|
|
+ raise ValueError("after_task_id must be a sibling under parent_task_id")
|
|
|
+ effective_parent_id = target.parent_task_id
|
|
|
+ else:
|
|
|
+ effective_parent_id = parent_task_id
|
|
|
+ if effective_parent_id and effective_parent_id not in ledger.tasks:
|
|
|
+ raise ValueError(f"Parent task not found: {effective_parent_id}")
|
|
|
+ siblings = sorted(
|
|
|
+ (t for t in ledger.tasks.values() if t.parent_task_id == effective_parent_id),
|
|
|
+ key=lambda item: _path_key(item.display_path),
|
|
|
+ )
|
|
|
+ if after_task_id:
|
|
|
+ target_index = next(i for i, sibling in enumerate(siblings) if sibling.task_id == after_task_id)
|
|
|
+ start = target_index + 2
|
|
|
+ for index, sibling in enumerate(siblings[target_index + 1:], start=start + len(drafts)):
|
|
|
+ sibling.display_path = self._display_path(ledger, effective_parent_id, index)
|
|
|
+ else:
|
|
|
+ start = len(siblings) + 1
|
|
|
+ parent_path = ledger.tasks[effective_parent_id].display_path if effective_parent_id else ""
|
|
|
+ created: List[str] = []
|
|
|
+ for offset, draft in enumerate(drafts):
|
|
|
+ objective = str(draft.get("objective", "")).strip()
|
|
|
+ if not objective:
|
|
|
+ raise ValueError("Task objective is required")
|
|
|
+ task_id = new_id()
|
|
|
+ criteria = [AcceptanceCriterion.from_dict(x) for x in draft.get("acceptance_criteria", [])]
|
|
|
+ spec = TaskSpec(
|
|
|
+ version=1,
|
|
|
+ objective=objective,
|
|
|
+ acceptance_criteria=criteria,
|
|
|
+ context_refs=list(draft.get("context_refs", [])),
|
|
|
+ )
|
|
|
+ index = start + offset
|
|
|
+ display_path = f"{parent_path}.{index}" if parent_path else str(index)
|
|
|
+ ledger.tasks[task_id] = TaskRecord(
|
|
|
+ task_id=task_id,
|
|
|
+ goal_id=None,
|
|
|
+ parent_task_id=effective_parent_id,
|
|
|
+ display_path=display_path,
|
|
|
+ specs=[spec],
|
|
|
+ )
|
|
|
+ if effective_parent_id:
|
|
|
+ children = ledger.tasks[effective_parent_id].child_task_ids
|
|
|
+ if after_task_id and after_task_id in children:
|
|
|
+ insert_at = children.index(after_task_id) + 1 + offset
|
|
|
+ children.insert(insert_at, task_id)
|
|
|
+ else:
|
|
|
+ children.append(task_id)
|
|
|
+ created.append(task_id)
|
|
|
+ if placement and placement.get("focus") and created:
|
|
|
+ ledger.focused_task_id = created[0]
|
|
|
+ return {
|
|
|
+ "task_ids": created,
|
|
|
+ "parent_task_id": effective_parent_id,
|
|
|
+ "after_task_id": after_task_id,
|
|
|
+ }
|
|
|
+
|
|
|
+ result = await self._mutate(root_trace_id, "tasks_created", mutate, idempotency_key)
|
|
|
+ after_task_id = result.get("after_task_id")
|
|
|
+ for task_id in result["task_ids"]:
|
|
|
+ await self._ensure_goal_projection(root_trace_id, task_id, after_task_id=after_task_id)
|
|
|
+ after_task_id = task_id
|
|
|
+ return await self._tasks_result(root_trace_id, result["task_ids"])
|
|
|
+
|
|
|
+ async def focus_task(self, root_trace_id: str, task_id: str) -> Dict[str, Any]:
|
|
|
+ def mutate(ledger: TaskLedger) -> Dict[str, Any]:
|
|
|
+ task = _task(ledger, task_id)
|
|
|
+ ledger.focused_task_id = task_id
|
|
|
+ return {"task_id": task_id, "status": task.status.value}
|
|
|
+
|
|
|
+ result = await self._mutate(root_trace_id, "task_focused", mutate)
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ task = ledger.tasks[task_id]
|
|
|
+ if self.trace_store and task.goal_id:
|
|
|
+ tree = await self.trace_store.get_goal_tree(root_trace_id)
|
|
|
+ if tree and tree.find(task.goal_id):
|
|
|
+ tree.focus(task.goal_id)
|
|
|
+ await self.trace_store.update_goal_tree(root_trace_id, tree)
|
|
|
+ return result
|
|
|
+
|
|
|
+ async def dispatch_tasks(
|
|
|
+ self,
|
|
|
+ root_trace_id: str,
|
|
|
+ task_ids: Sequence[str],
|
|
|
+ worker_presets: Optional[Sequence[str]] = None,
|
|
|
+ idempotency_key: Optional[str] = None,
|
|
|
+ ) -> List[TaskCycleResult]:
|
|
|
+ if not self.executor:
|
|
|
+ raise RuntimeError("TaskCoordinator has no AgentExecutor")
|
|
|
+ if not task_ids:
|
|
|
+ return []
|
|
|
+ if len(set(task_ids)) != len(task_ids):
|
|
|
+ raise TaskConflict("The same task cannot be dispatched twice in one call")
|
|
|
+ presets = list(worker_presets or [])
|
|
|
+ if presets and len(presets) not in (1, len(task_ids)):
|
|
|
+ raise ValueError("worker_presets must have length 1 or match task_ids")
|
|
|
+ if len(presets) == 1:
|
|
|
+ presets *= len(task_ids)
|
|
|
+ if not presets:
|
|
|
+ presets = [self.config.default_worker_preset] * len(task_ids)
|
|
|
+
|
|
|
+ full_idem = self._idem(root_trace_id, idempotency_key)
|
|
|
+ if full_idem:
|
|
|
+ existing = (await self.task_store.load(root_trace_id)).idempotency_results.get(full_idem)
|
|
|
+ if existing and "results" in existing:
|
|
|
+ return [TaskCycleResult.from_dict(x) for x in existing["results"]]
|
|
|
+
|
|
|
+ # Reserve every task before starting any worker so concurrent duplicate
|
|
|
+ # dispatches fail deterministically.
|
|
|
+ reservations: List[Tuple[str, str]] = []
|
|
|
+ for index, (task_id, preset) in enumerate(zip(task_ids, presets)):
|
|
|
+ key = f"{idempotency_key}:{index}" if idempotency_key else None
|
|
|
+ attempt = await self._create_attempt(root_trace_id, task_id, preset, key)
|
|
|
+ reservations.append((task_id, attempt["attempt_id"]))
|
|
|
+
|
|
|
+ semaphore = asyncio.Semaphore(self.config.max_parallel_tasks)
|
|
|
+
|
|
|
+ async def run_one(task_id: str, attempt_id: str) -> TaskCycleResult:
|
|
|
+ async with semaphore:
|
|
|
+ return await self._run_task_cycle(root_trace_id, task_id, attempt_id)
|
|
|
+
|
|
|
+ results = list(await asyncio.gather(*(run_one(t, a) for t, a in reservations)))
|
|
|
+ if idempotency_key:
|
|
|
+ def remember(ledger: TaskLedger) -> Dict[str, Any]:
|
|
|
+ return {"results": [x.to_dict() for x in results]}
|
|
|
+ await self._mutate(root_trace_id, "dispatch_completed", remember, idempotency_key)
|
|
|
+ return results
|
|
|
+
|
|
|
+ async def _create_attempt(
|
|
|
+ self,
|
|
|
+ root_trace_id: str,
|
|
|
+ task_id: str,
|
|
|
+ worker_preset: str,
|
|
|
+ idempotency_key: Optional[str],
|
|
|
+ ) -> Dict[str, Any]:
|
|
|
+ def mutate(ledger: TaskLedger) -> Dict[str, Any]:
|
|
|
+ task = _task(ledger, task_id)
|
|
|
+ if task.status not in (TaskStatus.PENDING, TaskStatus.NEEDS_REPLAN):
|
|
|
+ raise TaskConflict(f"Task {task_id} cannot dispatch from {task.status.value}")
|
|
|
+ active = [
|
|
|
+ ledger.attempts[x] for x in task.attempt_ids
|
|
|
+ if ledger.attempts[x].status == AttemptStatus.RUNNING
|
|
|
+ ]
|
|
|
+ if active:
|
|
|
+ raise TaskConflict(f"Task {task_id} already has a running attempt")
|
|
|
+
|
|
|
+ continue_trace_id = None
|
|
|
+ execution_mode = "new"
|
|
|
+ if task.decision_ids:
|
|
|
+ decision = ledger.decisions[task.decision_ids[-1]]
|
|
|
+ if decision.action == DecisionAction.REPAIR and task.attempt_ids:
|
|
|
+ previous = ledger.attempts[task.attempt_ids[-1]]
|
|
|
+ continue_trace_id = previous.worker_trace_id
|
|
|
+ execution_mode = "repair"
|
|
|
+
|
|
|
+ worker_trace_id = continue_trace_id or str(uuid4())
|
|
|
+ attempt = TaskAttempt(
|
|
|
+ attempt_id=new_id(),
|
|
|
+ task_id=task_id,
|
|
|
+ spec_version=task.current_spec_version,
|
|
|
+ worker_trace_id=worker_trace_id,
|
|
|
+ worker_preset=worker_preset,
|
|
|
+ execution_mode=execution_mode,
|
|
|
+ continue_from_trace_id=continue_trace_id,
|
|
|
+ )
|
|
|
+ transition(task.status, TaskStatus.RUNNING)
|
|
|
+ task.status = TaskStatus.RUNNING
|
|
|
+ task.attempt_ids.append(attempt.attempt_id)
|
|
|
+ task.updated_at = utc_now()
|
|
|
+ ledger.attempts[attempt.attempt_id] = attempt
|
|
|
+ return {
|
|
|
+ "task_id": task_id,
|
|
|
+ "attempt_id": attempt.attempt_id,
|
|
|
+ "worker_trace_id": worker_trace_id,
|
|
|
+ "execution_mode": execution_mode,
|
|
|
+ }
|
|
|
+
|
|
|
+ result = await self._mutate(root_trace_id, "attempt_created", mutate, idempotency_key)
|
|
|
+ await self.project_goal_state(root_trace_id, task_id)
|
|
|
+ return result
|
|
|
+
|
|
|
+ async def _run_task_cycle(
|
|
|
+ self,
|
|
|
+ root_trace_id: str,
|
|
|
+ task_id: str,
|
|
|
+ attempt_id: str,
|
|
|
+ ) -> TaskCycleResult:
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ task = _task(ledger, task_id)
|
|
|
+ attempt = ledger.attempts[attempt_id]
|
|
|
+ worker_context = {
|
|
|
+ "root_trace_id": root_trace_id,
|
|
|
+ "task_id": task_id,
|
|
|
+ "spec_version": task.current_spec_version,
|
|
|
+ "task_spec": _plain(asdict(task.current_spec)),
|
|
|
+ "attempt_id": attempt_id,
|
|
|
+ "worker_trace_id": attempt.worker_trace_id,
|
|
|
+ "worker_preset": attempt.worker_preset,
|
|
|
+ "continue_trace_id": attempt.continue_from_trace_id,
|
|
|
+ "prior_attempt_id": (
|
|
|
+ task.attempt_ids[-2]
|
|
|
+ if attempt.execution_mode == "repair" and len(task.attempt_ids) >= 2
|
|
|
+ else None
|
|
|
+ ),
|
|
|
+ "repair_feedback": self._repair_feedback(ledger, task) if attempt.execution_mode == "repair" else None,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ worker_result = await self.executor.run_worker(worker_context) # type: ignore[union-attr]
|
|
|
+ except Exception as exc:
|
|
|
+ worker_result = WorkerRunResult(
|
|
|
+ trace_id=attempt.worker_trace_id,
|
|
|
+ status="failed",
|
|
|
+ error=f"Worker executor error: {exc}",
|
|
|
+ )
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ attempt = ledger.attempts[attempt_id]
|
|
|
+ if attempt.status != AttemptStatus.SUBMITTED:
|
|
|
+ await self._mark_worker_failure(
|
|
|
+ root_trace_id,
|
|
|
+ task_id,
|
|
|
+ attempt_id,
|
|
|
+ worker_result.status,
|
|
|
+ worker_result.error or "Worker ended without submit_attempt",
|
|
|
+ )
|
|
|
+ return TaskCycleResult(
|
|
|
+ task_id=task_id,
|
|
|
+ task_status=TaskStatus.NEEDS_REPLAN,
|
|
|
+ attempt_id=attempt_id,
|
|
|
+ error=worker_result.error or "Worker ended without submit_attempt",
|
|
|
+ )
|
|
|
+
|
|
|
+ validation_id = await self._start_validation(root_trace_id, task_id, attempt_id)
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ task = ledger.tasks[task_id]
|
|
|
+ attempt = ledger.attempts[attempt_id]
|
|
|
+ validation = ledger.validations[validation_id]
|
|
|
+ snapshot = await self._artifact_store(root_trace_id).get(attempt.snapshot_id) # type: ignore[arg-type]
|
|
|
+ validator_context = {
|
|
|
+ "root_trace_id": root_trace_id,
|
|
|
+ "task_id": task_id,
|
|
|
+ "spec_version": task.current_spec_version,
|
|
|
+ "task_spec": _plain(asdict(task.current_spec)),
|
|
|
+ "attempt_id": attempt_id,
|
|
|
+ "snapshot_id": snapshot.snapshot_id,
|
|
|
+ "artifact_snapshot": _plain(asdict(snapshot)),
|
|
|
+ "validation_id": validation_id,
|
|
|
+ "validator_trace_id": validation.validator_trace_id,
|
|
|
+ "validator_preset": validation.validator_preset,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ validator_result = await self.executor.run_validator(validator_context) # type: ignore[union-attr]
|
|
|
+ except Exception as exc:
|
|
|
+ validator_result = ValidatorRunResult(
|
|
|
+ trace_id=validation.validator_trace_id,
|
|
|
+ status="error",
|
|
|
+ error=f"Validator executor error: {exc}",
|
|
|
+ )
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ validation = ledger.validations[validation_id]
|
|
|
+ if validation.status != ValidationRunStatus.COMPLETED:
|
|
|
+ await self._mark_validation_error(
|
|
|
+ root_trace_id,
|
|
|
+ task_id,
|
|
|
+ validation_id,
|
|
|
+ validator_result.status,
|
|
|
+ validator_result.error or "Validator ended without submit_validation",
|
|
|
+ )
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ validation = ledger.validations[validation_id]
|
|
|
+ return TaskCycleResult(
|
|
|
+ task_id=task_id,
|
|
|
+ task_status=ledger.tasks[task_id].status,
|
|
|
+ attempt_id=attempt_id,
|
|
|
+ validation_id=validation_id,
|
|
|
+ validation=validation,
|
|
|
+ error=validation.error,
|
|
|
+ )
|
|
|
+
|
|
|
+ async def submit_attempt(
|
|
|
+ self,
|
|
|
+ actor_context: Dict[str, Any],
|
|
|
+ submission: AttemptSubmission,
|
|
|
+ ) -> Dict[str, Any]:
|
|
|
+ _require_actor(actor_context, AgentRole.WORKER)
|
|
|
+ root_trace_id = _required(actor_context, "root_trace_id")
|
|
|
+ task_id = _required(actor_context, "task_id")
|
|
|
+ attempt_id = _required(actor_context, "attempt_id")
|
|
|
+ tool_call_id = _required(actor_context, "tool_call_id")
|
|
|
+
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ idem = self._idem(root_trace_id, tool_call_id)
|
|
|
+ if idem and idem in ledger.idempotency_results:
|
|
|
+ return dict(ledger.idempotency_results[idem])
|
|
|
+ task = _task(ledger, task_id)
|
|
|
+ attempt = ledger.attempts.get(attempt_id)
|
|
|
+ if not attempt or attempt.task_id != task_id:
|
|
|
+ raise OrchestrationError("Attempt does not belong to task")
|
|
|
+ if attempt.worker_trace_id != actor_context.get("trace_id"):
|
|
|
+ raise OrchestrationError("Worker trace does not own this attempt")
|
|
|
+ if attempt.spec_version != int(actor_context.get("spec_version")):
|
|
|
+ raise OrchestrationError("Worker submitted against the wrong TaskSpec version")
|
|
|
+ if attempt.status != AttemptStatus.RUNNING or task.status != TaskStatus.RUNNING:
|
|
|
+ raise TaskConflict("Attempt can only be submitted once while running")
|
|
|
+
|
|
|
+ snapshot = await self._artifact_store(root_trace_id).freeze(attempt_id, submission)
|
|
|
+
|
|
|
+ def mutate(current: TaskLedger) -> Dict[str, Any]:
|
|
|
+ current_task = _task(current, task_id)
|
|
|
+ current_attempt = current.attempts[attempt_id]
|
|
|
+ if current_attempt.status != AttemptStatus.RUNNING:
|
|
|
+ raise TaskConflict("Attempt was already submitted")
|
|
|
+ transition(current_task.status, TaskStatus.AWAITING_VALIDATION)
|
|
|
+ current_attempt.status = AttemptStatus.SUBMITTED
|
|
|
+ current_attempt.submission = submission
|
|
|
+ current_attempt.snapshot_id = snapshot.snapshot_id
|
|
|
+ current_attempt.updated_at = utc_now()
|
|
|
+ current_task.status = TaskStatus.AWAITING_VALIDATION
|
|
|
+ current_task.updated_at = utc_now()
|
|
|
+ return {
|
|
|
+ "task_id": task_id,
|
|
|
+ "attempt_id": attempt_id,
|
|
|
+ "snapshot_id": snapshot.snapshot_id,
|
|
|
+ "sha256": snapshot.sha256,
|
|
|
+ "status": current_task.status.value,
|
|
|
+ }
|
|
|
+
|
|
|
+ result = await self._mutate(root_trace_id, "attempt_submitted", mutate, tool_call_id)
|
|
|
+ await self.project_goal_state(root_trace_id, task_id)
|
|
|
+ return result
|
|
|
+
|
|
|
+ async def _start_validation(self, root_trace_id: str, task_id: str, attempt_id: str) -> str:
|
|
|
+ def mutate(ledger: TaskLedger) -> Dict[str, Any]:
|
|
|
+ task = _task(ledger, task_id)
|
|
|
+ attempt = ledger.attempts[attempt_id]
|
|
|
+ if task.status != TaskStatus.AWAITING_VALIDATION or attempt.status != AttemptStatus.SUBMITTED:
|
|
|
+ raise TaskConflict("Validation requires a submitted attempt")
|
|
|
+ validation = ValidationReport(
|
|
|
+ validation_id=new_id(),
|
|
|
+ task_id=task_id,
|
|
|
+ attempt_id=attempt_id,
|
|
|
+ spec_version=attempt.spec_version,
|
|
|
+ snapshot_id=attempt.snapshot_id or "",
|
|
|
+ validator_trace_id=str(uuid4()),
|
|
|
+ validator_preset=self.config.default_validator_preset,
|
|
|
+ status=ValidationRunStatus.RUNNING,
|
|
|
+ )
|
|
|
+ transition(task.status, TaskStatus.VALIDATING)
|
|
|
+ task.status = TaskStatus.VALIDATING
|
|
|
+ task.validation_ids.append(validation.validation_id)
|
|
|
+ task.updated_at = utc_now()
|
|
|
+ ledger.validations[validation.validation_id] = validation
|
|
|
+ return {"validation_id": validation.validation_id, "task_id": task_id}
|
|
|
+
|
|
|
+ result = await self._mutate(root_trace_id, "validation_started", mutate)
|
|
|
+ await self.project_goal_state(root_trace_id, task_id)
|
|
|
+ return result["validation_id"]
|
|
|
+
|
|
|
+ async def submit_validation(
|
|
|
+ self,
|
|
|
+ actor_context: Dict[str, Any],
|
|
|
+ verdict: ValidationVerdict,
|
|
|
+ criterion_results: Sequence[CriterionResult],
|
|
|
+ summary: str,
|
|
|
+ evidence_refs: Sequence[ArtifactRef],
|
|
|
+ unverified_claims: Sequence[str],
|
|
|
+ risks: Sequence[str],
|
|
|
+ recommendation: str,
|
|
|
+ ) -> Dict[str, Any]:
|
|
|
+ _require_actor(actor_context, AgentRole.VALIDATOR)
|
|
|
+ root_trace_id = _required(actor_context, "root_trace_id")
|
|
|
+ task_id = _required(actor_context, "task_id")
|
|
|
+ attempt_id = _required(actor_context, "attempt_id")
|
|
|
+ validation_id = _required(actor_context, "validation_id")
|
|
|
+ tool_call_id = _required(actor_context, "tool_call_id")
|
|
|
+ snapshot_id = _required(actor_context, "snapshot_id")
|
|
|
+
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ idem = self._idem(root_trace_id, tool_call_id)
|
|
|
+ if idem and idem in ledger.idempotency_results:
|
|
|
+ return dict(ledger.idempotency_results[idem])
|
|
|
+ task = _task(ledger, task_id)
|
|
|
+ validation = ledger.validations.get(validation_id)
|
|
|
+ if not validation or validation.task_id != task_id or validation.attempt_id != attempt_id:
|
|
|
+ raise OrchestrationError("Validation identity does not match task/attempt")
|
|
|
+ if validation.validator_trace_id != actor_context.get("trace_id"):
|
|
|
+ raise OrchestrationError("Validator trace does not own this validation")
|
|
|
+ if validation.snapshot_id != snapshot_id:
|
|
|
+ raise OrchestrationError("Validator must evaluate the fixed snapshot")
|
|
|
+ if validation.status != ValidationRunStatus.RUNNING or task.status != TaskStatus.VALIDATING:
|
|
|
+ raise TaskConflict("Validation can only be submitted once while running")
|
|
|
+ self._validate_report(task, verdict, criterion_results)
|
|
|
+
|
|
|
+ def mutate(current: TaskLedger) -> Dict[str, Any]:
|
|
|
+ current_task = _task(current, task_id)
|
|
|
+ report = current.validations[validation_id]
|
|
|
+ if report.status != ValidationRunStatus.RUNNING:
|
|
|
+ raise TaskConflict("Validation was already submitted")
|
|
|
+ report.status = ValidationRunStatus.COMPLETED
|
|
|
+ report.verdict = verdict
|
|
|
+ report.criterion_results = list(criterion_results)
|
|
|
+ report.summary = summary
|
|
|
+ report.evidence_refs = list(evidence_refs)
|
|
|
+ report.unverified_claims = list(unverified_claims)
|
|
|
+ report.risks = list(risks)
|
|
|
+ report.recommendation = recommendation
|
|
|
+ report.updated_at = utc_now()
|
|
|
+ transition(current_task.status, TaskStatus.AWAITING_DECISION)
|
|
|
+ current_task.status = TaskStatus.AWAITING_DECISION
|
|
|
+ current_task.updated_at = utc_now()
|
|
|
+ return {
|
|
|
+ "task_id": task_id,
|
|
|
+ "validation_id": validation_id,
|
|
|
+ "verdict": verdict.value,
|
|
|
+ "status": current_task.status.value,
|
|
|
+ }
|
|
|
+
|
|
|
+ result = await self._mutate(root_trace_id, "validation_submitted", mutate, tool_call_id)
|
|
|
+ await self.project_goal_state(root_trace_id, task_id)
|
|
|
+ return result
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _validate_report(
|
|
|
+ task: TaskRecord,
|
|
|
+ verdict: ValidationVerdict,
|
|
|
+ criterion_results: Sequence[CriterionResult],
|
|
|
+ ) -> None:
|
|
|
+ expected = {c.criterion_id: c for c in task.current_spec.acceptance_criteria}
|
|
|
+ actual = {c.criterion_id: c for c in criterion_results}
|
|
|
+ missing = set(expected) - set(actual)
|
|
|
+ if missing:
|
|
|
+ raise ValueError(f"Validation report is missing criteria: {sorted(missing)}")
|
|
|
+ if verdict == ValidationVerdict.PASSED:
|
|
|
+ non_passed = [
|
|
|
+ cid for cid, criterion in expected.items()
|
|
|
+ if criterion.hard and actual[cid].verdict != ValidationVerdict.PASSED
|
|
|
+ ]
|
|
|
+ if non_passed:
|
|
|
+ raise ValueError(f"Overall passed conflicts with hard criteria: {non_passed}")
|
|
|
+
|
|
|
+ async def decide_task(
|
|
|
+ self,
|
|
|
+ root_trace_id: str,
|
|
|
+ task_id: str,
|
|
|
+ validation_id: Optional[str],
|
|
|
+ action: DecisionAction,
|
|
|
+ payload: Optional[Dict[str, Any]],
|
|
|
+ idempotency_key: Optional[str],
|
|
|
+ ) -> Dict[str, Any]:
|
|
|
+ payload = dict(payload or {})
|
|
|
+
|
|
|
+ def mutate(ledger: TaskLedger) -> Dict[str, Any]:
|
|
|
+ task = _task(ledger, task_id)
|
|
|
+ before = task.status
|
|
|
+ validation = ledger.validations.get(validation_id) if validation_id else None
|
|
|
+ attempt_id = validation.attempt_id if validation else (task.attempt_ids[-1] if task.attempt_ids else None)
|
|
|
+ reason = str(payload.get("reason", "")).strip()
|
|
|
+ if not reason and action not in (DecisionAction.UNBLOCK,):
|
|
|
+ raise ValueError("Planner decision reason is required")
|
|
|
+
|
|
|
+ if action == DecisionAction.ACCEPT:
|
|
|
+ self._guard_accept(task, validation, ledger)
|
|
|
+ target = TaskStatus.COMPLETED
|
|
|
+ elif action == DecisionAction.REPAIR:
|
|
|
+ self._guard_replan_decision(task, validation, ledger)
|
|
|
+ count_key = str(task.current_spec_version)
|
|
|
+ used = task.repair_count_by_version.get(count_key, 0)
|
|
|
+ if used >= self.config.max_repair_continuations:
|
|
|
+ raise TaskConflict("Repair continuation limit reached; retry, revise, split, or block")
|
|
|
+ task.repair_count_by_version[count_key] = used + 1
|
|
|
+ target = TaskStatus.PENDING
|
|
|
+ elif action == DecisionAction.RETRY:
|
|
|
+ if task.status not in (TaskStatus.AWAITING_DECISION, TaskStatus.NEEDS_REPLAN):
|
|
|
+ raise TaskConflict("Retry requires awaiting_decision or needs_replan")
|
|
|
+ target = TaskStatus.PENDING
|
|
|
+ elif action == DecisionAction.REVISE:
|
|
|
+ if task.status not in (TaskStatus.AWAITING_DECISION, TaskStatus.NEEDS_REPLAN):
|
|
|
+ raise TaskConflict("Revise requires awaiting_decision or needs_replan")
|
|
|
+ objective = str(payload.get("objective") or task.current_spec.objective).strip()
|
|
|
+ criteria_data = payload.get("acceptance_criteria")
|
|
|
+ criteria = (
|
|
|
+ [AcceptanceCriterion.from_dict(x) for x in criteria_data]
|
|
|
+ if criteria_data is not None else list(task.current_spec.acceptance_criteria)
|
|
|
+ )
|
|
|
+ version = task.current_spec_version + 1
|
|
|
+ task.specs.append(TaskSpec(
|
|
|
+ version=version,
|
|
|
+ objective=objective,
|
|
|
+ acceptance_criteria=criteria,
|
|
|
+ context_refs=list(payload.get("context_refs", task.current_spec.context_refs)),
|
|
|
+ ))
|
|
|
+ task.current_spec_version = version
|
|
|
+ target = TaskStatus.PENDING
|
|
|
+ elif action == DecisionAction.SPLIT:
|
|
|
+ if task.status not in (TaskStatus.AWAITING_DECISION, TaskStatus.NEEDS_REPLAN):
|
|
|
+ raise TaskConflict("Split requires awaiting_decision or needs_replan")
|
|
|
+ drafts = payload.get("tasks") or []
|
|
|
+ if not drafts:
|
|
|
+ raise ValueError("Split requires payload.tasks")
|
|
|
+ child_ids = self._create_child_records(ledger, task, drafts)
|
|
|
+ payload["child_task_ids"] = child_ids
|
|
|
+ target = TaskStatus.WAITING_CHILDREN
|
|
|
+ elif action == DecisionAction.BLOCK:
|
|
|
+ if task.status in TERMINAL_TASK_STATUSES:
|
|
|
+ raise TaskConflict("Terminal task cannot be blocked")
|
|
|
+ task.blocked_reason = reason
|
|
|
+ target = TaskStatus.BLOCKED
|
|
|
+ elif action == DecisionAction.UNBLOCK:
|
|
|
+ if task.status != TaskStatus.BLOCKED:
|
|
|
+ raise TaskConflict("Only a blocked task can be unblocked")
|
|
|
+ task.blocked_reason = None
|
|
|
+ target = TaskStatus.NEEDS_REPLAN
|
|
|
+ elif action == DecisionAction.CANCEL:
|
|
|
+ if task.status in TERMINAL_TASK_STATUSES:
|
|
|
+ raise TaskConflict("Task is already terminal")
|
|
|
+ target = TaskStatus.CANCELLED
|
|
|
+ elif action == DecisionAction.SUPERSEDE:
|
|
|
+ if task.status in TERMINAL_TASK_STATUSES:
|
|
|
+ raise TaskConflict("Task is already terminal")
|
|
|
+ replacement = payload.get("replacement") or {}
|
|
|
+ replacement_ids = self._create_sibling_records(ledger, task, [replacement])
|
|
|
+ task.superseded_by = replacement_ids[0]
|
|
|
+ payload["replacement_task_id"] = replacement_ids[0]
|
|
|
+ target = TaskStatus.SUPERSEDED
|
|
|
+ elif action == DecisionAction.REVALIDATE:
|
|
|
+ raise ValueError("Use revalidate_attempt() / validate_attempt tool")
|
|
|
+ else:
|
|
|
+ raise ValueError(f"Unsupported decision action: {action.value}")
|
|
|
+
|
|
|
+ transition(before, target)
|
|
|
+ task.status = target
|
|
|
+ task.updated_at = utc_now()
|
|
|
+ decision = PlannerDecision(
|
|
|
+ decision_id=new_id(),
|
|
|
+ task_id=task_id,
|
|
|
+ action=action,
|
|
|
+ reason=reason,
|
|
|
+ from_status=before,
|
|
|
+ to_status=target,
|
|
|
+ attempt_id=attempt_id,
|
|
|
+ validation_id=validation_id,
|
|
|
+ payload=payload,
|
|
|
+ )
|
|
|
+ ledger.decisions[decision.decision_id] = decision
|
|
|
+ task.decision_ids.append(decision.decision_id)
|
|
|
+ self._update_parent_after_child(ledger, task)
|
|
|
+ return {
|
|
|
+ "task_id": task_id,
|
|
|
+ "decision_id": decision.decision_id,
|
|
|
+ "action": action.value,
|
|
|
+ "status": target.value,
|
|
|
+ "payload": payload,
|
|
|
+ }
|
|
|
+
|
|
|
+ result = await self._mutate(root_trace_id, "planner_decision", mutate, idempotency_key)
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ affected = [task_id]
|
|
|
+ affected.extend(result.get("payload", {}).get("child_task_ids", []))
|
|
|
+ replacement = result.get("payload", {}).get("replacement_task_id")
|
|
|
+ if replacement:
|
|
|
+ affected.append(replacement)
|
|
|
+ parent_id = ledger.tasks[task_id].parent_task_id
|
|
|
+ if parent_id:
|
|
|
+ affected.append(parent_id)
|
|
|
+ for affected_task in dict.fromkeys(affected):
|
|
|
+ await self._ensure_goal_projection(root_trace_id, affected_task)
|
|
|
+ await self.project_goal_state(root_trace_id, affected_task)
|
|
|
+ return result
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _guard_accept(
|
|
|
+ task: TaskRecord,
|
|
|
+ validation: Optional[ValidationReport],
|
|
|
+ ledger: TaskLedger,
|
|
|
+ ) -> None:
|
|
|
+ if task.status != TaskStatus.AWAITING_DECISION:
|
|
|
+ raise TaskConflict("Accept requires awaiting_decision")
|
|
|
+ if not validation or validation.status != ValidationRunStatus.COMPLETED:
|
|
|
+ raise TaskConflict("Accept requires a completed validation")
|
|
|
+ if validation.spec_version != task.current_spec_version:
|
|
|
+ raise TaskConflict("Validation belongs to an obsolete TaskSpec version")
|
|
|
+ if validation.verdict != ValidationVerdict.PASSED:
|
|
|
+ raise TaskConflict("Only a passed validation can be accepted")
|
|
|
+ if not task.attempt_ids or validation.attempt_id != task.attempt_ids[-1]:
|
|
|
+ raise TaskConflict("Validation does not belong to the current attempt")
|
|
|
+ attempt = ledger.attempts[validation.attempt_id]
|
|
|
+ if validation.snapshot_id != attempt.snapshot_id:
|
|
|
+ raise TaskConflict("Validation does not belong to the current snapshot")
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _guard_replan_decision(
|
|
|
+ task: TaskRecord,
|
|
|
+ validation: Optional[ValidationReport],
|
|
|
+ ledger: TaskLedger,
|
|
|
+ ) -> None:
|
|
|
+ if task.status != TaskStatus.AWAITING_DECISION:
|
|
|
+ raise TaskConflict("Repair requires awaiting_decision")
|
|
|
+ if not validation or validation.status != ValidationRunStatus.COMPLETED:
|
|
|
+ raise TaskConflict("Repair requires a completed validation")
|
|
|
+ if validation.spec_version != task.current_spec_version:
|
|
|
+ raise TaskConflict("Repair validation belongs to an obsolete TaskSpec version")
|
|
|
+ if not task.attempt_ids or validation.attempt_id != task.attempt_ids[-1]:
|
|
|
+ raise TaskConflict("Repair validation does not belong to the current attempt")
|
|
|
+ if validation.snapshot_id != ledger.attempts[validation.attempt_id].snapshot_id:
|
|
|
+ raise TaskConflict("Repair validation does not belong to the current snapshot")
|
|
|
+ if validation.verdict == ValidationVerdict.PASSED:
|
|
|
+ raise TaskConflict("Passed work should be accepted, not repaired")
|
|
|
+
|
|
|
+ async def revalidate_attempt(
|
|
|
+ self,
|
|
|
+ root_trace_id: str,
|
|
|
+ task_id: str,
|
|
|
+ attempt_id: str,
|
|
|
+ idempotency_key: Optional[str] = None,
|
|
|
+ ) -> TaskCycleResult:
|
|
|
+ if not self.executor:
|
|
|
+ raise RuntimeError("TaskCoordinator has no AgentExecutor")
|
|
|
+
|
|
|
+ def mutate(ledger: TaskLedger) -> Dict[str, Any]:
|
|
|
+ task = _task(ledger, task_id)
|
|
|
+ attempt = ledger.attempts.get(attempt_id)
|
|
|
+ if not attempt or attempt.task_id != task_id or not attempt.snapshot_id:
|
|
|
+ raise OrchestrationError("Attempt has no immutable snapshot")
|
|
|
+ prior = [ledger.validations[x] for x in task.validation_ids if ledger.validations[x].attempt_id == attempt_id]
|
|
|
+ if not prior or prior[-1].status not in {
|
|
|
+ ValidationRunStatus.ERROR, ValidationRunStatus.STOPPED, ValidationRunStatus.EXPIRED
|
|
|
+ }:
|
|
|
+ raise TaskConflict("Revalidation is only allowed after validator error/stopped/expired")
|
|
|
+ if task.status != TaskStatus.NEEDS_REPLAN:
|
|
|
+ raise TaskConflict("Task must be in needs_replan before revalidation")
|
|
|
+ report = ValidationReport(
|
|
|
+ validation_id=new_id(), task_id=task_id, attempt_id=attempt_id,
|
|
|
+ spec_version=attempt.spec_version, snapshot_id=attempt.snapshot_id,
|
|
|
+ validator_trace_id=str(uuid4()), validator_preset=self.config.default_validator_preset,
|
|
|
+ status=ValidationRunStatus.RUNNING,
|
|
|
+ )
|
|
|
+ transition(task.status, TaskStatus.VALIDATING)
|
|
|
+ task.status = TaskStatus.VALIDATING
|
|
|
+ task.validation_ids.append(report.validation_id)
|
|
|
+ ledger.validations[report.validation_id] = report
|
|
|
+ return {"validation_id": report.validation_id}
|
|
|
+
|
|
|
+ result = await self._mutate(root_trace_id, "validation_restarted", mutate, idempotency_key)
|
|
|
+ validation_id = result["validation_id"]
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ task = ledger.tasks[task_id]
|
|
|
+ attempt = ledger.attempts[attempt_id]
|
|
|
+ validation = ledger.validations[validation_id]
|
|
|
+ snapshot = await self._artifact_store(root_trace_id).get(attempt.snapshot_id) # type: ignore[arg-type]
|
|
|
+ validator_context = {
|
|
|
+ "root_trace_id": root_trace_id,
|
|
|
+ "task_id": task_id,
|
|
|
+ "spec_version": task.current_spec_version,
|
|
|
+ "task_spec": _plain(asdict(task.current_spec)),
|
|
|
+ "attempt_id": attempt_id,
|
|
|
+ "snapshot_id": snapshot.snapshot_id,
|
|
|
+ "artifact_snapshot": _plain(asdict(snapshot)),
|
|
|
+ "validation_id": validation_id,
|
|
|
+ "validator_trace_id": validation.validator_trace_id,
|
|
|
+ "validator_preset": validation.validator_preset,
|
|
|
+ }
|
|
|
+ try:
|
|
|
+ outcome = await self.executor.run_validator(validator_context)
|
|
|
+ except Exception as exc:
|
|
|
+ outcome = ValidatorRunResult(
|
|
|
+ trace_id=validation.validator_trace_id,
|
|
|
+ status="error",
|
|
|
+ error=f"Validator executor error: {exc}",
|
|
|
+ )
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ validation = ledger.validations[validation_id]
|
|
|
+ if validation.status != ValidationRunStatus.COMPLETED:
|
|
|
+ await self._mark_validation_error(
|
|
|
+ root_trace_id, task_id, validation_id, outcome.status,
|
|
|
+ outcome.error or "Validator ended without submit_validation",
|
|
|
+ )
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ validation = ledger.validations[validation_id]
|
|
|
+ return TaskCycleResult(
|
|
|
+ task_id=task_id,
|
|
|
+ task_status=ledger.tasks[task_id].status,
|
|
|
+ attempt_id=attempt_id,
|
|
|
+ validation_id=validation_id,
|
|
|
+ validation=validation,
|
|
|
+ error=validation.error,
|
|
|
+ )
|
|
|
+
|
|
|
+ async def validate_continue_from(
|
|
|
+ self,
|
|
|
+ root_trace_id: str,
|
|
|
+ task_id: str,
|
|
|
+ prior_attempt_id: str,
|
|
|
+ ) -> str:
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ task = _task(ledger, task_id)
|
|
|
+ attempt = ledger.attempts.get(prior_attempt_id)
|
|
|
+ if not attempt or attempt.task_id != task_id:
|
|
|
+ raise OrchestrationError("Prior attempt does not belong to task")
|
|
|
+ if attempt.spec_version != task.current_spec_version:
|
|
|
+ raise TaskConflict("Cannot continue a worker across TaskSpec revisions")
|
|
|
+ if not task.decision_ids or ledger.decisions[task.decision_ids[-1]].action != DecisionAction.REPAIR:
|
|
|
+ raise TaskConflict("Worker continuation requires the latest decision to be repair")
|
|
|
+ if task.repair_count_by_version.get(str(task.current_spec_version), 0) > self.config.max_repair_continuations:
|
|
|
+ raise TaskConflict("Repair continuation limit exceeded")
|
|
|
+ trace = await self.trace_store.get_trace(attempt.worker_trace_id) if self.trace_store else None
|
|
|
+ if not trace or trace.agent_role != AgentRole.WORKER.value:
|
|
|
+ raise TaskConflict("Prior worker trace is missing or has the wrong role")
|
|
|
+ if trace.status not in ("completed", "stopped"):
|
|
|
+ raise TaskConflict("Prior worker trace is not in a recoverable terminal state")
|
|
|
+ context = trace.context or {}
|
|
|
+ if context.get("root_trace_id") != root_trace_id or context.get("task_id") != task_id:
|
|
|
+ raise TaskConflict("Prior worker trace identity is invalid")
|
|
|
+ if int(context.get("spec_version", -1)) != task.current_spec_version:
|
|
|
+ raise TaskConflict("Prior worker trace TaskSpec version is invalid")
|
|
|
+ if self.trace_store:
|
|
|
+ messages = await self.trace_store.get_main_path_messages(
|
|
|
+ attempt.worker_trace_id, trace.head_sequence
|
|
|
+ )
|
|
|
+ if trace.head_sequence > 0 and not messages:
|
|
|
+ raise TaskConflict("Prior worker trace messages cannot be recovered")
|
|
|
+ return attempt.worker_trace_id
|
|
|
+
|
|
|
+ async def _mark_worker_failure(
|
|
|
+ self,
|
|
|
+ root_trace_id: str,
|
|
|
+ task_id: str,
|
|
|
+ attempt_id: str,
|
|
|
+ run_status: str,
|
|
|
+ error: str,
|
|
|
+ ) -> None:
|
|
|
+ def mutate(ledger: TaskLedger) -> Dict[str, Any]:
|
|
|
+ task = _task(ledger, task_id)
|
|
|
+ attempt = ledger.attempts[attempt_id]
|
|
|
+ status_map = {"stopped": AttemptStatus.STOPPED, "expired": AttemptStatus.EXPIRED}
|
|
|
+ attempt.status = status_map.get(run_status, AttemptStatus.FAILED)
|
|
|
+ attempt.error = error
|
|
|
+ attempt.updated_at = utc_now()
|
|
|
+ if task.status == TaskStatus.RUNNING:
|
|
|
+ transition(task.status, TaskStatus.NEEDS_REPLAN)
|
|
|
+ task.status = TaskStatus.NEEDS_REPLAN
|
|
|
+ return {"task_id": task_id, "attempt_id": attempt_id, "error": error}
|
|
|
+
|
|
|
+ await self._mutate(root_trace_id, "worker_failed", mutate)
|
|
|
+ await self.project_goal_state(root_trace_id, task_id)
|
|
|
+
|
|
|
+ async def _mark_validation_error(
|
|
|
+ self,
|
|
|
+ root_trace_id: str,
|
|
|
+ task_id: str,
|
|
|
+ validation_id: str,
|
|
|
+ run_status: str,
|
|
|
+ error: str,
|
|
|
+ ) -> None:
|
|
|
+ def mutate(ledger: TaskLedger) -> Dict[str, Any]:
|
|
|
+ task = _task(ledger, task_id)
|
|
|
+ report = ledger.validations[validation_id]
|
|
|
+ status_map = {
|
|
|
+ "stopped": ValidationRunStatus.STOPPED,
|
|
|
+ "expired": ValidationRunStatus.EXPIRED,
|
|
|
+ }
|
|
|
+ report.status = status_map.get(run_status, ValidationRunStatus.ERROR)
|
|
|
+ report.error = error
|
|
|
+ report.updated_at = utc_now()
|
|
|
+ if task.status == TaskStatus.VALIDATING:
|
|
|
+ transition(task.status, TaskStatus.NEEDS_REPLAN)
|
|
|
+ task.status = TaskStatus.NEEDS_REPLAN
|
|
|
+ return {"task_id": task_id, "validation_id": validation_id, "error": error}
|
|
|
+
|
|
|
+ await self._mutate(root_trace_id, "validation_error", mutate)
|
|
|
+ await self.project_goal_state(root_trace_id, task_id)
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _repair_feedback(ledger: TaskLedger, task: TaskRecord) -> Optional[Dict[str, Any]]:
|
|
|
+ if not task.validation_ids:
|
|
|
+ return None
|
|
|
+ report = ledger.validations[task.validation_ids[-1]]
|
|
|
+ return {
|
|
|
+ "verdict": report.verdict.value if report.verdict else None,
|
|
|
+ "summary": report.summary,
|
|
|
+ "criterion_results": [_plain(asdict(x)) for x in report.criterion_results],
|
|
|
+ "risks": report.risks,
|
|
|
+ "recommendation": report.recommendation,
|
|
|
+ }
|
|
|
+
|
|
|
+ def _create_child_records(
|
|
|
+ self,
|
|
|
+ ledger: TaskLedger,
|
|
|
+ parent: TaskRecord,
|
|
|
+ drafts: Sequence[Dict[str, Any]],
|
|
|
+ ) -> List[str]:
|
|
|
+ return self._create_records(ledger, drafts, parent.task_id, parent.display_path)
|
|
|
+
|
|
|
+ def _create_sibling_records(
|
|
|
+ self,
|
|
|
+ ledger: TaskLedger,
|
|
|
+ task: TaskRecord,
|
|
|
+ drafts: Sequence[Dict[str, Any]],
|
|
|
+ ) -> List[str]:
|
|
|
+ parent_path = ledger.tasks[task.parent_task_id].display_path if task.parent_task_id else ""
|
|
|
+ return self._create_records(ledger, drafts, task.parent_task_id, parent_path)
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _create_records(
|
|
|
+ ledger: TaskLedger,
|
|
|
+ drafts: Sequence[Dict[str, Any]],
|
|
|
+ parent_task_id: Optional[str],
|
|
|
+ parent_path: str,
|
|
|
+ ) -> List[str]:
|
|
|
+ siblings = sorted(
|
|
|
+ (t for t in ledger.tasks.values() if t.parent_task_id == parent_task_id),
|
|
|
+ key=lambda item: _path_key(item.display_path),
|
|
|
+ )
|
|
|
+ created: List[str] = []
|
|
|
+ for offset, draft in enumerate(drafts, start=1):
|
|
|
+ objective = str(draft.get("objective", "")).strip()
|
|
|
+ if not objective:
|
|
|
+ raise ValueError("New task objective is required")
|
|
|
+ task_id = new_id()
|
|
|
+ index = len(siblings) + offset
|
|
|
+ display_path = f"{parent_path}.{index}" if parent_path else str(index)
|
|
|
+ spec = TaskSpec(
|
|
|
+ version=1,
|
|
|
+ objective=objective,
|
|
|
+ acceptance_criteria=[AcceptanceCriterion.from_dict(x) for x in draft.get("acceptance_criteria", [])],
|
|
|
+ context_refs=list(draft.get("context_refs", [])),
|
|
|
+ )
|
|
|
+ ledger.tasks[task_id] = TaskRecord(
|
|
|
+ task_id=task_id, goal_id=None, parent_task_id=parent_task_id,
|
|
|
+ display_path=display_path, specs=[spec],
|
|
|
+ )
|
|
|
+ if parent_task_id:
|
|
|
+ ledger.tasks[parent_task_id].child_task_ids.append(task_id)
|
|
|
+ created.append(task_id)
|
|
|
+ return created
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _update_parent_after_child(ledger: TaskLedger, child: TaskRecord) -> None:
|
|
|
+ if not child.parent_task_id or child.status not in TERMINAL_TASK_STATUSES:
|
|
|
+ return
|
|
|
+ parent = ledger.tasks[child.parent_task_id]
|
|
|
+ children = [ledger.tasks[x] for x in parent.child_task_ids]
|
|
|
+ if child.status in (TaskStatus.CANCELLED, TaskStatus.SUPERSEDED):
|
|
|
+ if parent.status == TaskStatus.WAITING_CHILDREN:
|
|
|
+ transition(parent.status, TaskStatus.NEEDS_REPLAN)
|
|
|
+ parent.status = TaskStatus.NEEDS_REPLAN
|
|
|
+ elif children and all(x.status == TaskStatus.COMPLETED for x in children):
|
|
|
+ if parent.status == TaskStatus.WAITING_CHILDREN:
|
|
|
+ transition(parent.status, TaskStatus.NEEDS_REPLAN)
|
|
|
+ parent.status = TaskStatus.NEEDS_REPLAN
|
|
|
+ elif any(x.status in {TaskStatus.RUNNING, TaskStatus.VALIDATING, TaskStatus.BLOCKED, TaskStatus.PENDING} for x in children):
|
|
|
+ parent.status = TaskStatus.WAITING_CHILDREN
|
|
|
+ parent.updated_at = utc_now()
|
|
|
+
|
|
|
+ async def project_goal_state(self, root_trace_id: str, task_id: str) -> None:
|
|
|
+ if not self.trace_store:
|
|
|
+ return
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ task = _task(ledger, task_id)
|
|
|
+ if not task.goal_id:
|
|
|
+ return
|
|
|
+ status_map = {
|
|
|
+ TaskStatus.PENDING: "pending",
|
|
|
+ TaskStatus.RUNNING: "in_progress",
|
|
|
+ TaskStatus.AWAITING_VALIDATION: "in_progress",
|
|
|
+ TaskStatus.VALIDATING: "in_progress",
|
|
|
+ TaskStatus.AWAITING_DECISION: "in_progress",
|
|
|
+ TaskStatus.NEEDS_REPLAN: "in_progress",
|
|
|
+ TaskStatus.WAITING_CHILDREN: "in_progress",
|
|
|
+ TaskStatus.BLOCKED: "in_progress",
|
|
|
+ TaskStatus.COMPLETED: "completed",
|
|
|
+ TaskStatus.CANCELLED: "abandoned",
|
|
|
+ TaskStatus.SUPERSEDED: "abandoned",
|
|
|
+ }
|
|
|
+ summary = None
|
|
|
+ if task.decision_ids:
|
|
|
+ summary = ledger.decisions[task.decision_ids[-1]].reason
|
|
|
+ await self.trace_store.update_goal(
|
|
|
+ root_trace_id,
|
|
|
+ task.goal_id,
|
|
|
+ cascade_completion=False,
|
|
|
+ status=status_map[task.status],
|
|
|
+ summary=summary,
|
|
|
+ )
|
|
|
+
|
|
|
+ async def _ensure_goal_projection(
|
|
|
+ self,
|
|
|
+ root_trace_id: str,
|
|
|
+ task_id: str,
|
|
|
+ after_task_id: Optional[str] = None,
|
|
|
+ ) -> None:
|
|
|
+ if not self.trace_store:
|
|
|
+ return
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ task = _task(ledger, task_id)
|
|
|
+ tree = await self.trace_store.get_goal_tree(root_trace_id)
|
|
|
+ if tree is None:
|
|
|
+ tree = GoalTree(mission=ledger.mission)
|
|
|
+ if task.goal_id and tree.find(task.goal_id):
|
|
|
+ return
|
|
|
+ parent_goal_id = None
|
|
|
+ if task.parent_task_id:
|
|
|
+ parent = ledger.tasks[task.parent_task_id]
|
|
|
+ if not parent.goal_id:
|
|
|
+ await self._ensure_goal_projection(root_trace_id, parent.task_id)
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ parent = ledger.tasks[task.parent_task_id]
|
|
|
+ tree = await self.trace_store.get_goal_tree(root_trace_id) or tree
|
|
|
+ parent_goal_id = parent.goal_id
|
|
|
+ after_goal_id = None
|
|
|
+ if after_task_id and after_task_id in ledger.tasks:
|
|
|
+ after_goal_id = ledger.tasks[after_task_id].goal_id
|
|
|
+ if after_goal_id and tree.find(after_goal_id):
|
|
|
+ goal = tree.add_goals_after(
|
|
|
+ after_goal_id,
|
|
|
+ descriptions=[task.current_spec.objective],
|
|
|
+ reasons=["TaskLedger compatibility projection"],
|
|
|
+ )[0]
|
|
|
+ else:
|
|
|
+ goal = tree.add_goals(
|
|
|
+ descriptions=[task.current_spec.objective],
|
|
|
+ reasons=["TaskLedger compatibility projection"],
|
|
|
+ parent_id=parent_goal_id,
|
|
|
+ )[0]
|
|
|
+ await self.trace_store.update_goal_tree(root_trace_id, tree)
|
|
|
+
|
|
|
+ def link(current: TaskLedger) -> Dict[str, Any]:
|
|
|
+ current_task = _task(current, task_id)
|
|
|
+ if current_task.goal_id is None:
|
|
|
+ current_task.goal_id = goal.id
|
|
|
+ return {"task_id": task_id, "goal_id": current_task.goal_id}
|
|
|
+
|
|
|
+ await self._mutate(root_trace_id, "goal_projection_linked", link)
|
|
|
+
|
|
|
+ async def reconcile_goal_tree(self, root_trace_id: str) -> Dict[str, Any]:
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ for task_id in ledger.tasks:
|
|
|
+ await self._ensure_goal_projection(root_trace_id, task_id)
|
|
|
+ await self.project_goal_state(root_trace_id, task_id)
|
|
|
+ return {"root_trace_id": root_trace_id, "reconciled_tasks": len(ledger.tasks)}
|
|
|
+
|
|
|
+ async def _tasks_result(self, root_trace_id: str, task_ids: Iterable[str]) -> Dict[str, Any]:
|
|
|
+ ledger = await self.task_store.load(root_trace_id)
|
|
|
+ return {
|
|
|
+ "tasks": [
|
|
|
+ {
|
|
|
+ "task_id": task_id,
|
|
|
+ "goal_id": ledger.tasks[task_id].goal_id,
|
|
|
+ "display_path": ledger.tasks[task_id].display_path,
|
|
|
+ "status": ledger.tasks[task_id].status.value,
|
|
|
+ "spec_version": ledger.tasks[task_id].current_spec_version,
|
|
|
+ "objective": ledger.tasks[task_id].current_spec.objective,
|
|
|
+ }
|
|
|
+ for task_id in task_ids
|
|
|
+ ]
|
|
|
+ }
|
|
|
+
|
|
|
+ @staticmethod
|
|
|
+ def _display_path(ledger: TaskLedger, parent_task_id: Optional[str], index: int) -> str:
|
|
|
+ if not parent_task_id:
|
|
|
+ return str(index)
|
|
|
+ return f"{ledger.tasks[parent_task_id].display_path}.{index}"
|
|
|
+
|
|
|
+
|
|
|
+def _task(ledger: TaskLedger, task_id: str) -> TaskRecord:
|
|
|
+ try:
|
|
|
+ return ledger.tasks[task_id]
|
|
|
+ except KeyError as exc:
|
|
|
+ raise ValueError(f"Task not found: {task_id}") from exc
|
|
|
+
|
|
|
+
|
|
|
+def _required(context: Dict[str, Any], key: str) -> str:
|
|
|
+ value = context.get(key)
|
|
|
+ if value is None or value == "":
|
|
|
+ raise OrchestrationError(f"Protected tool context is missing {key}")
|
|
|
+ return str(value)
|
|
|
+
|
|
|
+
|
|
|
+def _require_actor(context: Dict[str, Any], expected: AgentRole) -> None:
|
|
|
+ actual = context.get("role")
|
|
|
+ if actual != expected.value:
|
|
|
+ raise OrchestrationError(f"Expected actor role {expected.value}, got {actual!r}")
|
|
|
+
|
|
|
+
|
|
|
+def _plain(value: Any) -> Any:
|
|
|
+ from enum import Enum
|
|
|
+ if isinstance(value, Enum):
|
|
|
+ return value.value
|
|
|
+ if isinstance(value, dict):
|
|
|
+ return {str(k): _plain(v) for k, v in value.items()}
|
|
|
+ if isinstance(value, (list, tuple)):
|
|
|
+ return [_plain(v) for v in value]
|
|
|
+ return value
|
|
|
+
|
|
|
+
|
|
|
+def _path_key(display_path: str) -> Tuple[int, ...]:
|
|
|
+ try:
|
|
|
+ return tuple(int(part) for part in display_path.split("."))
|
|
|
+ except ValueError:
|
|
|
+ return (10**9,)
|
|
|
+
|
|
|
+
|
|
|
+__all__ = ["TaskCoordinator", "OrchestrationError", "TaskConflict"]
|
SamLee