544 lines
20 KiB
Python
544 lines
20 KiB
Python
"""Portable multi-provider execution plans and aggregate release gates.
|
|
|
|
A plan is a DAG of executor-neutral stages. Only root stages are declared at
|
|
plan creation. A dependent stage is declared atomically after every dependency
|
|
passes its Truth + notification gate, so its start deadline cannot expire while
|
|
it is legitimately waiting for upstream work.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import sqlite3
|
|
import threading
|
|
from dataclasses import asdict, dataclass, field
|
|
from pathlib import Path
|
|
from typing import Any, Mapping, Sequence
|
|
|
|
from .executor_protocol import ExecutorProviderRegistry, normalize_executor
|
|
from .external_task_control import ExternalTaskCoordinator
|
|
from .external_task_query import list_records
|
|
from .task_execution_registry import (
|
|
ExecutionTaskManifest,
|
|
ExecutionTruthRecord,
|
|
ExecutionTruthState,
|
|
required,
|
|
utc_now,
|
|
)
|
|
|
|
PLAN_SCHEMA_VERSION = "aios.execution-plan.v2"
|
|
WAITING_DEPENDENCY = "WAITING_DEPENDENCY"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ExecutionStageManifest:
|
|
stage_id: str
|
|
task_id: str
|
|
executor_type: str
|
|
role: str
|
|
steps: Sequence[str]
|
|
depends_on: Sequence[str] = ()
|
|
required_evidence: Sequence[str] = ()
|
|
notification_events: Sequence[str] = ("STARTED", "SUCCEEDED", "FAILED")
|
|
owner_notification_ref: str = "AIOS_OWNER_WECHAT_TARGET_REF"
|
|
start_deadline_seconds: int = 300
|
|
heartbeat_timeout_seconds: int = 180
|
|
required: bool = True
|
|
metadata: Mapping[str, Any] = field(default_factory=dict)
|
|
|
|
def __post_init__(self) -> None:
|
|
for name in ("stage_id", "task_id", "role", "owner_notification_ref"):
|
|
object.__setattr__(self, name, required(getattr(self, name), name))
|
|
object.__setattr__(self, "executor_type", normalize_executor(self.executor_type))
|
|
steps = tuple(required(item, "step") for item in self.steps)
|
|
if not steps:
|
|
raise ValueError("execution stage requires at least one step")
|
|
object.__setattr__(self, "steps", steps)
|
|
object.__setattr__(
|
|
self,
|
|
"depends_on",
|
|
tuple(sorted({required(item, "depends_on") for item in self.depends_on})),
|
|
)
|
|
object.__setattr__(
|
|
self,
|
|
"required_evidence",
|
|
tuple(sorted({required(item, "required_evidence") for item in self.required_evidence})),
|
|
)
|
|
object.__setattr__(
|
|
self,
|
|
"notification_events",
|
|
tuple(sorted({required(item, "notification_event") for item in self.notification_events})),
|
|
)
|
|
if self.start_deadline_seconds < 1 or self.heartbeat_timeout_seconds < 1:
|
|
raise ValueError("stage deadlines must be positive")
|
|
object.__setattr__(
|
|
self,
|
|
"metadata",
|
|
json.loads(json.dumps(dict(self.metadata), ensure_ascii=False, default=str)),
|
|
)
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
value = asdict(self)
|
|
value["steps"] = list(self.steps)
|
|
value["depends_on"] = list(self.depends_on)
|
|
value["required_evidence"] = list(self.required_evidence)
|
|
value["notification_events"] = list(self.notification_events)
|
|
return value
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: Mapping[str, Any]) -> "ExecutionStageManifest":
|
|
return cls(**dict(value))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ExecutionPlanManifest:
|
|
plan_id: str
|
|
source_branch: str
|
|
source_oid: str
|
|
target_version: str
|
|
stages: Sequence[ExecutionStageManifest]
|
|
release_gate_policy: str = "all_required_stages"
|
|
schema_version: str = PLAN_SCHEMA_VERSION
|
|
metadata: Mapping[str, Any] = field(default_factory=dict)
|
|
|
|
def __post_init__(self) -> None:
|
|
for name in ("plan_id", "source_branch", "source_oid", "target_version"):
|
|
object.__setattr__(self, name, required(getattr(self, name), name))
|
|
if self.schema_version != PLAN_SCHEMA_VERSION:
|
|
raise ValueError("unsupported execution plan schema")
|
|
if self.release_gate_policy != "all_required_stages":
|
|
raise ValueError("unsupported release gate policy")
|
|
stages = tuple(self.stages)
|
|
if not stages:
|
|
raise ValueError("execution plan requires stages")
|
|
stage_ids = [item.stage_id for item in stages]
|
|
task_ids = [item.task_id for item in stages]
|
|
if len(stage_ids) != len(set(stage_ids)):
|
|
raise ValueError("execution plan stage_id values must be unique")
|
|
if len(task_ids) != len(set(task_ids)):
|
|
raise ValueError("execution plan task_id values must be unique")
|
|
known = set(stage_ids)
|
|
for stage in stages:
|
|
unknown = sorted(set(stage.depends_on) - known)
|
|
if unknown:
|
|
raise ValueError(f"stage {stage.stage_id} has unknown dependencies: {unknown}")
|
|
if stage.stage_id in stage.depends_on:
|
|
raise ValueError("stage cannot depend on itself")
|
|
self._assert_acyclic(stages)
|
|
object.__setattr__(self, "stages", stages)
|
|
object.__setattr__(
|
|
self,
|
|
"metadata",
|
|
json.loads(json.dumps(dict(self.metadata), ensure_ascii=False, default=str)),
|
|
)
|
|
|
|
@staticmethod
|
|
def _assert_acyclic(stages: Sequence[ExecutionStageManifest]) -> None:
|
|
dependencies = {item.stage_id: set(item.depends_on) for item in stages}
|
|
resolved: set[str] = set()
|
|
while len(resolved) < len(stages):
|
|
ready = {
|
|
stage_id
|
|
for stage_id, deps in dependencies.items()
|
|
if stage_id not in resolved and deps.issubset(resolved)
|
|
}
|
|
if not ready:
|
|
raise ValueError("execution plan contains a dependency cycle")
|
|
resolved.update(ready)
|
|
|
|
def as_dict(self) -> dict[str, Any]:
|
|
return {
|
|
"schema_version": self.schema_version,
|
|
"plan_id": self.plan_id,
|
|
"source_branch": self.source_branch,
|
|
"source_oid": self.source_oid,
|
|
"target_version": self.target_version,
|
|
"release_gate_policy": self.release_gate_policy,
|
|
"stages": [item.as_dict() for item in self.stages],
|
|
"metadata": dict(self.metadata),
|
|
}
|
|
|
|
@classmethod
|
|
def from_dict(cls, value: Mapping[str, Any]) -> "ExecutionPlanManifest":
|
|
raw = dict(value)
|
|
raw["stages"] = tuple(
|
|
ExecutionStageManifest.from_dict(item) for item in raw.get("stages") or ()
|
|
)
|
|
return cls(**raw)
|
|
|
|
def stage(self, stage_id: str) -> ExecutionStageManifest:
|
|
normalized = required(stage_id, "stage_id")
|
|
for stage in self.stages:
|
|
if stage.stage_id == normalized:
|
|
return stage
|
|
raise KeyError(f"execution stage not found: {normalized}")
|
|
|
|
|
|
class SQLiteExecutionPlanStore:
|
|
"""Durable plan declarations; stage state remains in Truth Registry."""
|
|
|
|
def __init__(self, path: str | Path) -> None:
|
|
self.path = Path(path).resolve()
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self._lock = threading.RLock()
|
|
with self._connect() as connection:
|
|
connection.executescript(
|
|
"""
|
|
CREATE TABLE IF NOT EXISTS execution_plans (
|
|
plan_id TEXT PRIMARY KEY,
|
|
manifest_json TEXT NOT NULL,
|
|
created_at TEXT NOT NULL,
|
|
updated_at TEXT NOT NULL
|
|
);
|
|
"""
|
|
)
|
|
|
|
def _connect(self) -> sqlite3.Connection:
|
|
connection = sqlite3.connect(str(self.path), timeout=30.0)
|
|
connection.row_factory = sqlite3.Row
|
|
connection.execute("PRAGMA journal_mode=WAL")
|
|
connection.execute("PRAGMA synchronous=FULL")
|
|
return connection
|
|
|
|
@staticmethod
|
|
def _encode(plan: ExecutionPlanManifest) -> str:
|
|
return json.dumps(
|
|
plan.as_dict(),
|
|
ensure_ascii=False,
|
|
sort_keys=True,
|
|
separators=(",", ":"),
|
|
)
|
|
|
|
def save(
|
|
self,
|
|
plan: ExecutionPlanManifest,
|
|
*,
|
|
allow_source_oid_supersession: bool = False,
|
|
) -> None:
|
|
payload = self._encode(plan)
|
|
now = utc_now()
|
|
with self._lock, self._connect() as connection:
|
|
connection.execute("BEGIN IMMEDIATE")
|
|
existing = connection.execute(
|
|
"SELECT manifest_json FROM execution_plans WHERE plan_id = ?",
|
|
(plan.plan_id,),
|
|
).fetchone()
|
|
if existing is not None:
|
|
if existing["manifest_json"] != payload:
|
|
previous = json.loads(existing["manifest_json"])
|
|
replacement = plan.as_dict()
|
|
previous_oid = previous.pop("source_oid", None)
|
|
replacement_oid = replacement.pop("source_oid", None)
|
|
if (
|
|
not allow_source_oid_supersession
|
|
or previous != replacement
|
|
or previous_oid == replacement_oid
|
|
):
|
|
connection.rollback()
|
|
raise RuntimeError(
|
|
"plan_id already exists with another manifest"
|
|
)
|
|
connection.execute(
|
|
"UPDATE execution_plans SET manifest_json = ?, updated_at = ? WHERE plan_id = ?",
|
|
(payload, now, plan.plan_id),
|
|
)
|
|
connection.commit()
|
|
return
|
|
connection.rollback()
|
|
return
|
|
connection.execute(
|
|
"INSERT INTO execution_plans VALUES (?, ?, ?, ?)",
|
|
(plan.plan_id, payload, now, now),
|
|
)
|
|
connection.commit()
|
|
|
|
def get(self, plan_id: str) -> ExecutionPlanManifest:
|
|
with self._lock, self._connect() as connection:
|
|
row = connection.execute(
|
|
"SELECT manifest_json FROM execution_plans WHERE plan_id = ?",
|
|
(required(plan_id, "plan_id"),),
|
|
).fetchone()
|
|
if row is None:
|
|
raise KeyError("execution plan not found")
|
|
value = json.loads(row["manifest_json"])
|
|
if not isinstance(value, Mapping):
|
|
raise ValueError("stored execution plan is invalid")
|
|
return ExecutionPlanManifest.from_dict(value)
|
|
|
|
def list(self) -> list[ExecutionPlanManifest]:
|
|
with self._lock, self._connect() as connection:
|
|
rows = connection.execute(
|
|
"SELECT manifest_json FROM execution_plans ORDER BY created_at, plan_id"
|
|
).fetchall()
|
|
return [
|
|
ExecutionPlanManifest.from_dict(json.loads(row["manifest_json"]))
|
|
for row in rows
|
|
]
|
|
|
|
|
|
class ExecutionPlanCoordinator:
|
|
"""Declares dependency-ready stages and evaluates aggregate release gates."""
|
|
|
|
def __init__(
|
|
self,
|
|
*,
|
|
task_coordinator: ExternalTaskCoordinator,
|
|
plan_store: SQLiteExecutionPlanStore,
|
|
providers: ExecutorProviderRegistry,
|
|
) -> None:
|
|
self.task_coordinator = task_coordinator
|
|
self.plan_store = plan_store
|
|
self.providers = providers
|
|
|
|
@classmethod
|
|
def from_file(
|
|
cls,
|
|
path: str | Path,
|
|
*,
|
|
providers: ExecutorProviderRegistry,
|
|
) -> "ExecutionPlanCoordinator":
|
|
return cls(
|
|
task_coordinator=ExternalTaskCoordinator.from_file(path),
|
|
plan_store=SQLiteExecutionPlanStore(path),
|
|
providers=providers,
|
|
)
|
|
|
|
@staticmethod
|
|
def _task_manifest(
|
|
plan: ExecutionPlanManifest,
|
|
stage: ExecutionStageManifest,
|
|
) -> ExecutionTaskManifest:
|
|
return ExecutionTaskManifest(
|
|
task_id=stage.task_id,
|
|
executor=stage.executor_type,
|
|
source_branch=plan.source_branch,
|
|
source_oid=plan.source_oid,
|
|
target_version=plan.target_version,
|
|
steps=stage.steps,
|
|
start_deadline_seconds=stage.start_deadline_seconds,
|
|
heartbeat_timeout_seconds=stage.heartbeat_timeout_seconds,
|
|
required_evidence=stage.required_evidence,
|
|
notification_events=stage.notification_events,
|
|
owner_notification_ref=stage.owner_notification_ref,
|
|
manifest_version="aios.external-task.v2",
|
|
)
|
|
|
|
def _record(self, task_id: str) -> ExecutionTruthRecord | None:
|
|
try:
|
|
return self.task_coordinator.registry.get(task_id)
|
|
except KeyError:
|
|
return None
|
|
|
|
def _declare_stage(
|
|
self,
|
|
plan: ExecutionPlanManifest,
|
|
stage: ExecutionStageManifest,
|
|
) -> ExecutionTruthRecord:
|
|
self.providers.validate_manifest(self._task_manifest(plan, stage))
|
|
return self.task_coordinator.declare(self._task_manifest(plan, stage))
|
|
|
|
def declare(self, plan: ExecutionPlanManifest) -> Mapping[str, Any]:
|
|
for stage in plan.stages:
|
|
self.providers.get(stage.executor_type)
|
|
allow_source_oid_supersession = False
|
|
try:
|
|
previous = self.plan_store.get(plan.plan_id)
|
|
except KeyError:
|
|
previous = None
|
|
if previous is not None and previous.source_oid != plan.source_oid:
|
|
previous_value = previous.as_dict()
|
|
replacement_value = plan.as_dict()
|
|
previous_value.pop("source_oid")
|
|
replacement_value.pop("source_oid")
|
|
root_records = [
|
|
self._record(stage.task_id)
|
|
for stage in previous.stages
|
|
if not stage.depends_on
|
|
]
|
|
allow_source_oid_supersession = bool(
|
|
previous_value == replacement_value
|
|
and root_records
|
|
and all(
|
|
record is not None
|
|
and record.state == ExecutionTruthState.NOT_STARTED
|
|
for record in root_records
|
|
)
|
|
)
|
|
self.plan_store.save(
|
|
plan,
|
|
allow_source_oid_supersession=allow_source_oid_supersession,
|
|
)
|
|
for stage in plan.stages:
|
|
if not stage.depends_on:
|
|
self._declare_stage(plan, stage)
|
|
return self.snapshot(plan.plan_id)
|
|
|
|
def declare_file(self, path: str | Path) -> Mapping[str, Any]:
|
|
value = json.loads(Path(path).read_text(encoding="utf-8"))
|
|
if not isinstance(value, Mapping):
|
|
raise ValueError("execution plan file must contain a JSON object")
|
|
return self.declare(ExecutionPlanManifest.from_dict(value))
|
|
|
|
def dependency_gate(self, plan_id: str, stage_id: str) -> Mapping[str, Any]:
|
|
plan = self.plan_store.get(plan_id)
|
|
stage = plan.stage(stage_id)
|
|
dependencies: list[Mapping[str, Any]] = []
|
|
passed = True
|
|
for dependency_id in stage.depends_on:
|
|
dependency = plan.stage(dependency_id)
|
|
record = self._record(dependency.task_id)
|
|
if record is None:
|
|
gate: Mapping[str, Any] = {
|
|
"task_id": dependency.task_id,
|
|
"state": WAITING_DEPENDENCY,
|
|
"passed": False,
|
|
"release_blocking": True,
|
|
}
|
|
else:
|
|
gate = self.task_coordinator.gate(dependency.task_id)
|
|
dependencies.append(
|
|
{
|
|
"stage_id": dependency.stage_id,
|
|
"task_id": dependency.task_id,
|
|
"gate": gate,
|
|
}
|
|
)
|
|
if not gate["passed"]:
|
|
passed = False
|
|
return {
|
|
"plan_id": plan.plan_id,
|
|
"stage_id": stage.stage_id,
|
|
"passed": passed,
|
|
"release_blocking": not passed,
|
|
"dependencies": dependencies,
|
|
}
|
|
|
|
def activate_ready(self, plan_id: str) -> list[Mapping[str, Any]]:
|
|
plan = self.plan_store.get(plan_id)
|
|
activated: list[Mapping[str, Any]] = []
|
|
for stage in plan.stages:
|
|
if self._record(stage.task_id) is not None:
|
|
continue
|
|
dependency = self.dependency_gate(plan.plan_id, stage.stage_id)
|
|
if not dependency["passed"]:
|
|
continue
|
|
record = self._declare_stage(plan, stage)
|
|
activated.append(
|
|
{
|
|
"stage_id": stage.stage_id,
|
|
"task_id": record.task_id,
|
|
"executor_type": stage.executor_type,
|
|
"state": record.state.value,
|
|
}
|
|
)
|
|
return activated
|
|
|
|
def discover_ready(
|
|
self,
|
|
*,
|
|
plan_id: str,
|
|
executor_type: str,
|
|
) -> list[Mapping[str, Any]]:
|
|
plan = self.plan_store.get(plan_id)
|
|
self.activate_ready(plan.plan_id)
|
|
normalized = normalize_executor(executor_type)
|
|
records = {
|
|
item.task_id: item
|
|
for item in list_records(
|
|
self.task_coordinator.registry.path,
|
|
executor=normalized,
|
|
states=(
|
|
ExecutionTruthState.DECLARED.value,
|
|
ExecutionTruthState.REGISTERED.value,
|
|
),
|
|
)
|
|
}
|
|
ready: list[Mapping[str, Any]] = []
|
|
for stage in plan.stages:
|
|
if stage.executor_type != normalized or stage.task_id not in records:
|
|
continue
|
|
dependency = self.dependency_gate(plan.plan_id, stage.stage_id)
|
|
if dependency["passed"]:
|
|
ready.append(
|
|
{
|
|
"plan_id": plan.plan_id,
|
|
"stage": stage.as_dict(),
|
|
"record": records[stage.task_id].as_dict(),
|
|
}
|
|
)
|
|
return ready
|
|
|
|
def snapshot(self, plan_id: str) -> Mapping[str, Any]:
|
|
plan = self.plan_store.get(plan_id)
|
|
stages: list[Mapping[str, Any]] = []
|
|
for stage in plan.stages:
|
|
record = self._record(stage.task_id)
|
|
dependency = self.dependency_gate(plan.plan_id, stage.stage_id)
|
|
if record is None:
|
|
truth: Mapping[str, Any] = {
|
|
"task_id": stage.task_id,
|
|
"state": WAITING_DEPENDENCY,
|
|
"passed": False,
|
|
"release_blocking": True,
|
|
"executor_run_id": None,
|
|
"executor_id": None,
|
|
"current_step": 0,
|
|
"total_steps": len(stage.steps),
|
|
"evidence": {},
|
|
"error": {},
|
|
"notifications": [],
|
|
}
|
|
else:
|
|
truth = self.task_coordinator.status(stage.task_id)
|
|
stages.append(
|
|
{
|
|
"stage": stage.as_dict(),
|
|
"dependency_gate": dependency,
|
|
"truth": truth,
|
|
}
|
|
)
|
|
return {
|
|
"manifest": plan.as_dict(),
|
|
"stages": stages,
|
|
}
|
|
|
|
def gate(self, plan_id: str) -> Mapping[str, Any]:
|
|
plan = self.plan_store.get(plan_id)
|
|
self.activate_ready(plan.plan_id)
|
|
stages: list[Mapping[str, Any]] = []
|
|
required_passed = True
|
|
for stage in plan.stages:
|
|
dependency = self.dependency_gate(plan.plan_id, stage.stage_id)
|
|
record = self._record(stage.task_id)
|
|
if record is None:
|
|
truth_gate: Mapping[str, Any] = {
|
|
"task_id": stage.task_id,
|
|
"state": WAITING_DEPENDENCY,
|
|
"passed": False,
|
|
"release_blocking": True,
|
|
}
|
|
else:
|
|
truth_gate = self.task_coordinator.gate(stage.task_id)
|
|
stage_passed = bool(truth_gate["passed"] and dependency["passed"])
|
|
if stage.required and not stage_passed:
|
|
required_passed = False
|
|
stages.append(
|
|
{
|
|
"stage_id": stage.stage_id,
|
|
"task_id": stage.task_id,
|
|
"executor_type": stage.executor_type,
|
|
"role": stage.role,
|
|
"required": stage.required,
|
|
"passed": stage_passed,
|
|
"truth_gate": truth_gate,
|
|
"dependency_gate": dependency,
|
|
}
|
|
)
|
|
return {
|
|
"plan_id": plan.plan_id,
|
|
"target_version": plan.target_version,
|
|
"policy": plan.release_gate_policy,
|
|
"passed": required_passed,
|
|
"release_blocking": not required_passed,
|
|
"stages": stages,
|
|
}
|