diff --git a/linktools-cntr/docs/lifecycle.md b/linktools-cntr/docs/lifecycle.md
index 598e45f767..b3efe880a6 100644
--- a/linktools-cntr/docs/lifecycle.md
+++ b/linktools-cntr/docs/lifecycle.md
@@ -111,7 +111,7 @@ The operation then runs:
 4. Prepare explicit images, then run `on_check` and `CHECK` for actual targets.
    All candidate files exist, but no service has been stopped or replaced.
 5. Apply services through Compose and inspect the resulting runtime state once
-   to verify the intended image, replica count and start/success-exit state.
+   to verify replica count and start/success-exit state. Compose owns image convergence.
    Do not wait for health checks or job completion; record successful models.
    The source paths in those models identify the exact generated files used.
 6. `on_started` and `AFTER_START` run for applied owners. A failure here reports
@@ -246,7 +246,9 @@ prepared configuration. They are not replaced by this file mechanism.
 
 Changed generated-file source paths or other Compose definitions recreate the affected service.
 Unchanged definitions are not force-recreated. Docker Compose can still replace a
-service when its locally selected image changes. Recreating a container does not
+service when its locally selected image changes. After Compose succeeds, ct-cntr
+checks replica count and observed state, not image IDs from different Docker
+objects. Recreating a container does not
 by itself rebuild its image or request a new certificate.
 
 Build identity comes from the final Compose inputs. The image preparer adds the
@@ -286,8 +288,10 @@ health and job outcomes are observed separately through status/logs. The persist
 container does not block later dependents through a framework readiness loop.
 
 The per-service resolved applied model is the sole recovery authority. Recovery
-pins the pre-operation image ID and restores failed or explicitly stopped
-services using their old immutable file paths. Existing recovery for changed
+pins a locally resolvable pre-operation image ID and restores failed or
+explicitly stopped services using their old immutable file paths. A failed batch
+leaves a service alone when all its original replicas remain in place and those
+originally running are still running, unless recovery must rebind a dependency. Existing recovery for changed
 shared bind inputs also applies, including transitive consumers. New services
 without a prior instance are stopped when required by recovery; confirmed
 independent successes remain applied. Initial migration may read old saved
diff --git a/linktools-cntr/src/linktools/cntr/_operations.py b/linktools-cntr/src/linktools/cntr/_operations.py
index bf77f676eb..c2c9daffd2 100644
--- a/linktools-cntr/src/linktools/cntr/_operations.py
+++ b/linktools-cntr/src/linktools/cntr/_operations.py
@@ -13,7 +13,7 @@
     from collections.abc import Sequence, Iterable
     from .container import BaseContainer
     from .manager import ContainerManager
-    from .runtime.inspect import ProjectRuntimeState
+    from .runtime.inspect import ProjectRuntimeState, ServiceRuntimeState
 
 
 @dataclass(frozen=True)
@@ -423,38 +423,21 @@ def warn_collateral(service, action, reason):
                         self._update_running_state(context, running, stop_targets)
                 for service in selection.services:
                     spec = context.compose_model["services"][service]
-                    existing = context.initial_runtime_state.existing_services
-                    before_image = context.initial_runtime_state.image_ids.get(service)
-                    image_changed = service in existing and before_image != target_image_ids[service]
-                    binds = set(namespace_dependencies(spec))
-                    recreate = (service in model_store.changed_services or image_changed or
-                                bool(binds & recreated))
-                    cascade = any((options.get("restart") and dependency in updated) or
-                                  (dependency in recreated and dependency in existing)
-                                  for dependency, options in service_dependencies(spec).items())
                     if spec.get("scale") == 0 or spec.get("deploy", {}).get("replicas") == 0:
-                        if service in context.initial_runtime_state.existing_services:
+                        if service in actual.existing_services:
                             warn_collateral(service, "remove", "the desired service scale is zero")
-                    elif recreate or service not in initial:
-                        action = "recreate" if service in context.initial_runtime_state.existing_services and recreate else "start"
-                        reason = ("a namespace provider was replaced" if binds & recreated else
-                                  "the selected image changed" if image_changed else
-                                  "the required service configuration changed" if service in initial else
-                                  "the required service is not running")
-                        warn_collateral(service, action, reason)
-                    elif cascade:
-                        warn_collateral(service, "restart", "a declared dependency was updated")
-                    if recreate or service not in existing:
-                        recreated.add(service)
-                    if recreate or cascade or service in stop_set or service not in initial:
-                        updated.add(service)
+                    elif service in model_store.changed_services and service in initial:
+                        warn_collateral(service, "recreate", "the required service configuration changed")
+                    elif service not in initial:
+                        warn_collateral(service, "start", "the required service is not running")
                 failed = set(selection.services)
                 with record_phase(context, "up", logger=manager.logger):
                     batch_attempted = True
                     runner.apply_services(context, selection.services)
-                    observed = runner.verify_services_started(context, selection.services, target_image_ids)
+                    observed = runner.verify_services_started(context, selection.services)
                     model_store.record(selection.services)
                 batch_completed = True
