diff --git a/apps/device-host-agent/host_agent/app.py b/apps/device-host-agent/host_agent/app.py index f7990d3..a571139 100644 --- a/apps/device-host-agent/host_agent/app.py +++ b/apps/device-host-agent/host_agent/app.py @@ -18,6 +18,7 @@ from host_agent.execution import create_execution_factories from host_agent.heartbeat import HeartbeatSynchronizer from host_agent.history import ConsoleHistoryStore from host_agent.identity import HostIdentityStore +from host_agent.instance_lock import InstanceLock from host_agent.lease import ActiveAssignmentRunner from host_agent.local_account import LocalAccountStore from host_agent.policy_cache import HostPolicyCacheStore @@ -40,6 +41,7 @@ class HostAgentApplication: console_server: uvicorn.Server | None = None console_enrollment_client: HostAgentEnrollmentClient | None = None dependency_supervisor: DependencySupervisor | None = None + instance_lock: InstanceLock | None = None def run(self) -> None: asyncio.run(self.run_async()) @@ -109,6 +111,8 @@ class HostAgentApplication: if self.console_enrollment_client is not None: self.console_enrollment_client.close() await self.client.aclose() + if self.instance_lock is not None: + self.instance_lock.release() async def _claim_until_stopped( self, @@ -151,114 +155,125 @@ def create_application( enrollment_client: HostAgentEnrollmentClient | None = None, ) -> HostAgentApplication: startup_config = config or load_host_agent_config() - config_store = device_config_store or DeviceConfigStore() - resolved_identity_store = identity_store or HostIdentityStore( - startup_config.identity_path - ) - owned_enrollment_client = enrollment_client is None - bootstrap_client = enrollment_client or HostAgentEnrollmentClient(startup_config) + instance_lock = InstanceLock(startup_config.identity_path.parent) + instance_lock.acquire() try: - resolved_config = resolve_host_identity( - startup_config, - identity_store=resolved_identity_store, - client=bootstrap_client, + config_store = device_config_store or DeviceConfigStore() + resolved_identity_store = identity_store or HostIdentityStore( + startup_config.identity_path ) - bootstrap_client.config = resolved_config - resolved_manager = manager or _configured_device_manager( - config_store, - config=resolved_config, - enrollment_client=bootstrap_client, + owned_enrollment_client = enrollment_client is None + bootstrap_client = enrollment_client or HostAgentEnrollmentClient( + startup_config ) - finally: - if owned_enrollment_client: - bootstrap_client.close() - client = HostAgentClient(resolved_config) + try: + resolved_config = resolve_host_identity( + startup_config, + identity_store=resolved_identity_store, + client=bootstrap_client, + ) + bootstrap_client.config = resolved_config + resolved_manager = manager or _configured_device_manager( + config_store, + config=resolved_config, + enrollment_client=bootstrap_client, + ) + finally: + if owned_enrollment_client: + bootstrap_client.close() + client = HostAgentClient(resolved_config) - history_store = ConsoleHistoryStore( - resolved_config.identity_path.parent / "host_console_history.sqlite3", - limit=resolved_config.console_history_limit, - ) - status_tracker = AgentStatusTracker() - console_enrollment_client: HostAgentEnrollmentClient | None = None - if resolved_config.enrollment_managed: - console_enrollment_client = HostAgentEnrollmentClient(resolved_config) - metadata_store = TaskMetadataStore(db_path=resolved_config.task_progress_db_path) - timeline = Timeline(ArtifactStore(root=resolved_config.task_artifact_dir)) - executor = AssignmentExecutor( - create_execution_factories( - resolved_manager, + history_store = ConsoleHistoryStore( + resolved_config.identity_path.parent / "host_console_history.sqlite3", + limit=resolved_config.console_history_limit, + ) + status_tracker = AgentStatusTracker() + console_enrollment_client: HostAgentEnrollmentClient | None = None + if resolved_config.enrollment_managed: + console_enrollment_client = HostAgentEnrollmentClient(resolved_config) + metadata_store = TaskMetadataStore( + db_path=resolved_config.task_progress_db_path + ) + timeline = Timeline(ArtifactStore(root=resolved_config.task_artifact_dir)) + executor = AssignmentExecutor( + create_execution_factories( + resolved_manager, + metadata_store=metadata_store, + timeline=timeline, + host_agent_config=resolved_config, + ) + ) + console_app = create_console_app( + config=resolved_config, + manager=resolved_manager, + config_store=config_store, + local_account_store=LocalAccountStore(resolved_config.local_account_path), + identity_store=resolved_identity_store, + history_store=history_store, + status_tracker=status_tracker, + session_manager=SessionManager( + ttl_seconds=resolved_config.console_session_ttl_seconds + ), + enrollment_client=console_enrollment_client, metadata_store=metadata_store, timeline=timeline, - host_agent_config=resolved_config, + executor=executor, ) - ) - console_app = create_console_app( - config=resolved_config, - manager=resolved_manager, - config_store=config_store, - local_account_store=LocalAccountStore(resolved_config.local_account_path), - identity_store=resolved_identity_store, - history_store=history_store, - status_tracker=status_tracker, - session_manager=SessionManager( - ttl_seconds=resolved_config.console_session_ttl_seconds - ), - enrollment_client=console_enrollment_client, - metadata_store=metadata_store, - timeline=timeline, - executor=executor, - ) - console_server = _EmbeddedConsoleServer( - uvicorn.Config( - console_app, - host=resolved_config.console_bind_host, - port=resolved_config.console_port, - log_level="warning", + console_server = _EmbeddedConsoleServer( + uvicorn.Config( + console_app, + host=resolved_config.console_bind_host, + port=resolved_config.console_port, + log_level="warning", + ) ) - ) - heartbeat = HeartbeatSynchronizer( - resolved_manager, - client, - resolved_config, - status_tracker=status_tracker, - on_sync=lambda device_count: history_store.record_heartbeat( - device_count=device_count - ), - policy_cache=HostPolicyCacheStore( - resolved_config.identity_path.parent / "host_governance_policy.json" - ), - on_policy_sync=lambda revision: history_store.record_policy_sync( - revision=revision - ), - ) - active_runner = ActiveAssignmentRunner(client, executor) - processor = AssignmentProcessor( - client, - active_runner, - status_tracker=status_tracker, - on_result=lambda assignment, result: _on_assignment_finished( - history_store, - metadata_store, - timeline, + heartbeat = HeartbeatSynchronizer( + resolved_manager, + client, resolved_config, - assignment, - result, - ), - ) - dependency_supervisor: DependencySupervisor | None = None - if resolved_config.dependency_supervisor_enabled: - dependency_supervisor = DependencySupervisor.from_host_agent_config( - resolved_config + status_tracker=status_tracker, + on_sync=lambda device_count: history_store.record_heartbeat( + device_count=device_count + ), + policy_cache=HostPolicyCacheStore( + resolved_config.identity_path.parent / "host_governance_policy.json" + ), + on_policy_sync=lambda revision: history_store.record_policy_sync( + revision=revision + ), ) - return HostAgentApplication( - client=client, - heartbeat=heartbeat, - processor=processor, - console_server=console_server, - console_enrollment_client=console_enrollment_client, - dependency_supervisor=dependency_supervisor, - ) + active_runner = ActiveAssignmentRunner(client, executor) + processor = AssignmentProcessor( + client, + active_runner, + status_tracker=status_tracker, + on_result=lambda assignment, result: _on_assignment_finished( + history_store, + metadata_store, + timeline, + resolved_config, + assignment, + result, + ), + ) + dependency_supervisor: DependencySupervisor | None = None + if resolved_config.dependency_supervisor_enabled: + dependency_supervisor = DependencySupervisor.from_host_agent_config( + resolved_config + ) + return HostAgentApplication( + client=client, + heartbeat=heartbeat, + processor=processor, + console_server=console_server, + console_enrollment_client=console_enrollment_client, + dependency_supervisor=dependency_supervisor, + instance_lock=instance_lock, + ) + except BaseException: + instance_lock.release() + raise def _on_assignment_finished( diff --git a/apps/device-host-agent/host_agent/cli.py b/apps/device-host-agent/host_agent/cli.py index 9906853..4fba57b 100644 --- a/apps/device-host-agent/host_agent/cli.py +++ b/apps/device-host-agent/host_agent/cli.py @@ -8,6 +8,7 @@ from dataclasses import replace from host_agent.app import create_application from host_agent.config import load_host_agent_config +from host_agent.instance_lock import InstanceAlreadyRunningError from host_agent.local_account import LocalAccountStore @@ -30,16 +31,22 @@ def main(argv: Sequence[str] | None = None) -> None: print(f"error: {exc}", file=sys.stderr) raise SystemExit(1) from exc - create_application(config=config).run() + try: + create_application(config=config).run() + except InstanceAlreadyRunningError as exc: + print(f"error: {exc}", file=sys.stderr) + raise SystemExit(1) from exc def _run_setup() -> None: config = load_host_agent_config() store = LocalAccountStore(config.local_account_path) if store.load() is not None: - confirm = input( - "A local account already exists. Overwrite it? [y/N] " - ).strip().lower() + confirm = ( + input("A local account already exists. Overwrite it? [y/N] ") + .strip() + .lower() + ) if confirm != "y": print("Setup cancelled; existing account left unchanged.") return diff --git a/apps/device-host-agent/host_agent/instance_lock.py b/apps/device-host-agent/host_agent/instance_lock.py new file mode 100644 index 0000000..b9f771c --- /dev/null +++ b/apps/device-host-agent/host_agent/instance_lock.py @@ -0,0 +1,71 @@ +"""Per-installation exclusive-execution lock for the Host Agent process. + +Backed by ``filelock.FileLock`` (OS-level advisory file locking) so that +the lock is released automatically when the holding process exits for +any reason — clean shutdown, unhandled exception, or ``SIGKILL`` — without +requiring any stale-lock cleanup step. See +``openspec/changes/host-agent-single-instance-lock/design.md`` for the +rationale behind choosing ``filelock`` over hand-rolled ``fcntl``/``msvcrt`` +branching. +""" + +from __future__ import annotations + +from pathlib import Path + +from filelock import FileLock, Timeout + +_LOCK_FILENAME = "host_agent.lock" + + +class InstanceAlreadyRunningError(RuntimeError): + """Raised when another live process already holds the instance lock.""" + + def __init__(self, lock_path: Path) -> None: + self.lock_path = lock_path + super().__init__( + "another Host Agent instance is already running for this " + f"identity state directory (lock held at {lock_path}); " + "stop the other process before starting a new one" + ) + + +class InstanceLock: + """Exclusive, non-blocking lock scoped to a Host Agent state directory. + + The state directory (typically ``identity_path.parent``) is created on + construction to match the colocated-file convention already used for + ``host_console_history.sqlite3`` and ``host_governance_policy.json``. + """ + + def __init__(self, state_dir: Path) -> None: + state_dir.mkdir(parents=True, exist_ok=True) + self._lock_path = state_dir / _LOCK_FILENAME + self._lock = FileLock(str(self._lock_path), timeout=0) + self._acquired = False + + @property + def lock_path(self) -> Path: + return self._lock_path + + def acquire(self) -> None: + try: + self._lock.acquire() + except Timeout as exc: + raise InstanceAlreadyRunningError(self._lock_path) from exc + self._acquired = True + + def release(self) -> None: + if not self._acquired: + return + try: + self._lock.release() + finally: + self._acquired = False + + def __enter__(self) -> InstanceLock: + self.acquire() + return self + + def __exit__(self, *exc_info: object) -> None: + self.release() diff --git a/apps/device-host-agent/pyproject.toml b/apps/device-host-agent/pyproject.toml index 7299377..e0b02bc 100644 --- a/apps/device-host-agent/pyproject.toml +++ b/apps/device-host-agent/pyproject.toml @@ -7,6 +7,7 @@ dependencies = [ "device-agent-runtime==0.1.0", "device-cloud-platform==0.1.0", "fastapi>=0.115.0", + "filelock>=3.0", "httpx>=0.27.0", "jinja2>=3.1", "uvicorn[standard]>=0.30.0", diff --git a/apps/device-host-agent/tests/test_app.py b/apps/device-host-agent/tests/test_app.py index 90673f1..9b425eb 100644 --- a/apps/device-host-agent/tests/test_app.py +++ b/apps/device-host-agent/tests/test_app.py @@ -6,6 +6,7 @@ from contextlib import suppress from datetime import UTC, datetime, timedelta import httpx +import pytest from cloud.internal_api.models import ( AssignmentModel, @@ -16,6 +17,7 @@ from device.manager import DeviceManager from host_agent.app import HostAgentApplication, create_application from host_agent.config import HostAgentConfig from host_agent.identity import HostIdentityStore +from host_agent.instance_lock import InstanceAlreadyRunningError from storage.device_config import DeviceConfigStore @@ -576,3 +578,123 @@ def test_run_async_starts_supervisor_before_first_heartbeat_connect() -> None: assert events.index("supervisor-stop") < events.index("closed") asyncio.run(scenario()) + + +def test_second_create_application_against_held_lock_raises_before_enrollment( + tmp_path, monkeypatch +) -> None: + monkeypatch.chdir(tmp_path) + identity_path = tmp_path / "host_identity.json" + config = HostAgentConfig( + control_plane_url="https://control.example", + host_id="host-a", + token="secret", + identity_path=identity_path, + ) + + class TrackingEnrollmentClient: + def __init__(self) -> None: + self.config = config + self.calls: list[str] = [] + + def enroll_host(self, **payload): + self.calls.append("host") + return HostEnrollmentResponse(host_id="host-a") + + def enroll_device(self, **payload): + self.calls.append("device") + return DeviceEnrollmentResponse(device_id="device-a") + + def close(self) -> None: + return None + + first_client = TrackingEnrollmentClient() + first_app = create_application( + config=config, + identity_store=HostIdentityStore(identity_path), + enrollment_client=first_client, # type: ignore[arg-type] + ) + try: + second_client = TrackingEnrollmentClient() + with pytest.raises(InstanceAlreadyRunningError) as info: + create_application( + config=config, + identity_store=HostIdentityStore(identity_path), + enrollment_client=second_client, # type: ignore[arg-type] + ) + assert info.value.lock_path == identity_path.parent / "host_agent.lock" + assert second_client.calls == [] + finally: + asyncio.run(first_app.client.aclose()) + + +def test_create_application_with_independent_identity_paths_coexist( + tmp_path, monkeypatch +) -> None: + monkeypatch.chdir(tmp_path) + config_a = HostAgentConfig( + control_plane_url="https://control.example", + host_id="host-a", + token="secret", + identity_path=tmp_path / "identity-a" / "host_identity.json", + ) + config_b = HostAgentConfig( + control_plane_url="https://control.example", + host_id="host-b", + token="secret", + identity_path=tmp_path / "identity-b" / "host_identity.json", + ) + + app_a = create_application(config=config_a, manager=DeviceManager()) + try: + app_b = create_application(config=config_b, manager=DeviceManager()) + asyncio.run(app_b.client.aclose()) + finally: + asyncio.run(app_a.client.aclose()) + + +def test_lock_released_after_run_async_allows_restart(tmp_path, monkeypatch) -> None: + monkeypatch.chdir(tmp_path) + identity_path = tmp_path / "host_identity.json" + config = HostAgentConfig( + control_plane_url="https://control.example", + host_id="host-a", + token="secret", + identity_path=identity_path, + ) + + first_app = create_application(config=config, manager=DeviceManager()) + + class StoppedClient: + async def claim(self): + raise AssertionError("polling must not start") + + async def aclose(self): + return None + + class FastHeartbeat: + async def run(self, stop): + await stop.wait() + + async def sync_once(self): + return None + + class IdleProcessor: + async def process(self, assignment): + raise AssertionError("no assignment expected") + + def request_stop(self): + return None + + first_app.client = StoppedClient() # type: ignore[assignment] + first_app.heartbeat = FastHeartbeat() # type: ignore[assignment] + first_app.processor = IdleProcessor() # type: ignore[assignment] + + stop = asyncio.Event() + stop.set() + asyncio.run(first_app.run_async(stop)) + + # Lock must be free now; a fresh create_application against the same + # identity_path must succeed (simulates a clean restart). + second_app = create_application(config=config, manager=DeviceManager()) + asyncio.run(second_app.client.aclose()) diff --git a/apps/device-host-agent/tests/test_cli.py b/apps/device-host-agent/tests/test_cli.py index 3817584..bdfcea0 100644 --- a/apps/device-host-agent/tests/test_cli.py +++ b/apps/device-host-agent/tests/test_cli.py @@ -5,6 +5,7 @@ import sys import pytest from host_agent import cli +from host_agent.instance_lock import InstanceAlreadyRunningError from host_agent.local_account import LocalAccountStore @@ -60,7 +61,9 @@ def test_existing_account_fast_path_skips_prompting(monkeypatch, tmp_path) -> No assert captured["config"].display_name == "operator" -def test_interactive_first_run_prompts_and_creates_account(monkeypatch, tmp_path) -> None: +def test_interactive_first_run_prompts_and_creates_account( + monkeypatch, tmp_path +) -> None: _set_env(monkeypatch, tmp_path) monkeypatch.setattr(sys.stdin, "isatty", lambda: True) inputs = iter(["operator"]) @@ -120,3 +123,26 @@ def test_setup_subcommand_refuses_overwrite_without_confirmation( cli.main(["setup"]) assert store.load() == original + + +def test_duplicate_instance_exits_with_clear_error( + monkeypatch, tmp_path, capsys +) -> None: + _set_env(monkeypatch, tmp_path) + LocalAccountStore(tmp_path / "account.json").create( + "operator", "correct horse battery staple" + ) + lock_path = tmp_path / "identity_dir" / "host_agent.lock" + + def raising_create_application(*, config=None, **kwargs): + raise InstanceAlreadyRunningError(lock_path) + + monkeypatch.setattr(cli, "create_application", raising_create_application) + + with pytest.raises(SystemExit) as exc_info: + cli.main([]) + + assert exc_info.value.code == 1 + err = capsys.readouterr().err + assert "another Host Agent instance" in err + assert str(lock_path) in err diff --git a/apps/device-host-agent/tests/test_instance_lock.py b/apps/device-host-agent/tests/test_instance_lock.py new file mode 100644 index 0000000..f0589d2 --- /dev/null +++ b/apps/device-host-agent/tests/test_instance_lock.py @@ -0,0 +1,80 @@ +from __future__ import annotations + +import pytest + +from host_agent.instance_lock import ( + InstanceAlreadyRunningError, + InstanceLock, +) + + +def test_acquire_succeeds_when_free(tmp_path) -> None: + lock = InstanceLock(tmp_path) + lock.acquire() + assert lock.lock_path == tmp_path / "host_agent.lock" + lock.release() + + +def test_acquire_raises_when_already_held(tmp_path) -> None: + first = InstanceLock(tmp_path) + first.acquire() + try: + second = InstanceLock(tmp_path) + with pytest.raises(InstanceAlreadyRunningError) as info: + second.acquire() + assert info.value.lock_path == tmp_path / "host_agent.lock" + assert "another Host Agent instance" in str(info.value) + finally: + first.release() + + +def test_release_then_reacquire_succeeds(tmp_path) -> None: + lock = InstanceLock(tmp_path) + lock.acquire() + lock.release() + # Same handle can reacquire after release. + lock.acquire() + lock.release() + # A fresh handle against the same path also succeeds. + other = InstanceLock(tmp_path) + other.acquire() + other.release() + + +def test_release_is_idempotent_and_safe_before_acquire(tmp_path) -> None: + lock = InstanceLock(tmp_path) + # Releasing before acquire must be a no-op, not raise. + lock.release() + lock.acquire() + lock.release() + lock.release() + + +def test_independent_paths_never_contend(tmp_path) -> None: + dir_a = tmp_path / "identity-a" + dir_b = tmp_path / "identity-b" + lock_a = InstanceLock(dir_a) + lock_a.acquire() + try: + lock_b = InstanceLock(dir_b) + lock_b.acquire() + lock_b.release() + finally: + lock_a.release() + + +def test_context_manager_releases_on_exit(tmp_path) -> None: + with InstanceLock(tmp_path): + second = InstanceLock(tmp_path) + with pytest.raises(InstanceAlreadyRunningError): + second.acquire() + # After the context exits, a fresh handle can acquire. + InstanceLock(tmp_path).acquire() + + +def test_state_directory_is_created_if_missing(tmp_path) -> None: + nested = tmp_path / "nested" / "state" + lock = InstanceLock(nested) + assert nested.exists() + lock.acquire() + lock.release() diff --git a/docs/MACOS_IPHONE_SETUP.md b/docs/MACOS_IPHONE_SETUP.md index e496015..5cb8a31 100644 --- a/docs/MACOS_IPHONE_SETUP.md +++ b/docs/MACOS_IPHONE_SETUP.md @@ -404,6 +404,24 @@ Host Agent 会在本机回环地址提供 Console。必须保留并保护 `tasks 直接注册意味着任何能访问该云端地址的设备都能自行注册成为 Host，没有审批环节， 也没有限流保护；这一取舍依赖网络边界（防火墙/反向代理）而非应用层限制。 +### 重复实例保护 + +Host Agent 启动时会先尝试获取一个独占的本地文件锁（位于 +`tasks/host_agent.lock`，与 `host_identity.json` 同目录），确保同一个 identity +状态目录下同一时刻只有一个 Host Agent 进程在运行。这是为了避免两个进程用同一个 +`host_id` 同时心跳和 claim，导致任务派发落到从未注册过该设备的进程上（典型的 +`DeviceNotFoundError` 事故场景）。 + +如果启动时锁已被另一个仍在运行的进程持有，Host Agent 会立即以非零退出码退出， +stderr 打印 `error: another Host Agent instance is already running ...`，**不会** +联系云端、不会触发任何 enrollment。处理方式： + +1. 用 `lsof tasks/host_agent.lock`（macOS）/ 任务管理器（Windows）或 + `pgrep -fa device-host-agent` 找到仍在运行的旧进程。 +2. 确认旧进程应被停止后，再 `kill` 它（或等它的 graceful shutdown 完成）。 +3. 重新启动 Host Agent。崩溃/被 `kill -9` 的旧进程退出时 OS 会自动释放文件锁， + 不需要手动删除 `host_agent.lock`。 + ### 使用本地 Web Console Host Agent 启动时会同时启动本地 Web Console，用于在这台 Mac 上直接查看和管理 diff --git a/uv.lock b/uv.lock index 29b88d4..8d807ac 100644 --- a/uv.lock +++ b/uv.lock @@ -569,6 +569,7 @@ dependencies = [ { name = "device-agent-runtime" }, { name = "device-cloud-platform" }, { name = "fastapi" }, + { name = "filelock" }, { name = "httpx" }, { name = "jinja2" }, { name = "uvicorn", extra = ["standard"] }, @@ -579,6 +580,7 @@ requires-dist = [ { name = "device-agent-runtime", editable = "." }, { name = "device-cloud-platform", editable = "packages/cloud-platform" }, { name = "fastapi", specifier = ">=0.115.0" }, + { name = "filelock", specifier = ">=3.0" }, { name = "httpx", specifier = ">=0.27.0" }, { name = "jinja2", specifier = ">=3.1" }, { name = "uvicorn", extras = ["standard"], specifier = ">=0.30.0" },