Harden leoclean first-start readiness

ops/install_gcp_leoclean_nosend_release.py

@@ -28,6 +28,7 @@ if str(CONTRACT_ROOT) not in sys.path:
     sys.path.insert(0, str(CONTRACT_ROOT))
 
 from ops import gcp_leoclean_nosend_package as package  # noqa: E402
+from ops import leoclean_nosend_service_control as service_control  # noqa: E402
 
 INSTALL_RECEIPT_SCHEMA = "livingip.leocleanNoSendInstallReceipt.v2"
 ROLLBACK_STATE_SCHEMA = "livingip.leocleanNoSendRollbackState.v1"
@@ -142,6 +143,7 @@ class InstallPaths:
     receipt: Path
     rollback_state: Path
     lock: Path
+    cidfile: Path
     proc_root: Path
     dropin_directories: tuple[Path, ...]
 
@@ -163,6 +165,7 @@ class InstallPaths:
             receipt=state / "install-receipt.json",
             rollback_state=state / "rollback-state.json",
             lock=rooted("/run/leoclean-gcp-nosend-installer.lock"),
+            cidfile=rooted(package.SERVICE_CIDFILE),
             proc_root=proc_root or rooted("/proc"),
             dropin_directories=tuple(
                 rooted(path)
@@ -940,6 +943,36 @@ def _docker_inspect(arguments: list[str], runner: Runner, paths: InstallPaths, l
     return value[0]
 
 
+def _require_candidate_absent(
+    *,
+    runner: Runner,
+    paths: InstallPaths,
+    sleeper: Sleeper,
+) -> None:
+    for readback in range(2):
+        _require(
+            not paths.cidfile.exists() and not paths.cidfile.is_symlink(),
+            "candidate container ID file remained after stop",
+        )
+        reference = package.CONTAINER_NAME
+        completed = _run(
+            [*_docker_prefix(paths), "container", "inspect", reference],
+            runner,
+            label="stopped candidate inspection",
+        )
+        _require(completed.returncode != 0, "candidate container remained after stop")
+        try:
+            error = completed.stderr.decode("utf-8", "strict").strip()
+        except UnicodeError as exc:
+            raise InstallError("candidate container absence could not be proven") from exc
+        absence = re.compile(
+            rf"^(?:Error response from daemon: |Error: )?No such (?:container|object): {re.escape(reference)}$"
+        )
+        _require(absence.fullmatch(error) is not None, "candidate container absence could not be proven")
+        if readback == 0:
+            sleeper(0.05)
+
+
 def _validate_local_image(release: dict[str, Any], runner: Runner, paths: InstallPaths) -> None:
     reference = release["image"]["reference"]
     inspected = _docker_inspect(["image", "inspect", reference], runner, paths, "local image inspection")
@@ -951,7 +984,12 @@ def _validate_local_image(release: dict[str, Any], runner: Runner, paths: Instal
     _require(normalized == release["inspection"], "local image differed from the finalized inspection")
 
 
-def _validate_running_container(value: dict[str, Any], release: dict[str, Any]) -> dict[str, Any]:
+def _validate_starting_container(
+    value: dict[str, Any],
+    release: dict[str, Any],
+    *,
+    expected_container_id: str,
+) -> dict[str, Any] | None:
     config = value.get("Config")
     state = value.get("State")
     health = state.get("Health") if isinstance(state, dict) else None
@@ -959,9 +997,10 @@ def _validate_running_container(value: dict[str, Any], release: dict[str, Any])
     labels = config.get("Labels") if isinstance(config, dict) else None
     runtime = release["inspection"]["runtime_config"]
     container_id = value.get("Id")
-    _require(isinstance(container_id, str) and HEX_64.fullmatch(container_id) is not None, "container ID was invalid")
     _require(
-        value.get("Name") == f"/{package.CONTAINER_NAME}"
+        container_id == expected_container_id
+        and HEX_64.fullmatch(expected_container_id) is not None
+        and value.get("Name") == f"/{package.CONTAINER_NAME}"
         and value.get("Image") == release["image"]["config_digest"]
         and isinstance(config, dict)
         and config.get("Image") == release["image"]["reference"]
@@ -970,11 +1009,6 @@ def _validate_running_container(value: dict[str, Any], release: dict[str, Any])
         and labels.get("livingip.release.sha256") == release["release_sha256"]
         and labels.get("livingip.container.role") == "leoclean-gcp-nosend"
         and isinstance(state, dict)
-        and state.get("Running") is True
-        and isinstance(state.get("Pid"), int)
-        and state["Pid"] > 1
-        and isinstance(health, dict)
-        and health.get("Status") == "healthy"
         and isinstance(host, dict)
         and host.get("ReadonlyRootfs") is True
         and host.get("NetworkMode") == "bridge"
@@ -990,16 +1024,39 @@ def _validate_running_container(value: dict[str, Any], release: dict[str, Any])
         value.get("Path") == expected_path and value.get("Args") == expected_args,
         "running container command differed from the release contract",
     )
+    running = state.get("Running")
+    pid = state.get("Pid")
+    status = state.get("Status")
+    _require(isinstance(running, bool) and isinstance(pid, int), "container startup state was invalid")
+    if running is False:
+        _require(status == "created" and pid == 0, "container exited before becoming healthy")
+        return None
+    _require(pid > 1 and isinstance(health, dict), "container startup state was invalid")
+    health_status = health.get("Status")
+    _require(health_status in {"starting", "healthy"}, "container became unhealthy during startup")
+    if health_status == "starting":
+        return None
     return {"container_id": container_id, "container_pid": state["Pid"], "health": "healthy"}
 
 
+def _read_startup_container_id(paths: InstallPaths) -> str | None:
+    try:
+        record = service_control._read_cidfile(paths.cidfile)
+    except service_control.ControlError as exc:
+        raise InstallError("container ID file was unsafe or invalid") from exc
+    if record is None:
+        return None
+    container_id, _identity = record
+    return container_id
+
+
 def verify_running_release(
     release: dict[str, Any],
     *,
     runner: Runner,
     paths: InstallPaths,
     sleeper: Sleeper = time.sleep,
-    attempts: int = 30,
+    attempts: int = 240,
 ) -> dict[str, Any]:
     first: dict[str, Any] | None = None
     for _attempt in range(attempts):
@@ -1019,14 +1076,40 @@ def verify_running_release(
         release,
         str(first["ControlGroup"]),
     )
+    container: dict[str, Any] | None = None
+    for _attempt in range(attempts):
+        current = _service_state(runner)
+        _require(
+            current["MainPID"] == first["MainPID"]
+            and current["NRestarts"] == first["NRestarts"]
+            and current["InvocationID"] == first["InvocationID"]
+            and current["ControlGroup"] == first["ControlGroup"]
+            and current["ActiveState"] == "active"
+            and current["SubState"] == "running",
+            "service changed during startup verification",
+        )
+        container_id = _read_startup_container_id(paths)
+        if container_id is None:
+            sleeper(0.25)
+            continue
         first_container = _docker_inspect(
             ["container", "inspect", package.CONTAINER_NAME],
             runner,
             paths,
             "running container inspection",
         )
-    _require(first_container is not None, "running container was absent")
+        if first_container is None:
-    container = _validate_running_container(first_container, release)
+            sleeper(0.25)
+            continue
+        container = _validate_starting_container(
+            first_container,
+            release,
+            expected_container_id=container_id,
+        )
+        if container is not None:
+            break
+        sleeper(0.25)
+    _require(container is not None, "container did not become healthy within the bounded wait")
     _validate_local_image(release, runner, paths)
     second_container = _docker_inspect(
         ["container", "inspect", container["container_id"]],
@@ -1035,7 +1118,12 @@ def verify_running_release(
         "stable container inspection",
     )
     _require(second_container is not None, "running container exited during verification")
-    stable_container = _validate_running_container(second_container, release)
+    stable_container = _validate_starting_container(
+        second_container,
+        release,
+        expected_container_id=container["container_id"],
+    )
+    _require(stable_container is not None, "running container changed during verification")
     second = _service_state(runner)
     _validate_unit_configuration(second, paths, release)
     stable_supervising_process = _validate_supervising_process(
@@ -1044,6 +1132,7 @@ def verify_running_release(
         release,
         str(second["ControlGroup"]),
     )
+    stable_container_id = _read_startup_container_id(paths)
     _require(
         first["MainPID"] == second["MainPID"]
         and first["NRestarts"] == second["NRestarts"]
@@ -1052,6 +1141,7 @@ def verify_running_release(
         and first["ActiveState"] == second["ActiveState"] == "active"
         and first["SubState"] == second["SubState"] == "running"
         and container == stable_container
+        and stable_container_id == container["container_id"]
         and supervising_process == stable_supervising_process,
         "service or container changed during verification",
     )
@@ -1561,8 +1651,13 @@ def _install_release_transaction(
         if candidate_may_be_active:
             try:
                 _systemctl("stop", runner)
-            except InstallError:
+                _require_candidate_absent(
-                rollback_errors.append("candidate_stop")
+                    runner=runner,
+                    paths=paths,
+                    sleeper=sleeper,
+                )
+            except InstallError as stop_exc:
+                raise InstallError("installation failed and automatic rollback was incomplete") from stop_exc
         for path in (paths.receipt, paths.rollback_state, paths.release, paths.unit, paths.helper):
             if path not in attempted_paths:
                 continue

ops/leoclean_nosend_service_control.py

@@ -24,6 +24,7 @@ IMAGE_REPOSITORY = "europe-west6-docker.pkg.dev/teleo-501523/teleo/leoclean-nose
 SAFE_PATH = "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin"
 MAX_DOCKER_OUTPUT_BYTES = 1024 * 1024
 HEX_64 = re.compile(r"^[0-9a-f]{64}$")
+HEX_PARTIAL = re.compile(r"^[0-9a-f]{0,63}$")
 IMAGE_REFERENCE = re.compile(rf"^{re.escape(IMAGE_REPOSITORY)}@sha256:[0-9a-f]{{64}}$")
 IMAGE_CONFIG_DIGEST = re.compile(r"^sha256:[0-9a-f]{64}$")
 
@@ -160,7 +161,7 @@ def _cidfile_parent_descriptor(cidfile: Path) -> int:
     return descriptor
 
 
-def _read_cidfile(cidfile: Path) -> tuple[str, tuple[int, int]] | None:
+def _read_cidfile(cidfile: Path) -> tuple[str | None, tuple[int, int]] | None:
     parent_descriptor = _cidfile_parent_descriptor(cidfile)
     descriptor: int | None = None
     try:
@@ -187,8 +188,10 @@ def _read_cidfile(cidfile: Path) -> tuple[str, tuple[int, int]] | None:
             container_id = raw.decode("ascii", "strict").strip()
         except UnicodeError as exc:
             raise ControlError("container ID file was invalid") from exc
-        _require(HEX_64.fullmatch(container_id) is not None, "container ID file was invalid")
+        if HEX_64.fullmatch(container_id) is not None:
             return container_id, (observed.st_dev, observed.st_ino)
+        _require(HEX_PARTIAL.fullmatch(container_id) is not None, "container ID file was invalid")
+        return None, (observed.st_dev, observed.st_ino)
     except OSError as exc:
         raise ControlError("container ID file could not be read") from exc
     finally:
@@ -231,6 +234,52 @@ def _stop_exact(container_id: str, runner: Runner, sleeper: Callable[[float], No
     _wait_absent(container_id, runner, sleeper)
 
 
+def _remove_created_exact(
+    container_id: str,
+    *,
+    release_sha256: str,
+    image_reference: str,
+    image_config_digest: str,
+    runner: Runner,
+    sleeper: Callable[[float], None],
+) -> None:
+    for _attempt in range(4):
+        completed = _docker(["container", "rm", container_id], runner)
+        if completed.returncode == 0:
+            _wait_absent(container_id, runner, sleeper)
+            return
+        observed = _inspect(container_id, runner)
+        if observed is None:
+            return
+        running = _validate_container(
+            observed,
+            container_id=container_id,
+            release_sha256=release_sha256,
+            image_reference=image_reference,
+            image_config_digest=image_config_digest,
+        )
+        if running:
+            _stop_exact(container_id, runner, sleeper)
+            return
+        sleeper(0.25)
+    raise ControlError("created container removal failed")
+
+
+def _complete_cidfile(
+    cidfile: Path,
+    identity: tuple[int, int],
+    *,
+    sleeper: Callable[[float], None],
+) -> tuple[str, tuple[int, int]]:
+    for _attempt in range(40):
+        sleeper(0.25)
+        record = _read_cidfile(cidfile)
+        _require(record is not None and record[1] == identity, "container ID file changed during startup")
+        if record[0] is not None:
+            return record[0], record[1]
+    raise ControlError("container ID file did not become complete within the bounded wait")
+
+
 def control_container(
     action: str,
     release_sha256: str,
@@ -253,6 +302,13 @@ def control_container(
         return {"action": action, "status": "pass", "container": "absent"}
 
     container_id, cid_identity = cid
+    if container_id is None:
+        container_id, cid_identity = _complete_cidfile(
+            cidfile,
+            cid_identity,
+            sleeper=sleeper,
+        )
+        by_name = _inspect(CONTAINER_NAME, runner)
     by_id = _inspect(container_id, runner)
     if by_id is None:
         _require(by_name is None, "container name was rebound after the trusted container exited")
@@ -274,7 +330,18 @@ def control_container(
     if running:
         _stop_exact(container_id, runner, sleeper)
     else:
-        _wait_absent(container_id, runner, sleeper)
+        _require(
+            isinstance(by_id.get("State"), dict) and by_id["State"].get("Status") == "created",
+            "non-running container was not safely removable",
+        )
+        _remove_created_exact(
+            container_id,
+            release_sha256=release_sha256,
+            image_reference=image_reference,
+            image_config_digest=image_config_digest,
+            runner=runner,
+            sleeper=sleeper,
+        )
     _require_name_unbound(runner)
     _unlink_cidfile(cidfile, cid_identity)
     return {"action": action, "status": "pass", "container": "stopped"}

tests/test_gcp_leoclean_nosend_installer.py

@@ -189,7 +189,12 @@ class FakeHost:
                 "Env": runtime["environment"],
                 "Labels": labels,
             },
-            "State": {"Running": True, "Pid": 2000 + self.generation, "Health": {"Status": "healthy"}},
+            "State": {
+                "Status": "running",
+                "Running": True,
+                "Pid": 2000 + self.generation,
+                "Health": {"Status": "healthy"},
+            },
             "HostConfig": {
                 "ReadonlyRootfs": True,
                 "NetworkMode": "bridge",
@@ -199,6 +204,10 @@ class FakeHost:
                 "SecurityOpt": ["no-new-privileges:true"],
             },
         }
+        self.paths.cidfile.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+        self.paths.cidfile.parent.chmod(0o700)
+        self.paths.cidfile.write_text(container_id, encoding="ascii")
+        self.paths.cidfile.chmod(0o600)
         self.current_release = release
         self._write_process_environment()
 
@@ -209,6 +218,11 @@ class FakeHost:
         self.invocation = ""
         self.container = None
         self.current_release = None
+        self.paths.cidfile.unlink(missing_ok=True)
+        try:
+            self.paths.cidfile.parent.rmdir()
+        except FileNotFoundError:
+            pass
 
     def _show(self, arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
         self.show_count += 1
@@ -279,7 +293,8 @@ class FakeHost:
             return subprocess.CompletedProcess(arguments, 0, stdout=json.dumps([image]).encode(), stderr=b"")
         if operation[:2] == ["container", "inspect"]:
             if self.container is None or operation[2] not in {package.CONTAINER_NAME, self.container["Id"]}:
-                return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"not found")
+                error = f"Error: No such container: {operation[2]}\n".encode()
+                return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=error)
             observed = copy.deepcopy(self.container)
             release_sha = self.current_release.get("release_sha256") if self.current_release else None
             if operation[2] != package.CONTAINER_NAME and release_sha == self.race_release and not self.race_applied:
@@ -455,6 +470,136 @@ def test_first_install_accepts_production_shaped_systemd_not_found_defaults(tmp_
     assert host.restart_calls == 1
 
 
+def test_first_install_waits_for_complete_cid_and_healthy_container(tmp_path: Path) -> None:
+    bundle, release, image = release_bundle(tmp_path, "2")
+    paths = install_paths(tmp_path)
+    host = FakeHost(paths)
+    host.add_release(release, image)
+    container_inspections = 0
+    completed_cid = False
+
+    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
+        nonlocal container_inspections
+        result = host(arguments)
+        if arguments[:2] == [installer.SYSTEMCTL, "restart"]:
+            assert host.container is not None
+            paths.cidfile.write_text(str(host.container["Id"])[:12], encoding="ascii")
+            paths.cidfile.chmod(0o600)
+        operation = arguments[3:]
+        if operation == ["container", "inspect", package.CONTAINER_NAME] and result.returncode == 0:
+            container_inspections += 1
+            if container_inspections == 1:
+                observed = json.loads(result.stdout)
+                observed[0]["State"]["Health"]["Status"] = "starting"
+                return subprocess.CompletedProcess(
+                    arguments,
+                    0,
+                    stdout=json.dumps(observed).encode(),
+                    stderr=b"",
+                )
+        return result
+
+    def sleeper(_seconds: float) -> None:
+        nonlocal completed_cid
+        if host.active and not completed_cid:
+            assert host.container is not None
+            paths.cidfile.write_text(str(host.container["Id"]), encoding="ascii")
+            paths.cidfile.chmod(0o600)
+            completed_cid = True
+
+    receipt = installer.install_release(
+        bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=runner,
+        sleeper=sleeper,
+        execute_restart=True,
+    )
+
+    assert receipt["status"] == "pass"
+    assert completed_cid
+    assert container_inspections == 2
+    assert host.stop_calls == 0
+
+
+def test_first_install_health_timeout_rolls_back_to_absent(tmp_path: Path) -> None:
+    bundle, release, image = release_bundle(tmp_path, "2")
+    paths = install_paths(tmp_path)
+    host = FakeHost(paths)
+    host.add_release(release, image)
+
+    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
+        result = host(arguments)
+        if arguments[3:] == ["container", "inspect", package.CONTAINER_NAME] and result.returncode == 0:
+            observed = json.loads(result.stdout)
+            observed[0]["State"]["Health"]["Status"] = "starting"
+            return subprocess.CompletedProcess(
+                arguments,
+                0,
+                stdout=json.dumps(observed).encode(),
+                stderr=b"",
+            )
+        return result
+
+    with pytest.raises(installer.InstallError, match="did not become healthy"):
+        installer.install_release(
+            bundle,
+            paths=paths,
+            source_binding=source_binding(),
+            runner=runner,
+            sleeper=lambda _seconds: None,
+            execute_restart=True,
+        )
+
+    assert host.stop_calls == 1
+    assert not host.active
+    assert host.container is None
+    assert not paths.cidfile.exists()
+    assert not paths.unit.exists()
+    assert not paths.helper.exists()
+    assert not paths.state_dir.exists()
+
+
+def test_verification_rebinds_stable_container_to_unchanged_cidfile(tmp_path: Path) -> None:
+    bundle, release, image = release_bundle(tmp_path, "2")
+    paths = install_paths(tmp_path)
+    host = FakeHost(paths)
+    host.add_release(release, image)
+    installer.install_release(
+        bundle,
+        paths=paths,
+        source_binding=source_binding(),
+        runner=host,
+        sleeper=lambda _seconds: None,
+        execute_restart=True,
+    )
+    changed = False
+
+    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
+        nonlocal changed
+        result = host(arguments)
+        operation = arguments[3:]
+        if (
+            operation[:2] == ["container", "inspect"]
+            and operation[2] != package.CONTAINER_NAME
+            and not changed
+        ):
+            paths.cidfile.write_text("f" * 64, encoding="ascii")
+            paths.cidfile.chmod(0o600)
+            changed = True
+        return result
+
+    with pytest.raises(installer.InstallError, match="changed during verification"):
+        installer.verify_running_release(
+            release,
+            runner=runner,
+            paths=paths,
+            sleeper=lambda _seconds: None,
+        )
+
+    assert changed
+
+
 def test_systemd_252_omitted_environment_files_property_is_normalized(tmp_path: Path) -> None:
     bundle, release, image = release_bundle(tmp_path, "2")
     paths = install_paths(tmp_path)
@@ -992,6 +1137,67 @@ def test_restart_command_failure_after_side_effect_stops_candidate_and_cleans_in
     assert not paths.state_dir.exists()
 
 
+def test_candidate_stop_failure_preserves_recovery_artifacts(tmp_path: Path) -> None:
+    bundle, release, image = release_bundle(tmp_path, "2")
+    paths = install_paths(tmp_path)
+    host = FakeHost(paths)
+    host.add_release(release, image)
+    host.race_release = str(release["release_sha256"])
+    host.race_kind = "service_restart"
+
+    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
+        if arguments[:2] == [installer.SYSTEMCTL, "stop"]:
+            return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"stop failed")
+        return host(arguments)
+
+    with pytest.raises(installer.InstallError, match="automatic rollback was incomplete"):
+        installer.install_release(
+            bundle,
+            paths=paths,
+            source_binding=source_binding(),
+            runner=runner,
+            sleeper=lambda _seconds: None,
+            execute_restart=True,
+        )
+
+    assert paths.unit.is_file()
+    assert paths.helper.is_file()
+    assert paths.state_dir.is_dir()
+    assert host.active
+    assert host.container is not None
+
+
+def test_successful_stop_command_with_candidate_residue_preserves_recovery_artifacts(tmp_path: Path) -> None:
+    bundle, release, image = release_bundle(tmp_path, "2")
+    paths = install_paths(tmp_path)
+    host = FakeHost(paths)
+    host.add_release(release, image)
+    host.race_release = str(release["release_sha256"])
+    host.race_kind = "service_restart"
+
+    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
+        if arguments[:2] == [installer.SYSTEMCTL, "stop"]:
+            return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
+        return host(arguments)
+
+    with pytest.raises(installer.InstallError, match="automatic rollback was incomplete"):
+        installer.install_release(
+            bundle,
+            paths=paths,
+            source_binding=source_binding(),
+            runner=runner,
+            sleeper=lambda _seconds: None,
+            execute_restart=True,
+        )
+
+    assert paths.unit.is_file()
+    assert paths.helper.is_file()
+    assert paths.state_dir.is_dir()
+    assert paths.cidfile.is_file()
+    assert host.active
+    assert host.container is not None
+
+
 def test_later_dropin_failure_restores_previous_release_and_running_service(tmp_path: Path) -> None:
     first_bundle, first_release, first_image = release_bundle(tmp_path, "2")
     second_bundle, second_release, second_image = release_bundle(tmp_path, "4")
@@ -1481,6 +1687,187 @@ def test_control_helper_stops_only_trusted_full_container_id(tmp_path: Path) ->
     assert stop_calls == [["container", "stop", "--time=30", container_id]]
 
 
+@pytest.mark.parametrize("running", [True, False])
+def test_control_helper_waits_for_complete_cid_before_exact_cleanup(
+    tmp_path: Path,
+    running: bool,
+) -> None:
+    cid_parent = tmp_path / "runtime"
+    cid_parent.mkdir(mode=0o700)
+    cidfile = cid_parent / "container.cid"
+    container_id = "c" * 64
+    cidfile.write_text(container_id[:12], encoding="ascii")
+    cidfile.chmod(0o600)
+    release_sha = "d" * 64
+    image_reference = f"{package.IMAGE_REPOSITORY}@sha256:{'e' * 64}"
+    pending_container: dict[str, object] = {
+        "Id": container_id,
+        "Name": f"/{package.CONTAINER_NAME}",
+        "Image": "sha256:" + "f" * 64,
+        "Config": {
+            "Image": image_reference,
+            "Labels": {
+                "livingip.release.sha256": release_sha,
+                "livingip.container.role": "leoclean-gcp-nosend",
+            },
+        },
+        "State": {
+            "Status": "running" if running else "created",
+            "Running": running,
+        },
+    }
+    container: dict[str, object] | None = None
+    calls: list[list[str]] = []
+
+    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
+        nonlocal container
+        calls.append(arguments)
+        operation = arguments[3:]
+        if operation[:2] == ["container", "inspect"]:
+            if container is None:
+                error = f"Error: No such container: {operation[2]}\n".encode()
+                return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=error)
+            return subprocess.CompletedProcess(arguments, 0, stdout=json.dumps([container]).encode(), stderr=b"")
+        expected_mutation = (
+            ["container", "stop", "--time=30", container_id]
+            if running
+            else ["container", "rm", container_id]
+        )
+        if operation == expected_mutation:
+            container = None
+            return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
+        raise AssertionError(arguments)
+
+    completed = False
+
+    def sleeper(_seconds: float) -> None:
+        nonlocal completed, container
+        if not completed:
+            cidfile.write_text(container_id, encoding="ascii")
+            cidfile.chmod(0o600)
+            container = pending_container
+            completed = True
+
+    result = control.control_container(
+        "stop",
+        release_sha,
+        image_reference,
+        "sha256:" + "f" * 64,
+        runner=runner,
+        sleeper=sleeper,
+        cidfile=cidfile,
+    )
+
+    assert result == {"action": "stop", "status": "pass", "container": "stopped"}
+    assert completed
+    assert not cidfile.exists()
+    mutations = [
+        call[3:]
+        for call in calls
+        if call[3:5] in (["container", "stop"], ["container", "rm"])
+    ]
+    assert mutations == [
+        ["container", "stop", "--time=30", container_id]
+        if running
+        else ["container", "rm", container_id]
+    ]
+
+
+def test_control_helper_reconciles_created_to_running_race_by_full_id(tmp_path: Path) -> None:
+    cid_parent = tmp_path / "runtime"
+    cid_parent.mkdir(mode=0o700)
+    cidfile = cid_parent / "container.cid"
+    container_id = "c" * 64
+    cidfile.write_text(container_id, encoding="ascii")
+    cidfile.chmod(0o600)
+    release_sha = "d" * 64
+    image_reference = f"{package.IMAGE_REPOSITORY}@sha256:{'e' * 64}"
+    container: dict[str, object] | None = {
+        "Id": container_id,
+        "Name": f"/{package.CONTAINER_NAME}",
+        "Image": "sha256:" + "f" * 64,
+        "Config": {
+            "Image": image_reference,
+            "Labels": {
+                "livingip.release.sha256": release_sha,
+                "livingip.container.role": "leoclean-gcp-nosend",
+            },
+        },
+        "State": {"Status": "created", "Running": False},
+    }
+    calls: list[list[str]] = []
+
+    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
+        nonlocal container
+        calls.append(arguments)
+        operation = arguments[3:]
+        if operation[:2] == ["container", "inspect"]:
+            if container is None:
+                error = f"Error: No such container: {operation[2]}\n".encode()
+                return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=error)
+            return subprocess.CompletedProcess(arguments, 0, stdout=json.dumps([container]).encode(), stderr=b"")
+        if operation == ["container", "rm", container_id]:
+            assert container is not None
+            container["State"] = {"Status": "running", "Running": True}
+            return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=b"started during removal")
+        if operation == ["container", "stop", "--time=30", container_id]:
+            container = None
+            return subprocess.CompletedProcess(arguments, 0, stdout=b"", stderr=b"")
+        raise AssertionError(arguments)
+
+    result = control.control_container(
+        "stop",
+        release_sha,
+        image_reference,
+        "sha256:" + "f" * 64,
+        runner=runner,
+        sleeper=lambda _seconds: None,
+        cidfile=cidfile,
+    )
+
+    assert result == {"action": "stop", "status": "pass", "container": "stopped"}
+    assert not cidfile.exists()
+    mutations = [
+        call[3:]
+        for call in calls
+        if call[3:5] in (["container", "stop"], ["container", "rm"])
+    ]
+    assert mutations == [
+        ["container", "rm", container_id],
+        ["container", "stop", "--time=30", container_id],
+    ]
+
+
+def test_control_helper_partial_cid_timeout_retains_authority_and_does_not_mutate(tmp_path: Path) -> None:
+    cid_parent = tmp_path / "runtime"
+    cid_parent.mkdir(mode=0o700)
+    cidfile = cid_parent / "container.cid"
+    cidfile.write_text("c" * 12, encoding="ascii")
+    cidfile.chmod(0o600)
+    calls: list[list[str]] = []
+
+    def runner(arguments: list[str]) -> subprocess.CompletedProcess[bytes]:
+        calls.append(arguments)
+        operation = arguments[3:]
+        assert operation == ["container", "inspect", package.CONTAINER_NAME]
+        error = f"Error: No such container: {package.CONTAINER_NAME}\n".encode()
+        return subprocess.CompletedProcess(arguments, 1, stdout=b"", stderr=error)
+
+    with pytest.raises(control.ControlError, match="did not become complete"):
+        control.control_container(
+            "stop",
+            "d" * 64,
+            f"{package.IMAGE_REPOSITORY}@sha256:{'e' * 64}",
+            "sha256:" + "f" * 64,
+            runner=runner,
+            sleeper=lambda _seconds: None,
+            cidfile=cidfile,
+        )
+
+    assert cidfile.read_text(encoding="ascii") == "c" * 12
+    assert all(call[3:5] not in (["container", "stop"], ["container", "rm"]) for call in calls)
+
+
 def test_control_helper_refuses_config_digest_mismatch_without_stopping(tmp_path: Path) -> None:
     cid_parent = tmp_path / "runtime"
     cid_parent.mkdir(mode=0o700)