+                updated, recreated = self._observed_service_changes(actual, observed, selection.services)
                 running.difference_update(selection.services)
                 running.update(service for service in selection.services if any(
                     item.state in ("running", "restarting") for item in observed.service_instances(service)))
@@ -481,9 +464,21 @@ def warn_collateral(service, action, reason):
                     failed.clear()
             except (Exception, KeyboardInterrupt) as error:
                 try:
+                    retained = set()
                     if batch_attempted and not batch_completed:
                         observed = manager.docker_inspector.get_project_state(selection.project_containers)
-                        confirmed = runner.started_services(context, selection.services, observed, target_image_ids)
+                        confirmed = runner.started_services(context, selection.services, observed)
+                        updated, recreated = self._observed_service_changes(actual, observed, selection.services)
+                        # An old replica (including an exited one-shot) does not prove
+                        # that a failed batch applied the image now behind its tag.
+                        for service in actual.existing_services.intersection(selection.services) - pending_restart:
+                            before = actual.service_instances(service)
+                            after = observed.service_instances(service)
+                            old_ids = {item.container_id for item in before if item.container_id}
+                            if any(item.container_id in old_ids for item in after):
+                                confirmed.discard(service)
+                            if service in initial and self._running_replicas_retained(before, after):
+                                retained.add(service)
                         for service in confirmed & model_store.changed_services & actual.existing_services:
                             previous_hashes = {item.labels.get("com.docker.compose.config-hash")
                                                for item in actual.service_instances(service)}
@@ -520,7 +515,7 @@ def warn_collateral(service, action, reason):
                             if related <= affected:
                                 break
                             affected.update(related)
-                        restore.update(affected & initial)
+                        restore.update((affected & initial) - (failed & retained))
                         cleanup = affected - initial
                         if cleanup:
                             discarded = set(cleanup)
@@ -640,6 +635,29 @@ def stop_temporary():
         if report:
             render_report(manager.logger, get_records(context))
 
+    def _observed_service_changes(self, before: "ProjectRuntimeState", after: "ProjectRuntimeState",
+                                  services: "Sequence[str]") -> "tuple[set[str], set[str]]":
+        updated, recreated = set(), set()
+        for service in services:
+            old = {item.container_id: item.started_at for item in before.service_instances(service)}
+            new = {item.container_id: item.started_at for item in after.service_instances(service)}
+            if None in old or None in new:
+                continue
+            if old.keys() != new.keys():
+                recreated.add(service)
+                updated.add(service)
+            elif any(started_at and old[identity] != started_at for identity, started_at in new.items()):
+                updated.add(service)
+        return updated, recreated
+
+    def _running_replicas_retained(self, before: "Sequence[ServiceRuntimeState]",
+                                   after: "Sequence[ServiceRuntimeState]") -> bool:
+        old = {item.container_id: item.state for item in before}
+        new = {item.container_id: item.state for item in after}
+        return None not in old and old.keys() == new.keys() and all(
+            new[identity] in ("running", "restarting") if state in ("running", "restarting")
+            else new[identity] == state for identity, state in old.items())
+
     def _check_build_scope(self, model: dict, services: "Iterable[str]", selection: ComposeSelection) -> None:
         pending, visited, outside = list(services), set(), set()
         selected = set(selection.services)
diff --git a/linktools-cntr/src/linktools/cntr/runtime/compose.py b/linktools-cntr/src/linktools/cntr/runtime/compose.py
index 887e7169fa..c58f8df2b0 100644
--- a/linktools-cntr/src/linktools/cntr/runtime/compose.py
+++ b/linktools-cntr/src/linktools/cntr/runtime/compose.py
@@ -389,15 +389,15 @@ def apply_services(self, context: "OperationContext", services: "Sequence[str]")
             return self.manager.runtime.create_docker_process(*prefix, *args).check_call()
 
     def started_services(self, context: "OperationContext", services: "Sequence[str]",
-                       state: "ProjectRuntimeState", image_ids: "Mapping[str, str]") -> "set[str]":
-        """Services started with the intended image and replica count."""
+                       state: "ProjectRuntimeState") -> "set[str]":
+        """Services with the requested replica count and start state after Compose."""
         definitions = context.compose_model["services"]
         started = set()
         for service in services:
             spec = definitions[service]
             replicas = spec.get("scale", spec.get("deploy", {}).get("replicas", 1))
             matches = state.service_instances(service)
