feat(host-agent): include mcp_busy_device_ids in heartbeat payload

This commit is contained in:
2026-07-21 14:56:21 +08:00
parent b0932dd398
commit ab15218b27
4 changed files with 134 additions and 10 deletions
@@ -353,6 +353,60 @@ def test_submit_self_task_does_not_duplicate_when_response_is_lost() -> None:
assert attempts == 1
def test_heartbeat_includes_mcp_busy_device_ids_in_payload() -> None:
"""When mcp_busy_device_ids is passed, the client sends it in the request."""
captured: list[dict[str, object]] = []
def handler(request: httpx.Request) -> httpx.Response:
captured.append(json.loads(request.content))
return httpx.Response(
200,
json={
"host_id": "host-a",
"accepted_devices": 0,
"received_at": "2026-07-12T00:00:00Z",
},
)
async def scenario() -> None:
async with httpx.AsyncClient(
transport=httpx.MockTransport(handler),
base_url="https://control.example",
) as http_client:
client = HostAgentClient(_config(), http_client=http_client)
await client.heartbeat([], mcp_busy_device_ids=["phone-1"])
asyncio.run(scenario())
assert captured[0]["mcp_busy_device_ids"] == ["phone-1"]
def test_heartbeat_omits_mcp_busy_device_ids_when_empty() -> None:
"""Empty list is omitted from the payload (backward compatible)."""
captured: list[dict[str, object]] = []
def handler(request: httpx.Request) -> httpx.Response:
captured.append(json.loads(request.content))
return httpx.Response(
200,
json={
"host_id": "host-a",
"accepted_devices": 0,
"received_at": "2026-07-12T00:00:00Z",
},
)
async def scenario() -> None:
async with httpx.AsyncClient(
transport=httpx.MockTransport(handler),
base_url="https://control.example",
) as http_client:
client = HostAgentClient(_config(), http_client=http_client)
await client.heartbeat([], mcp_busy_device_ids=[])
asyncio.run(scenario())
assert "mcp_busy_device_ids" not in captured[0]
def test_bootstrap_client_directly_enrolls_and_enrolls_device() -> None:
requests: list[httpx.Request] = []
host_attempts = 0
+59 -3
View File
@@ -8,6 +8,7 @@ from cloud.internal_api.models import HostGovernancePolicyModel
from device.manager import DeviceManager
from host_agent.config import HostAgentConfig
from host_agent.heartbeat import HeartbeatSynchronizer, build_device_snapshot
from host_agent.mcp_lock import McpBusyTracker
from host_agent.policy_cache import HostPolicyCacheStore
from host_agent.status import AgentStatusTracker
@@ -60,7 +61,7 @@ def test_heartbeat_synchronizer_runs_at_configured_interval_until_stopped() -> N
calls: list[list[str]] = []
class FakeClient:
async def heartbeat(self, devices, *, address=None, policy_revision=0):
async def heartbeat(self, devices, *, address=None, policy_revision=0, **kwargs):
calls.append([device.device_id for device in devices])
return HeartbeatResponse(
host_id="host-a",
@@ -98,7 +99,7 @@ def test_sync_once_notifies_status_tracker_and_on_sync_with_device_count() -> No
)
class FakeClient:
async def heartbeat(self, devices, *, address=None, policy_revision=0):
async def heartbeat(self, devices, *, address=None, policy_revision=0, **kwargs):
return HeartbeatResponse(
host_id="host-a",
accepted_devices=len(devices),
@@ -133,7 +134,7 @@ def test_heartbeat_caches_safe_host_policy_and_reuses_its_revision(tmp_path) ->
revisions: list[int] = []
class UpdatingClient:
async def heartbeat(self, devices, *, address=None, policy_revision=0):
async def heartbeat(self, devices, *, address=None, policy_revision=0, **kwargs):
revisions.append(policy_revision)
return HeartbeatResponse(
host_id="host-a",
@@ -178,3 +179,58 @@ def test_heartbeat_caches_safe_host_policy_and_reuses_its_revision(tmp_path) ->
assert '"token":' not in (
tmp_path / "host_policy.json"
).read_text(encoding="utf-8")
def test_sync_once_passes_mcp_busy_device_ids_to_client() -> None:
"""When mcp_busy_tracker has a lease, sync_once relays the device_ids."""
manager = DeviceManager()
tracker = McpBusyTracker()
assert tracker.acquire("phone-1", "sess-a")
last_kwargs: dict[str, object] = {}
class FakeClient:
async def heartbeat(self, devices, *, address=None, policy_revision=0, **kwargs):
last_kwargs.update(kwargs)
return HeartbeatResponse(
host_id="host-a",
accepted_devices=len(devices),
received_at=datetime.now(UTC),
)
async def scenario() -> None:
sync = HeartbeatSynchronizer(
manager,
FakeClient(), # type: ignore[arg-type]
_config(),
mcp_busy_tracker=tracker,
)
await sync.sync_once()
asyncio.run(scenario())
assert last_kwargs.get("mcp_busy_device_ids") == ["phone-1"]
def test_sync_once_passes_empty_when_tracker_is_none() -> None:
"""Default: no tracker → no busy device ids forwarded."""
manager = DeviceManager()
last_kwargs: dict[str, object] = {}
class FakeClient:
async def heartbeat(self, devices, *, address=None, policy_revision=0, **kwargs):
last_kwargs.update(kwargs)
return HeartbeatResponse(
host_id="host-a",
accepted_devices=len(devices),
received_at=datetime.now(UTC),
)
async def scenario() -> None:
sync = HeartbeatSynchronizer(
manager,
FakeClient(), # type: ignore[arg-type]
_config(),
)
await sync.sync_once()
asyncio.run(scenario())
assert not last_kwargs.get("mcp_busy_device_ids")