fix(cntr): trust Compose image convergence

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

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)

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(

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 = []

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)