-            if len(matches) != replicas or any(item.image_id != image_ids[service] for item in matches):
+            if len(matches) != replicas:
                 continue
             valid = all(item.state in ("running", "restarting") or
                         (item.state == "exited" and item.exit_code == 0)
@@ -406,12 +406,11 @@ def started_services(self, context: "OperationContext", services: "Sequence[str]
                 started.add(service)
         return started
 
-    def verify_services_started(self, context: "OperationContext", services: "Sequence[str]",
-                              image_ids: "Mapping[str, str]") -> "ProjectRuntimeState":
+    def verify_services_started(self, context: "OperationContext", services: "Sequence[str]") -> "ProjectRuntimeState":
         """Validate one observed batch after detached Compose start."""
         from ..errors import ContainerError
         state = self.manager.docker_inspector.get_project_state(context.project_containers, services=services)
-        pending = set(services) - self.started_services(context, services, state, image_ids)
+        pending = set(services) - self.started_services(context, services, state)
         if pending:
             raise ContainerError("Service {} did not reach its desired runtime state".format(
                 ", ".join(sorted(pending))))
@@ -563,6 +562,10 @@ def apply_saved_services(self, context: "OperationContext", services: "Sequence[
             if not image:
                 raise ContainerError("No original image ID for service " + service)
             overlay[service] = {"image": image}
+        available = self.manager.image_preparer.inspect_images(tuple(spec["image"] for spec in overlay.values()))
+        for service, spec in overlay.items():
+            if available[spec["image"]] is None:
+                raise ContainerError("Original image is not available locally for service " + service)
         contents = [*files.values(), yaml.safe_dump({"services": overlay})]
         with self._saved_compose_args(context, contents) as args:
             model = self._resolved_model(self.manager.runtime.create_docker_process(
diff --git a/linktools-cntr/src/linktools/cntr/runtime/inspect.py b/linktools-cntr/src/linktools/cntr/runtime/inspect.py
index 548038122b..47b8bc111b 100644
--- a/linktools-cntr/src/linktools/cntr/runtime/inspect.py
+++ b/linktools-cntr/src/linktools/cntr/runtime/inspect.py
@@ -63,6 +63,8 @@ class ServiceRuntimeState:
     labels: "dict[str, str]"
     image_id: "str | None" = None
     namespace_bindings: "dict[str, str | list[str]] | None" = None
+    container_id: "str | None" = None
+    started_at: "str | None" = None
 
 
 _RUNNING_STATES = ("running", "restarting")
@@ -234,6 +236,8 @@ def _map_inspect_item(
         exit_code=_normalize_exit_code(state_data.get("ExitCode")),
         labels={str(k): str(v) for k, v in labels.items()},
         image_id=item.get("Image"),
+        container_id=item.get("Id"),
+        started_at=state_data.get("StartedAt"),
     )
 
 
@@ -388,7 +392,7 @@ def get_project_state(self, containers: "Iterable[BaseContainer]",
             if mapped is not None:
                 mapped_items.append((item, mapped))
                 if mapped.service and not _is_oneoff(mapped):
-                    for identity in (item.get("Id"), mapped.runtime_name):
+                    for identity in (mapped.container_id, mapped.runtime_name):
                         if identity:
                             identities[identity] = mapped.service
         services = []
diff --git a/tests/cntr/_harness.py b/tests/cntr/_harness.py
index a9ee633e30..e901a1b93a 100644
--- a/tests/cntr/_harness.py
+++ b/tests/cntr/_harness.py
@@ -221,8 +221,8 @@ def stub_generated_runtime(manager, monkeypatch):
     monkeypatch.setattr(manager.docker_inspector, "get_project_state", lambda containers, **kwargs:
                         ProjectRuntimeState(manager.project_name, (), "docker"))
     monkeypatch.setattr(manager.compose_runner, "verify_services_started",
-                        lambda context, services, image_ids: ProjectRuntimeState(manager.project_name, tuple(
-                            SimpleNamespace(service=name, state="running", labels={}) for name in services), "docker"))
+                        lambda context, services: ProjectRuntimeState(manager.project_name, tuple(
+                            SimpleNamespace(service=name, state="running", labels={}, container_id=name, started_at=None) for name in services), "docker"))
     monkeypatch.setattr(manager.compose_runner, "validate_service", lambda *args, **kwargs:
                         SimpleNamespace(succeeded=True, stdout="", stderr="", returncode=0))
     monkeypatch.setattr(manager.image_preparer, "image_id", lambda image: "sha256:local-" + image)
diff --git a/tests/cntr/test_batch_service_application.py b/tests/cntr/test_batch_service_application.py
index b4d65b6e66..50b56fd76b 100644
--- a/tests/cntr/test_batch_service_application.py
+++ b/tests/cntr/test_batch_service_application.py
@@ -14,8 +14,8 @@
 from test_lifecycle_rebuild import setup_case
 
 
-def instance(service, image="sha256:target", state="running", health=None, code=None, config="new"):
-    return SimpleNamespace(service=service, image_id=image, state=state, health=health, exit_code=code,
+def instance(service, image="sha256:target", state="running", health=None, code=None, config="new", container_id=None, started_at=None):
+    return SimpleNamespace(service=service, image_id=image, container_id=container_id, started_at=started_at, state=state, health=health, exit_code=code,
                            labels={"com.docker.compose.config-hash": config})
 
 
@@ -44,12 +44,12 @@ def process(*args):
 
 
 @pytest.mark.parametrize("observed", [[], [instance("app")],
-    [instance("app"), instance("app", image="sha256:old")],
-    [instance("app"), instance("app", image="sha256:old")]])
-def test_batch_rejects_incomplete_replicas_and_wrong_images(observed):
+    [instance("app"), instance("app", state="created")],
+    [instance("app"), instance("app", state="exited", code=1)]])
+def test_batch_rejects_incomplete_or_failed_replicas(observed):
     runner, context, queries = readiness_case({"app": {"deploy": {"replicas": 2}}}, [observed])
     with pytest.raises(ContainerError, match="Service app did not reach"):
-        runner.verify_services_started(context, ("app",), {"app": "sha256:target"})
+        runner.verify_services_started(context, ("app",))
     assert len(queries) == 1
 
 
@@ -59,16 +59,16 @@ def test_completed_success_dependency_does_not_wait_for_job_exit():
     runner, context, queries = readiness_case(
         specs, [[instance("job", state="running"), instance("app")]])
     result = runner.verify_services_started(
-        context, ("job", "app"), {"job": "sha256:target", "app": "sha256:target"})
+        context, ("job", "app"))
     assert result.running_services == {"job", "app"}
     assert queries == [("job", "app")]
 
 
 def test_zero_replicas_requires_observed_removal():
     runner, context, _ = readiness_case({"app": {"scale": 0}}, [[]])
-    assert runner.verify_services_started(context, ("app",), {"app": "sha256:target"}).services == ()
+    assert runner.verify_services_started(context, ("app",)).services == ()
     state = ProjectRuntimeState("test", (instance("app"),), "docker")
-    assert runner.started_services(context, ("app",), state, {"app": "sha256:target"}) == set()
+    assert runner.started_services(context, ("app",), state) == set()
 
 
 def test_partial_batch_failure_preserves_successful_sibling_but_restores_mixed_config_replicas(tmp_path):
@@ -104,19 +104,24 @@ def test_maintenance_oneoff_does_not_count_as_a_service_replica():
     maintenance = instance("app", state="exited", code=0)
     maintenance.labels["com.docker.compose.oneoff"] = "True"
     runner, context, queries = readiness_case({"app": {}}, [[instance("app"), maintenance]])
-    runner.verify_services_started(context, ("app",), {"app": "sha256:target"})
+    runner.verify_services_started(context, ("app",))
     assert len(queries) == 1
-    assert runner.started_services(context, ("app",), ProjectRuntimeState("test", (maintenance,), "docker"),
-                                 {"app": "sha256:target"}) == set()
+    assert runner.started_services(context, ("app",), ProjectRuntimeState("test", (maintenance,), "docker")) == set()
 
 
-def test_failed_batch_does_not_commit_or_rerun_an_untouched_previously_completed_job(tmp_path):
+@pytest.mark.parametrize("action", ["up", "restart"])
+@pytest.mark.parametrize("config_changed", [False, True])
+def test_failed_batch_does_not_commit_or_rerun_an_untouched_previously_completed_job(
+        tmp_path, monkeypatch, config_changed, action):
     manager = setup_case(tmp_path, [("job", {"job": {
         "image": "job:current", "environment": {"VERSION": "new"}}})])
     previous = deepcopy(manager.model)
-    previous["services"]["job"]["environment"]["VERSION"] = "old"
+    if config_changed:
+        previous["services"]["job"]["environment"]["VERSION"] = "old"
     AppliedServiceModels(manager, previous).record(("job",))
-    stopped = instance("job", image="sha256:old-job", state="exited", code=0, config="old")
+    manager.image_preparer.image_id = lambda reference: "sha256:platform-index"
+    stopped = instance("job", image="sha256:old-job", state="exited", code=0, config="old",
+                       container_id="old-job")
     manager.docker_inspector.get_project_state = lambda *args, **kwargs: ProjectRuntimeState(
         "test", (stopped,), "docker")
 
@@ -126,9 +131,66 @@ def fail_before_application(context, services):
     runner = manager.compose_runner
     runner.apply_services = fail_before_application
     runner.started_services = ComposeRunner(manager).started_services
+    original_record = AppliedServiceModels.record
+
+    def record(models, services):
+        assert "job" not in services, "old successful exit is not evidence that the job reran"
+        return original_record(models, services)
+
+    monkeypatch.setattr(AppliedServiceModels, "record", record)
     with pytest.raises(ContainerError, match="native failure before job application"):
-        manager.compose_operations.up()
+        getattr(manager.compose_operations, action)(["job"])
     receipts = AppliedServiceModels(manager, manager.model)
-    assert receipts.previous_model("job")["services"]["job"]["environment"]["VERSION"] == "old"
+    assert receipts.previous_model("job")["services"]["job"]["environment"]["VERSION"] == (
+        "old" if config_changed else "new")
     assert not any(event[0] in ("apply", "restore", "restart") for event in manager.events)
     assert manager.running_state.get_persisted() == []
+
+
+@pytest.mark.parametrize("rebuilt", [False, True])
+def test_native_batch_accepts_platform_image_after_compose_converges(tmp_path, rebuilt):
+    manager = setup_case(tmp_path, [("app", {"app": {"image": "app:latest"}})], running=("app",))
+    AppliedServiceModels(manager, manager.model).record(("app",))
+    observed = [instance("app", image="sha256:original-config", container_id="original")]
+    manager.docker_inspector.get_project_state = lambda *args, **kwargs: ProjectRuntimeState(
+        "test", tuple(observed), "docker")
+    manager.image_preparer.image_id = lambda reference: "sha256:platform-index"
+    runner = manager.compose_runner
+    runner.verify_services_started = ComposeRunner(manager).verify_services_started
+
+    def apply(context, services):
+        assert tuple(services) == ("app",)
+        if rebuilt:
+            observed[:] = [instance("app", image="sha256:new-config", container_id="replacement")]
+
+    runner.apply_services = apply
+    manager.compose_operations.up()
+    assert manager.running_state.get_persisted() == ["app"]
+    assert observed[0].container_id == ("replacement" if rebuilt else "original")
+    assert not any(event[0] in ("stop", "restore") for event in manager.events)
+
+
+@pytest.mark.parametrize("remaining_ids,restore", [
+    (("original-1", "original-2"), False),
+    (("original-1", "replacement"), True),
+    (("original-1",), True),
+])
+def test_partial_failure_preserves_only_the_untouched_original_replicas(tmp_path, remaining_ids, restore):
+    manager = setup_case(tmp_path, [("app", {"app": {"image": "app:latest", "scale": 2}})], running=("app",))
+    original = [instance("app", image="sha256:old-config", config="old", container_id=identity)
+                for identity in ("original-1", "original-2")]
+    observed = list(original)
+    manager.docker_inspector.get_project_state = lambda *args, **kwargs: ProjectRuntimeState(
+        "test", tuple(observed), "docker")
+    runner = manager.compose_runner
+    runner.started_services = ComposeRunner(manager).started_services
+
+    def fail(context, services):
+        observed[:] = [instance("app", image="sha256:old-config", config="old", container_id=identity)
+                       for identity in remaining_ids]
+        raise ContainerError("native failure")
+
+    runner.apply_services = fail
+    with pytest.raises(ContainerError, match="^native failure$"):
+        manager.compose_operations.up()
+    assert [event[1] for event in manager.events if event[0] == "restore"] == ([("app",)] if restore else [])
diff --git a/tests/cntr/test_cli_routing.py b/tests/cntr/test_cli_routing.py
index fd81764cb5..97fbadce1c 100644
--- a/tests/cntr/test_cli_routing.py
+++ b/tests/cntr/test_cli_routing.py
@@ -11,7 +11,7 @@
 from linktools.cntr.lifecycle.hooks import HookRegistry
 from types import SimpleNamespace
 from linktools.cntr.runtime.images import ImagePlan
-from linktools.cntr.runtime.inspect import ProjectRuntimeState
+from linktools.cntr.runtime.inspect import ProjectRuntimeState, ServiceRuntimeState
 
 _PROXY_KEYS = ("http_proxy", "https_proxy", "all_proxy", "no_proxy",
                "HTTP_PROXY", "HTTPS_PROXY", "ALL_PROXY", "NO_PROXY")
@@ -25,8 +25,10 @@ def _ready_dependencies(manager, monkeypatch) -> None:
     model = {"services": {name: {"image": name + ":current"}
                           for container in manager.containers.values() for name in container.services}}
     actual = ProjectRuntimeState(manager.project_name, tuple(
-        SimpleNamespace(service=name, state="running", health="healthy", exit_code=None, labels={},
-                        image_id="sha256:local-" + name + ":current") for name in sorted(services)), "docker")
+        ServiceRuntimeState(logical_containers=(), service=name, runtime_name=name, state="running",
+                            health="healthy", image=name + ":current", exit_code=None, labels={},
+                            image_id="sha256:local-" + name + ":current", container_id=name,
+                            started_at="2026-01-01T00:00:00Z") for name in sorted(services)), "docker")
     monkeypatch.setattr(manager.docker_inspector, "get_project_state", lambda containers, **kwargs: actual)
     AppliedServiceModels(manager, model).record(services)
 
diff --git a/tests/cntr/test_deployment_recovery_boundaries.py b/tests/cntr/test_deployment_recovery_boundaries.py
index e217a6c21d..b87c508882 100644
--- a/tests/cntr/test_deployment_recovery_boundaries.py
+++ b/tests/cntr/test_deployment_recovery_boundaries.py
@@ -90,7 +90,7 @@ def check(context):
 def test_running_orphan_does_not_abort_after_applying_current_services(tmp_path, names):
     manager = setup_case(tmp_path, [("app", {"app": {"image": "app:new"}})], running=("app",))
     actual = manager.docker_inspector.get_project_state(None)
-    actual = replace(actual, services=actual.services + (SimpleNamespace(service="retired", state="running", image_id="sha256:retired",
+    actual = replace(actual, services=actual.services + (SimpleNamespace(service="retired", container_id=None, started_at=None, state="running", image_id="sha256:retired",
                                        labels={}, health=None, exit_code=None),))
     manager.docker_inspector.get_project_state = lambda containers: actual
     manager.compose_operations.up(names)
@@ -391,7 +391,7 @@ def stopped_provider_case(tmp_path, binding):
     AppliedServiceModels(manager, manager.model).record(("db", "net", "child", "stopped"))
     manager.model["services"]["db"]["environment"] = {"VERSION": "new"}
     actual = manager.docker_inspector.get_project_state(None)
-    actual = replace(actual, services=actual.services + (SimpleNamespace(service="db", state="exited", image_id="sha256:stopped-db",
+    actual = replace(actual, services=actual.services + (SimpleNamespace(service="db", container_id=None, started_at=None, state="exited", image_id="sha256:stopped-db",
                                        labels={}, health=None, exit_code=0),))
     manager.docker_inspector.get_project_state = lambda containers: actual
     return manager
@@ -518,7 +518,8 @@ def process(*args, **kwargs):
         return SimpleNamespace(check_call=lambda: 0)
 
     runner = ComposeRunner(SimpleNamespace(data_path=tmp_path, project_name="test",
-                                            runtime=SimpleNamespace(create_docker_process=process)))
+                                            runtime=SimpleNamespace(create_docker_process=process),
+                                            image_preparer=SimpleNamespace(inspect_images=lambda refs: dict.fromkeys(refs, object()))))
     runner._resolved_model = lambda process: {"services": {"db": {"image": "sha256:old-stopped"}}}
     context = SimpleNamespace(initial_runtime_state=ProjectRuntimeState("test", (), "docker"), project_containers=(),
                               is_full_project=False, target_services=("db",))
@@ -538,10 +539,11 @@ def process(*args, **kwargs):
         return SimpleNamespace(args=args, check_call=lambda: 0)
 
     runner = ComposeRunner(SimpleNamespace(data_path=tmp_path, project_name="test",
-                                          runtime=SimpleNamespace(create_docker_process=process)))
+                                          runtime=SimpleNamespace(create_docker_process=process),
+                                            image_preparer=SimpleNamespace(inspect_images=lambda refs: dict.fromkeys(refs, object()))))
     runner._resolved_model = lambda process: model if process.args[-1] == "optional" else {"services": {}}
     context = SimpleNamespace(initial_runtime_state=ProjectRuntimeState("test", (
-        SimpleNamespace(service="optional", state="running", image_id="sha256:original", labels={}),), "docker"), project_containers=(),
+        SimpleNamespace(service="optional", container_id=None, started_at=None, state="running", image_id="sha256:original", labels={}),), "docker"), project_containers=(),
                               service_models=SimpleNamespace(previous={}),
                               is_full_project=False, target_services=("optional",),
                               previous_compose_contents={"old.yml": yaml.safe_dump(model)})
@@ -577,6 +579,7 @@ def capture_needed(context, services):
         return capture(context, services)
 
     manager.compose_runner.saved_service_models = capture_needed
+    manager.docker_inspector.get_project_state(None).service_instances("db")[0].container_id = "original-db"
     manager.compose_operations.up()
     assert [event[1] for event in manager.events if event[0] == "apply"] == ["db"]
     if binding and "network_mode" in binding:
@@ -650,12 +653,26 @@ def test_native_false_or_empty_dependency_label_does_not_capture_unrelated_legac
     assert all(event[1] == ("db",) for event in manager.events if event[0] == "capture-restore")
 
 
-@pytest.mark.parametrize("scope", ["provider", "joint", "full"])
+@pytest.mark.parametrize("scope", ["provider", "joint", "full-replace", "full-restart"])
 def test_native_implicit_restart_edge_respects_operation_scope(tmp_path, scope):
     manager = legacy_labeled_case(tmp_path, {"depends_on": {"db": {"restart": True}}},
                                  "db:service_started:true", child_restart=True)
-    if scope == "full":
+    if scope.startswith("full-"):
         profile_nonproviders(manager, "db")
+        original = manager.docker_inspector.get_project_state(None).service_instances("db")[0]
+        original.container_id = "original-db"
+        original.started_at = "before"
+        if scope == "full-restart":
+            verify = manager.compose_runner.verify_services_started
+
+            def verify_native_restart(context, services):
+                state = verify(context, services)
+                db = state.service_instances("db")[0]
+                db.container_id = "original-db"
+                db.started_at = "after"
+                return state
+
+            manager.compose_runner.verify_services_started = verify_native_restart
         manager.compose_operations.up()
         assert [event[1] for event in manager.events if event[0] == "restart"] == ["worker", "child"]
         assert not any(event[0] == "restore" for event in manager.events)
@@ -765,7 +782,7 @@ def test_stopped_unrelated_service_does_not_add_a_recovery_requirement(tmp_path)
         ("net", {"net": {"image": "net:new"}}),
     ], running=("net",))
     actual = manager.docker_inspector.get_project_state(None)
