v2r1c2_guard_plan_oid_supersession

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)

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)