diff --git a/active-runtime/app/aios_v2/execution_plan.py b/active-runtime/app/aios_v2/execution_plan.py index 0b7198a..c07c346 100644 --- a/active-runtime/app/aios_v2/execution_plan.py +++ b/active-runtime/app/aios_v2/execution_plan.py @@ -208,7 +208,12 @@ class SQLiteExecutionPlanStore: separators=(",", ":"), ) - def save(self, plan: ExecutionPlanManifest) -> None: + 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: @@ -219,8 +224,25 @@ class SQLiteExecutionPlanStore: ).fetchone() if existing is not None: if existing["manifest_json"] != payload: - connection.rollback() - raise RuntimeError("plan_id already exists with another manifest") + 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( @@ -317,7 +339,34 @@ class ExecutionPlanCoordinator: def declare(self, plan: ExecutionPlanManifest) -> Mapping[str, Any]: for stage in plan.stages: self.providers.get(stage.executor_type) - self.plan_store.save(plan) + 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) diff --git a/tests/web/test_portable_executor_release.py b/tests/web/test_portable_executor_release.py index b3e2778..2de3046 100644 --- a/tests/web/test_portable_executor_release.py +++ b/tests/web/test_portable_executor_release.py @@ -6,6 +6,7 @@ import sys import tempfile import time import unittest +from datetime import timedelta from pathlib import Path ROOT = Path(__file__).resolve().parents[2] @@ -26,7 +27,7 @@ from aios_v2.executor_protocol import ( default_provider_registry, ) from aios_v2.executor_worker import UnifiedExecutorWorker -from aios_v2.task_execution_registry import ExecutionTaskManifest +from aios_v2.task_execution_registry import ExecutionTaskManifest, parse_time from aios_v2.official_executor_providers import official_provider_factories from aios_web.console import PORTAL_HTML @@ -292,6 +293,48 @@ class PortableExecutorReleaseTests(unittest.TestCase): executor_run_id="retry-run-2", ) + def test_plan_source_oid_supersession_requires_not_started_root(self) -> None: + coordinator = self.coordinator() + plan = ExecutionPlanManifest( + plan_id="test-plan-supersession", + source_branch="patch/test", + source_oid="4" * 40, + target_version="VTEST", + stages=( + ExecutionStageManifest( + stage_id="root", + task_id="task-plan-supersession", + executor_type="openclaw", + role="candidate_deploy", + steps=("deploy",), + required_evidence=("deploy_log",), + start_deadline_seconds=1, + ), + ), + ) + coordinator.declare(plan) + replacement = ExecutionPlanManifest.from_dict( + {**plan.as_dict(), "source_oid": "5" * 40} + ) + with self.assertRaises(RuntimeError): + coordinator.declare(replacement) + record = coordinator.task_coordinator.registry.get( + "task-plan-supersession" + ) + coordinator.task_coordinator.registry.evaluate_deadlines( + record.task_id, + now=parse_time(record.declared_at) + timedelta(seconds=2), + ) + coordinator.declare(replacement) + self.assertEqual( + "5" * 40, + coordinator.plan_store.get(plan.plan_id).source_oid, + ) + updated = coordinator.task_coordinator.registry.get(record.task_id) + self.assertEqual("5" * 40, updated.manifest.source_oid) + events = coordinator.task_coordinator.registry.events(record.task_id) + self.assertIn("MANIFEST_SUPERSEDED", {item["event_type"] for item in events}) + def test_execution_center_has_no_diff_markers(self) -> None: self.assertIn('id="execution"', PORTAL_HTML) self.assertIn("async function execution()", PORTAL_HTML)