-    actual = replace(actual, services=actual.services + (SimpleNamespace(service="db", state="exited", image_id=None,
+    actual = replace(actual, services=actual.services + (SimpleNamespace(service="db", container_id=None, started_at=None, state="exited", image_id=None,
                                        labels={}, health=None, exit_code=0),))
     manager.docker_inspector.get_project_state = lambda containers: actual
     manager.compose_operations.up(["db"])
@@ -780,7 +797,7 @@ def test_legacy_stopped_namespace_preflight_follows_newly_captured_models(tmp_pa
         ("net", {"net": {"image": "net:new", "network_mode": "service:db"}}),
     ], running=("net",))
     actual = manager.docker_inspector.get_project_state(None)
-    actual = replace(actual, services=actual.services + tuple(SimpleNamespace(service=name, state="exited", image_id="sha256:old-" + name,
+    actual = replace(actual, services=actual.services + tuple(SimpleNamespace(service=name, container_id=None, started_at=None, state="exited", image_id="sha256:old-" + name,
                                              labels={}, health=None, exit_code=0) for name in ("db", "vpn")))
     manager.docker_inspector.get_project_state = lambda containers: actual
     original = manager.compose_runner.saved_service_models
diff --git a/tests/cntr/test_integration_consumer_application.py b/tests/cntr/test_integration_consumer_application.py
index 6aeabaefba..f84c99ad78 100644
--- a/tests/cntr/test_integration_consumer_application.py
+++ b/tests/cntr/test_integration_consumer_application.py
@@ -255,7 +255,7 @@ def test_collateral_image_comes_only_from_unambiguous_running_replicas(tmp_path,
     from linktools.cntr.runtime.inspect import ProjectRuntimeState
     manager = consumer_case(tmp_path)
     actual = manager.docker_inspector.get_project_state(None)
-    extra = SimpleNamespace(service="consumer", state=other_state, image_id="sha256:other", labels={})
+    extra = SimpleNamespace(service="consumer", container_id=None, started_at=None, state=other_state, image_id="sha256:other", labels={})
     mixed = ProjectRuntimeState("test", actual.services + (extra,), "docker")
     manager.docker_inspector.get_project_state = lambda containers, **kwargs: mixed
     if other_state == "running":
diff --git a/tests/cntr/test_lifecycle_rebuild.py b/tests/cntr/test_lifecycle_rebuild.py
index e6b28e3a79..6a430a1ee9 100644
--- a/tests/cntr/test_lifecycle_rebuild.py
+++ b/tests/cntr/test_lifecycle_rebuild.py
@@ -138,13 +138,13 @@ def apply_services(self, context, services):
             if changed or cascade or service not in initial:
                 updated.add(service)
 
-    def started_services(self, context, services, state, image_ids):
+    def started_services(self, context, services, state):
         return set(self.applied)
 
-    def verify_services_started(self, context, services, image_ids):
-        active = self.started_services(context, services, None, image_ids)
+    def verify_services_started(self, context, services):
+        active = self.started_services(context, services, None)
         return ProjectRuntimeState("test", tuple(SimpleNamespace(
-            service=service, state="running", labels={}) for service in active), "docker")
+            service=service, state="running", labels={}, container_id=service + ":after", started_at=None) for service in active), "docker")
 
     def restart_service(self, context, service, model=None):
         self.manager.events.append(("restart", service))
@@ -188,7 +188,7 @@ def setup_case(tmp_path, groups, running=()):
     manager.resolver = SimpleNamespace(resolve_dependencies=lambda selected: selected)
     manager.model = {"services": {name: spec for owner in owners for name, spec in owner.services.items()}}
     state = ProjectRuntimeState("test", tuple(
-        SimpleNamespace(service=name, state="running", image_id="sha256:old-" + name, labels={}, health=None,
+        SimpleNamespace(service=name, state="running", container_id=None, started_at=None, image_id="sha256:old-" + name, labels={}, health=None,
                         exit_code=None) for name in running), "docker")
     def observed_state(containers, **kwargs):
         applied = set(getattr(manager.compose_runner, "applied", ()))
@@ -196,7 +196,7 @@ def observed_state(containers, **kwargs):
         previous = tuple(SimpleNamespace(**dict(vars(item), state="exited", exit_code=0))
                          if item.service in stopped else item for item in state.services if item.service not in applied)
         return ProjectRuntimeState("test", previous + tuple(
-            SimpleNamespace(service=name, state="running", health=None, exit_code=None,
+            SimpleNamespace(service=name, state="running", health=None, exit_code=None, container_id=name + ":after", started_at=None,
                             image_id=manager.image_preparer.image_id(manager.model["services"][name]["image"]),
                             labels={"com.docker.compose.config-hash": "new-" + name}) for name in applied), "docker")
     manager.docker_inspector = SimpleNamespace(get_project_state=observed_state)
@@ -338,7 +338,8 @@ def test_completed_job_not_marked_as_running(tmp_path):
     assert "job" in AppliedServiceModels(manager, manager.model).previous
 
 
-def test_restore_pins_original_image_and_does_not_inject_generation_label(tmp_path):
+@pytest.mark.parametrize("image_available", [True, False])
+def test_restore_requires_locally_available_original_image(tmp_path, image_available):
     commands = []
     documents = []
 
@@ -347,13 +348,20 @@ def process(*args, **kwargs):
         documents.append([Path(args[index + 1]).read_text() for index, value in enumerate(args[:-1]) if value == "--file"])
         return SimpleNamespace(check_call=lambda: 0)
 
-    manager = SimpleNamespace(data_path=tmp_path, project_name="test", runtime=SimpleNamespace(create_docker_process=process))
+    manager = SimpleNamespace(data_path=tmp_path, project_name="test", runtime=SimpleNamespace(create_docker_process=process),
+                              image_preparer=SimpleNamespace(inspect_images=lambda refs: dict.fromkeys(refs, object() if image_available else None)))
     runner = ComposeRunner(manager)
     runner._resolved_model = lambda process: {"services": {"app": {"image": "sha256:old"}}}
     ctx = OperationContext(project_containers=[SimpleNamespace(name="app", services={"app": {}})])
     ctx.initial_runtime_state = ProjectRuntimeState("test", (
         SimpleNamespace(service="app", state="running", image_id="sha256:old", labels={}),), "docker")
-    runner.apply_saved_services(ctx, ("app",), {"previous.yml": "services:\n  app:\n    image: mutable:tag\n"})
+    saved = {"previous.yml": "services:\n  app:\n    image: mutable:tag\n"}
+    if not image_available:
+        with pytest.raises(ContainerError, match="Original image is not available locally for service app"):
+            runner.apply_saved_services(ctx, ("app",), saved)
+        assert not commands
+        return
+    runner.apply_saved_services(ctx, ("app",), saved)
     overlay = yaml.safe_load(documents[-1][-1])
     assert overlay == {"services": {"app": {"image": "sha256:old"}}}
     assert "--force-recreate" in commands[-1]
diff --git a/tests/cntr/test_oneoff_runtime_state.py b/tests/cntr/test_oneoff_runtime_state.py
index 5f67b0a09c..826bede3ab 100644
--- a/tests/cntr/test_oneoff_runtime_state.py
+++ b/tests/cntr/test_oneoff_runtime_state.py
@@ -45,7 +45,8 @@ def process(*args, **kwargs):
         return SimpleNamespace(check_call=lambda: 0)
 
     manager = SimpleNamespace(data_path=tmp_path, project_name="test",
-                              runtime=SimpleNamespace(create_docker_process=process))
+                              runtime=SimpleNamespace(create_docker_process=process),
+                              image_preparer=SimpleNamespace(inspect_images=lambda refs: dict.fromkeys(refs, object())))
     runner = ComposeRunner(manager)
     runner._resolved_model = lambda process: {"services": {"app": {"image": "sha256:original"}}}
     context = OperationContext(project_containers=(), target_services=("app",),
diff --git a/tests/cntr/test_readiness_observation.py b/tests/cntr/test_readiness_observation.py
index 744f961c0c..7603142c4a 100644
--- a/tests/cntr/test_readiness_observation.py
+++ b/tests/cntr/test_readiness_observation.py
@@ -61,7 +61,7 @@ def test_started_service_does_not_wait_for_health(state, health):
     runner = ComposeRunner(SimpleNamespace(docker_inspector=SimpleNamespace(
         get_project_state=lambda containers, services=None: calls.append(services) or state)))
     context = SimpleNamespace(project_containers=(), compose_model={"services": {"app": {}}})
-    result = runner.verify_services_started(context, ("app",), {"app": "sha256:app"})
+    result = runner.verify_services_started(context, ("app",))
     assert result is state
     assert calls == [("app",)]
 
@@ -74,7 +74,7 @@ def test_unhealthy_service_does_not_block_start():
     runner = ComposeRunner(SimpleNamespace(docker_inspector=SimpleNamespace(
         get_project_state=lambda containers, services=None: calls.append(services) or state)))
     context = SimpleNamespace(project_containers=(), compose_model={"services": {"app": {}}})
-    assert runner.verify_services_started(context, ("app",), {"app": "sha256:app"}) is state
+    assert runner.verify_services_started(context, ("app",)) is state
     assert calls == [("app",)]
 
 
@@ -86,7 +86,7 @@ def test_successfully_exited_service_does_not_check_health(health):
     runner = ComposeRunner(SimpleNamespace(docker_inspector=SimpleNamespace(
         get_project_state=lambda containers, services=None: state)))
     context = SimpleNamespace(project_containers=(), compose_model={"services": {"job": {}}})
-    assert runner.verify_services_started(context, ("job",), {"job": "sha256:job"}) is state
+    assert runner.verify_services_started(context, ("job",)) is state
 
 
 @pytest.mark.parametrize(("condition", "state", "health", "exit_code"), [