@@ -88,6 +88,11 @@
     "experience-incubation": "experience",
     "profile-source-window": "profile",
 }
+_COMPATIBLE_DREAM_BINDING_ADDITIONS = {
+    "handoff.dream.v1": "handoff",
+    "prompt.dream.v1": "prompt",
+}
+_COMPATIBLE_DREAM_CAPABILITY_ADDITIONS = {"handoff", "prompt"}
 _STATE_COLUMNS = {
     "last_schedule_checkpoint_at": "DATETIME NULL",
     "scan_generation": "BIGINT NOT NULL DEFAULT 0",
@@ -164,6 +169,7 @@ async def bootstrap_processing_schema(
 
     marker = await _marker(connection)
     if marker is not None:
+        await _upgrade_compatible_dream_manifest(connection, config_manifest)
         await assert_processing_schema_ready(connection, config_manifest)
         return
     columns = await _columns(connection, STATES.name)
@@ -234,6 +240,82 @@ async def assert_processing_schema_ready(
         raise ProcessingSchemaNotReadyError(reason="; ".join(problems))
 
 
+async def _upgrade_compatible_dream_manifest(
+    connection: AsyncConnection,
+    config_manifest: Mapping[str, Any] | str | None,
+) -> None:
+    """Record the additive Handoff and Prompt bindings on existing deployments."""
+
+    if config_manifest is None:
+        return
+    marker = await _marker(connection, for_update=True)
+    if marker is None or marker["schema_version"] != PROCESSING_SCHEMA_VERSION or marker["phase"] != "complete":
+        return
+    previous_manifest = _manifest(str(marker["config_manifest"]))
+    current_manifest = _manifest(config_manifest)
+    if _deployment_manifest(previous_manifest) == _deployment_manifest(current_manifest):
+        return
+    previous = json.loads(previous_manifest)
+    current = json.loads(current_manifest)
+    if not _is_compatible_dream_manifest_extension(previous, current):
+        return
+    upgraded = dict(previous)
+    upgraded["bindings"] = {**previous["bindings"], **_manifest_binding_additions(previous, current)}
+    upgraded["capabilities"] = current["capabilities"]
+    upgraded_manifest = _manifest(upgraded)
+    if _deployment_manifest(upgraded_manifest) != _deployment_manifest(current_manifest):
+        return
+    await connection.execute(
+        update(SCHEMA)
+        .where(SCHEMA.c.singleton == 1, SCHEMA.c.config_manifest == marker["config_manifest"])
+        .values(config_manifest=upgraded_manifest)
+    )
+
+
+def _is_compatible_dream_manifest_extension(previous: Mapping[str, Any], current: Mapping[str, Any]) -> bool:
+    if set(previous) != set(current) or previous.get("mode") != current.get("mode"):
+        return False
+    previous_bindings = previous.get("bindings")
+    current_bindings = current.get("bindings")
+    previous_capabilities = previous.get("capabilities")
+    current_capabilities = current.get("capabilities")
+    if (
+        not isinstance(previous_bindings, dict)
+        or not isinstance(current_bindings, dict)
+        or not isinstance(previous_capabilities, list)
+        or not isinstance(current_capabilities, list)
+        or any(not isinstance(item, str) for item in (*previous_capabilities, *current_capabilities))
+    ):
+        return False
+    if any(current_bindings.get(name) != family for name, family in previous_bindings.items()):
+        return False
+    additions = _manifest_binding_additions(previous, current)
+    if not additions or any(
+        _COMPATIBLE_DREAM_BINDING_ADDITIONS.get(name) != family for name, family in additions.items()
+    ):
+        return False
+    previous_capability_set = set(previous_capabilities)
+    current_capability_set = set(current_capabilities)
+    added_capabilities = current_capability_set - previous_capability_set
+    return (
+        previous_capability_set <= current_capability_set
+        and added_capabilities <= _COMPATIBLE_DREAM_CAPABILITY_ADDITIONS
+        and added_capabilities <= set(additions.values())
+    )
+
+
+def _manifest_binding_additions(previous: Mapping[str, Any], current: Mapping[str, Any]) -> dict[str, str]:
+    previous_bindings = previous.get("bindings")
+    current_bindings = current.get("bindings")
+    if not isinstance(previous_bindings, dict) or not isinstance(current_bindings, dict):
+        return {}
+    return {
+        name: family
+        for name, family in current_bindings.items()
+        if isinstance(name, str) and isinstance(family, str) and name not in previous_bindings
+    }
+
+
 async def apply_processing_migration(  # noqa: C901 - explicit durable maintenance phases
     connection: AsyncConnection,
     *,
 
@@ -592,6 +592,10 @@ async def approve(  # noqa: C901 - each Family keeps its own transactional publi
         async with self._connection() as connection:
             current = await self._candidates.get(connection, self._scope_id, candidate_id)
         await self._authorize_decision("approve", current)
+        if current.status is CandidateStatus.APPROVED:
+            if current.version != expected_version:
+                raise CandidateConflictError(candidate_id, expected_version, current.version)
+            return _reviewed_candidate(current)
         if current.family == Memory.family:
             return await self._approve_memory_dream(candidate_id, expected_version)
         if current.family == TopicMemory.family:
 
@@ -15,6 +15,7 @@
 from __future__ import annotations
 
 import asyncio
+import json
 from contextlib import asynccontextmanager
 from pathlib import Path
 from types import SimpleNamespace
@@ -29,6 +30,7 @@
 from powercontext.builtin.persistence.processing_migration import (
     ProcessingSchemaNotReadyError,
     apply_processing_migration,
+    bootstrap_processing_schema,
 )
 from powercontext.builtin.persistence.sqlite import SQLiteConfig, SQLiteProfile
 from powercontext.builtin.persistence.supervision import ArtifactProcessingFence
@@ -56,6 +58,18 @@
 
 _FAMILIES = {"memory", "topic-memory", "experience", "profile"}
 _BINDING = "topic-memory-source-window"
+_PRE_HANDOFF_PROMPT_MANIFEST = {
+    "mode": "global",
+    "capabilities": [],
+    "bindings": {
+        "memory-source-window": "memory",
+        "topic-memory-source-window": "topic-memory",
+        "experience-incubation": "experience",
+        "profile-source-window": "profile",
+        "skill.dream.v1": "skill",
+    },
+    "legacy_automatic_bindings": ["topic-memory-source-window"],
+}
 _OCEANBASE = OceanBaseConfig(
     url=SecretStr("mysql+aoceanbase://root%40test@127.0.0.1:2881/composition_unused?charset=utf8mb4")
 )
@@ -103,6 +117,37 @@ async def scenario() -> None:
     asyncio.run(scenario())
 
 
+def test_startup_upgrades_manifest_written_by_pre_handoff_prompt_runtime(tmp_path: Path) -> None:
+    async def scenario() -> None:
+        database = _sqlite(tmp_path / "pre-dream-bindings.db")
+        config = BuiltinConfig(database=database, runtime=RuntimeConfig(dream_enabled=False))
+        # This is the completed manifest emitted by the pre-Handoff/Prompt runtime,
+        # bootstrapped through the same SQLite schema path used by that version.
+        async with (
+            SQLiteProfile.open(database, tables=SHARED_TABLES) as profile,
+            profile.database.transaction() as connection,
+        ):
+            await bootstrap_processing_schema(connection, _PRE_HANDOFF_PROMPT_MANIFEST)
+
+        async with open_builtin_contexts(config):
+            pass
+
+        async with (
+            SQLiteProfile.open(database, tables=SHARED_TABLES) as profile,
+            profile.database.transaction() as connection,
+        ):
+            marker = (await connection.execute(select(ARTIFACT_PROCESSING_SCHEMA_TABLE))).mappings().one()
+        stored = json.loads(marker["config_manifest"])
+        current = canonical_processing_manifest(config)
+        assert marker["phase"] == "complete"
+        assert marker["migration_id"] == "fresh"
+        assert stored["bindings"] == current["bindings"]
+        assert stored["capabilities"] == current["capabilities"]
+        assert stored["legacy_automatic_bindings"] == _PRE_HANDOFF_PROMPT_MANIFEST["legacy_automatic_bindings"]
+
+    asyncio.run(scenario())
+
+
 def test_nonempty_legacy_sqlite_is_rejected_without_importing_or_discarding_work(tmp_path: Path) -> None:
     async def scenario() -> None:
         database = _sqlite(tmp_path / "legacy.db")
 
@@ -666,7 +666,7 @@ async def scenario() -> None:
                 completed.error,
             )
             assert (await handoffs.latest()).revision == 1
-            from powercontext.builtin.review.errors import InvalidCandidateError
+            from powercontext.builtin.review.errors import CandidateConflictError, InvalidCandidateError
 
             candidate = await runtime.review.for_scope(scope_id).get(
                 GetArtifactCandidateRequest(candidate_id=completed.candidate.candidate_id)
@@ -690,6 +690,20 @@ async def scenario() -> None:
                 )
             )
             assert approved.result_artifact is not None
+            retry = await runtime.review.for_scope(scope_id).approve(
+                ApproveArtifactCandidateRequest(
+                    candidate_id=completed.candidate.candidate_id,
+                    expected_version=completed.candidate.version,
+                )
+            )
+            assert retry == approved
+            with pytest.raises(CandidateConflictError):
+                await runtime.review.for_scope(scope_id).approve(
+                    ApproveArtifactCandidateRequest(
+                        candidate_id=completed.candidate.candidate_id,
+                        expected_version=completed.candidate.version + 1,
+                    )
+                )
             updated = await handoffs.latest()
             assert updated is not None and updated.revision == 2
             assert updated.content.generation is None
 
@@ -38,7 +38,11 @@
 from powercontext.builtin.inference.models import GenerationResult, InferenceUsage
 from powercontext.builtin.persistence.artifacts import RepositoryArtifactDraft
 from powercontext.builtin.records import ArtifactWrite
-from powercontext.builtin.review.errors import ArtifactTargetConflictError, InvalidCandidateError
+from powercontext.builtin.review.errors import (
+    ArtifactTargetConflictError,
+    CandidateConflictError,
+    InvalidCandidateError,
+)
 from powercontext.builtin.runtime import (
     ApproveArtifactCandidateRequest,
     BuiltinConfig,
@@ -187,6 +191,20 @@ async def scenario():
                     )
                 )
                 assert approved.result_artifact.revision == 2
+                assert (
+                    await review.approve(
+                        ApproveArtifactCandidateRequest(
+                            candidate_id=candidate.candidate_id, expected_version=candidate.version
+                        )
+                    )
+                    == approved
+                )
+                with pytest.raises(CandidateConflictError):
+                    await review.approve(
+                        ApproveArtifactCandidateRequest(
+                            candidate_id=candidate.candidate_id, expected_version=candidate.version + 1
+                        )
+                    )
                 assert current_prompt("memory.extract") == frozen
                 assert frozen.artifact == target
             fresh = await prompts.resolve(scope.scope_id, "memory.extract")