fix(init): guard persistent task startup (#616)

## Description

Prevent `headroom init` hooks from spawning duplicate persistent-task
runners while a proxy is still starting.

Fixes #615

## Type of Change

- [x] Bug fix (non-breaking change that fixes an issue)
- [ ] New feature (non-breaking change that adds functionality)
- [ ] Breaking change (fix or feature that would cause existing
functionality to change)
- [ ] Documentation update
- [ ] Performance improvement
- [ ] Code refactoring (no functional changes)

## Problem

`_ensure_profile_running()` checked readiness for only one second and
then launched `start_detached_agent()` whenever the proxy was not ready
yet. When Claude/Codex hooks fired close together, each hook could race
through that path and spawn another detached persistent-task runner.

## Changes Made

- Add a profile-local, nonblocking runtime start lock around init hook
startup.
- Re-check readiness after acquiring the lock so late-arriving hooks do
not start a duplicate runner.
- If a runtime is already alive, wait up to 15 seconds for readiness
before stopping and restarting it.
- Add regression tests for lock contention, slow startup, and
cross-process lock behavior.

## Testing

- [x] Unit tests pass (`pytest`)
- [x] Linting passes (`ruff check .`)
- [x] Type checking passes (`mypy headroom`)
- [x] New tests added for new functionality
- [x] Manual testing performed

## Test Output

```
UV_SKIP_WHEEL_FILENAME_CHECK=1 uv run --frozen --extra dev --extra proxy pytest tests/test_cli/test_init_cli.py tests/test_cli/test_install_cli.py tests/test_cli/test_wrap_persistent.py tests/test_install/test_runtime.py
# 89 passed in 0.61s

UV_SKIP_WHEEL_FILENAME_CHECK=1 uv run --frozen --extra dev --extra proxy ruff check .
# All checks passed!

UV_SKIP_WHEEL_FILENAME_CHECK=1 uv run --frozen --extra dev --extra proxy ruff format --check .
# 775 files already formatted

UV_SKIP_WHEEL_FILENAME_CHECK=1 uv run --frozen --extra dev --extra proxy mypy headroom --ignore-missing-imports
# Success: no issues found in 346 source files
```

Manual sandbox check:

```
# before this change: 3 ensure calls spawned 3 detached starts
# after this change: 3 ensure calls spawned 1 detached start while the runtime was still starting
```

## Checklist

- [x] My code follows the project's style guidelines
- [x] I have performed a self-review of my code
- [x] I have commented my code, particularly in hard-to-understand areas
- [ ] I have made corresponding changes to the documentation
- [x] My changes generate no new warnings
- [x] I have added tests that prove my fix is effective or that my
feature works
- [x] New and existing unit tests pass locally with my changes
- [ ] I have updated the CHANGELOG.md if applicable

## Additional Notes

Docs and CHANGELOG were left unchanged because this is a small runtime
bug fix with no user-facing CLI/API change.

headroom/cli/init.py

@@ -20,7 +20,9 @@ from headroom.install.paths import claude_settings_path, codex_config_path, vali
 from headroom.install.planner import build_manifest
 from headroom.install.providers import _apply_unix_env_scope, _apply_windows_env_scope
 from headroom.install.runtime import (
+    acquire_runtime_start_lock,
     resolve_headroom_command,
+    runtime_status,
     start_detached_agent,
     start_persistent_docker,
     stop_runtime,
@@ -46,6 +48,7 @@ _CODEX_FEATURE_MARKER_END = "# --- end Headroom init features ---"
 _SUPPORTED_TARGETS = ("claude", "copilot", "codex", "openclaw")
 _LOCAL_TARGETS = {"claude", "codex"}
 _GLOBAL_TARGETS = {"claude", "copilot", "codex", "openclaw"}
+_STARTUP_READY_TIMEOUT_SECONDS = 15
 
 
 def _command_string(parts: list[str]) -> str:
@@ -553,13 +556,22 @@ def _ensure_profile_running(profile: str) -> None:
     if wait_ready(manifest, timeout_seconds=1):
         return
     try:
-        if manifest.preset == InstallPreset.PERSISTENT_DOCKER.value:
-            start_persistent_docker(manifest)
-        elif manifest.supervisor_kind == SupervisorKind.SERVICE.value:
-            start_supervisor(manifest)
-        else:
-            start_detached_agent(manifest.profile)
-        wait_ready(manifest, timeout_seconds=45)
+        with acquire_runtime_start_lock(manifest.profile) as acquired:
+            if not acquired:
+                return
+            if wait_ready(manifest, timeout_seconds=1):
+                return
+            if runtime_status(manifest) == "running":
+                if wait_ready(manifest, timeout_seconds=_STARTUP_READY_TIMEOUT_SECONDS):
+                    return
+                stop_runtime(manifest)
+            if manifest.preset == InstallPreset.PERSISTENT_DOCKER.value:
+                start_persistent_docker(manifest)
+            elif manifest.supervisor_kind == SupervisorKind.SERVICE.value:
+                start_supervisor(manifest)
+            else:
+                start_detached_agent(manifest.profile)
+            wait_ready(manifest, timeout_seconds=45)
     except Exception:
         return
 

headroom/install/runtime.py

@@ -8,12 +8,14 @@ import signal
 import subprocess
 import sys
 import time
+from collections.abc import Iterator
+from contextlib import contextmanager
 from pathlib import Path
-from typing import Any
+from typing import Any, cast
 
 from .health import probe_ready
 from .models import DeploymentManifest, InstallPreset, RuntimeKind
-from .paths import log_path, pid_path
+from .paths import log_path, pid_path, profile_root
 
 PASSTHROUGH_ENV_PREFIXES = (
     "HEADROOM_",
@@ -165,6 +167,57 @@ def _clear_pid(profile: str) -> None:
         path.unlink()
 
 
+@contextmanager
+def acquire_runtime_start_lock(profile: str) -> Iterator[bool]:
+    """Try to hold the profile-local runtime start lock."""
+
+    path = profile_root(profile) / "runner.start.lock"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with open(path, "a+", encoding="utf-8", errors="replace") as lock_file:
+        acquired = False
+        if _is_windows():
+            import msvcrt
+
+            lock_file.seek(0)
+            msvcrt_any = cast(Any, msvcrt)
+            try:
+                msvcrt_any.locking(lock_file.fileno(), msvcrt_any.LK_NBLCK, 1)
+                acquired = True
+            except OSError:
+                yield False
+                return
+        else:
+            import fcntl
+
+            try:
+                fcntl.flock(lock_file.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
+                acquired = True
+            except BlockingIOError:
+                yield False
+                return
+        try:
+            lock_file.seek(0)
+            lock_file.truncate()
+            lock_file.write(str(os.getpid()))
+            lock_file.flush()
+            yield True
+        finally:
+            if acquired:
+                if _is_windows():
+                    import msvcrt
+
+                    lock_file.seek(0)
+                    msvcrt_any = cast(Any, msvcrt)
+                    try:
+                        msvcrt_any.locking(lock_file.fileno(), msvcrt_any.LK_UNLCK, 1)
+                    except OSError:
+                        pass
+                else:
+                    import fcntl
+
+                    fcntl.flock(lock_file.fileno(), fcntl.LOCK_UN)
+
+
 def run_foreground(manifest: DeploymentManifest) -> int:
     """Run the raw runtime command in the foreground."""
 

tests/test_cli/test_init_cli.py

@@ -4,6 +4,7 @@ import importlib
 import json
 import sys
 import types
+from contextlib import contextmanager
 from pathlib import Path
 from types import SimpleNamespace
 
@@ -780,6 +781,13 @@ def test_ensure_profile_running_covers_runtime_modes(monkeypatch) -> None:
     wait_calls: list[tuple[str, int]] = []
 
     monkeypatch.setattr(init_cli, "load_manifest", lambda profile: manifests.get(profile))
+    monkeypatch.setattr(init_cli, "runtime_status", lambda manifest: "stopped")
+
+    @contextmanager
+    def fake_start_lock(profile: str):
+        yield True
+
+    monkeypatch.setattr(init_cli, "acquire_runtime_start_lock", fake_start_lock)
 
     def fake_wait_ready(manifest, timeout_seconds: int) -> bool:
         wait_calls.append((manifest.profile, timeout_seconds))
@@ -829,6 +837,12 @@ def test_ensure_profile_running_returns_when_ready_or_on_exception(monkeypatch)
     init_cli._ensure_profile_running("task-profile")
     assert detached_calls == []
 
+    @contextmanager
+    def fake_start_lock(profile: str):
+        yield True
+
+    monkeypatch.setattr(init_cli, "acquire_runtime_start_lock", fake_start_lock)
+    monkeypatch.setattr(init_cli, "runtime_status", lambda manifest: "stopped")
     monkeypatch.setattr(init_cli, "wait_ready", lambda manifest, timeout_seconds: False)
     monkeypatch.setattr(
         init_cli,
@@ -838,6 +852,75 @@ def test_ensure_profile_running_returns_when_ready_or_on_exception(monkeypatch)
     init_cli._ensure_profile_running("task-profile")
 
 
+def test_ensure_profile_running_skips_spawn_when_start_lock_is_held(monkeypatch) -> None:
+    init_cli, _ = _load_init_module(monkeypatch)
+    manifest = SimpleNamespace(
+        preset=init_cli.InstallPreset.PERSISTENT_TASK.value,
+        supervisor_kind=init_cli.SupervisorKind.NONE.value,
+        profile="task-profile",
+    )
+    detached_calls: list[str] = []
+
+    @contextmanager
+    def fake_start_lock(profile: str):
+        yield False
+
+    monkeypatch.setattr(init_cli, "load_manifest", lambda profile: manifest)
+    monkeypatch.setattr(init_cli, "wait_ready", lambda manifest, timeout_seconds: False)
+    monkeypatch.setattr(init_cli, "acquire_runtime_start_lock", fake_start_lock)
+    monkeypatch.setattr(
+        init_cli,
+        "start_detached_agent",
+        lambda profile: detached_calls.append(profile),
+    )
+
+    init_cli._ensure_profile_running("task-profile")
+
+    assert detached_calls == []
+
+
+def test_ensure_profile_running_does_not_spawn_again_during_slow_startup(monkeypatch) -> None:
+    init_cli, _ = _load_init_module(monkeypatch)
+    manifest = SimpleNamespace(
+        preset=init_cli.InstallPreset.PERSISTENT_TASK.value,
+        supervisor_kind=init_cli.SupervisorKind.NONE.value,
+        profile="task-profile",
+    )
+    detached_calls: list[str] = []
+    wait_calls: list[int] = []
+    stop_calls: list[object] = []
+
+    @contextmanager
+    def fake_start_lock(profile: str):
+        yield True
+
+    def fake_wait_ready(manifest, timeout_seconds: int) -> bool:
+        wait_calls.append(timeout_seconds)
+        return bool(detached_calls and timeout_seconds == init_cli._STARTUP_READY_TIMEOUT_SECONDS)
+
+    monkeypatch.setattr(init_cli, "load_manifest", lambda profile: manifest)
+    monkeypatch.setattr(init_cli, "wait_ready", fake_wait_ready)
+    monkeypatch.setattr(init_cli, "acquire_runtime_start_lock", fake_start_lock)
+    monkeypatch.setattr(
+        init_cli,
+        "runtime_status",
+        lambda manifest: "running" if detached_calls else "stopped",
+    )
+    monkeypatch.setattr(
+        init_cli,
+        "start_detached_agent",
+        lambda profile: detached_calls.append(profile),
+    )
+    monkeypatch.setattr(init_cli, "stop_runtime", lambda manifest: stop_calls.append(manifest))
+
+    init_cli._ensure_profile_running("task-profile")
+    init_cli._ensure_profile_running("task-profile")
+
+    assert detached_calls == ["task-profile"]
+    assert init_cli._STARTUP_READY_TIMEOUT_SECONDS in wait_calls
+    assert stop_calls == []
+
+
 def test_init_codex_windows_warns_about_upstream_hook_limitation(monkeypatch) -> None:
     init_cli, _ = _load_init_module(monkeypatch)
     messages: list[str] = []

tests/test_install/test_runtime.py

@@ -1,6 +1,9 @@
 from __future__ import annotations
 
+import os
 import signal
+import subprocess
+import sys
 from pathlib import Path
 
 from headroom.install.models import DeploymentManifest, InstallPreset
@@ -11,6 +14,7 @@ from headroom.install.runtime import (
     _read_pid,
     _runtime_env,
     _write_pid,
+    acquire_runtime_start_lock,
     build_runtime_command,
     resolve_headroom_command,
     run_foreground,
@@ -206,6 +210,40 @@ def test_write_read_and_clear_pid(monkeypatch, tmp_path: Path) -> None:
     assert _read_pid("default") is None
 
 
+def test_runtime_start_lock_is_nonblocking(monkeypatch, tmp_path: Path) -> None:
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+
+    with acquire_runtime_start_lock("default") as first_acquired:
+        assert first_acquired is True
+        with acquire_runtime_start_lock("default") as second_acquired:
+            assert second_acquired is False
+
+    with acquire_runtime_start_lock("default") as acquired_after_release:
+        assert acquired_after_release is True
+
+
+def test_runtime_start_lock_blocks_another_process(monkeypatch, tmp_path: Path) -> None:
+    monkeypatch.setattr(Path, "home", lambda: tmp_path)
+    script = (
+        "from headroom.install.runtime import acquire_runtime_start_lock\n"
+        "with acquire_runtime_start_lock('default') as acquired:\n"
+        "    print(acquired)\n"
+    )
+    env = {**os.environ, "HOME": str(tmp_path), "PYTHONPATH": str(Path.cwd())}
+
+    with acquire_runtime_start_lock("default") as acquired:
+        assert acquired is True
+        result = subprocess.run(
+            [sys.executable, "-c", script],
+            capture_output=True,
+            check=True,
+            env=env,
+            text=True,
+        )
+
+    assert result.stdout.strip() == "False"
+
+
 def test_run_foreground_and_detached_helpers(monkeypatch, tmp_path: Path) -> None:
     monkeypatch.setattr(Path, "home", lambda: tmp_path)
     monkeypatch.setattr(