q792602257/agentic-mobile-control
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(host-agent): add single-instance lock to prevent duplicate-process dispatch races

Browse Source 
Acquire an exclusive, non-blocking filelock on the identity state directory
as the first action of create_application(), before resolve_host_identity()
or any enrollment/heartbeat side effect. A second process against the same
identity_path exits immediately with InstanceAlreadyRunningError naming the
lock path; the lock releases automatically on any process exit (including
SIGKILL) via OS-level advisory locking, and explicitly during run_async()'s
shutdown finally block. filelock is promoted from transitive to direct
dependency (version unchanged at 3.29.7).

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
q792602257
2026-07-14 15:49:30 +08:00
co-authored by Claude Opus 4.6
parent 99bde4febb
commit d00ada67a5
9 changed files with 444 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/device-host-agent/host_agent/app.py
+112 -97
View File
@@ -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(
apps/device-host-agent/host_agent/cli.py
+11 -4
View File
@@ -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
apps/device-host-agent/host_agent/instance_lock.py
+71
View File
@@ -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()
apps/device-host-agent/pyproject.toml
+1
View File
@@ -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",
apps/device-host-agent/tests/test_app.py
+122
View File
@@ -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())
apps/device-host-agent/tests/test_cli.py
+27 -1
View File
@@ -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
apps/device-host-agent/tests/test_instance_lock.py
+80
View File
@@ -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()