From 8619786a51eab29e03a196a2cee392f868160253 Mon Sep 17 00:00:00 2001 From: speriaswamy-amd Date: Wed, 26 Aug 2026 13:46:17 -0400 Subject: [PATCH 1/9] intial mock up --- cvs/core/agent/http_client.py | 111 ++++++++++++++++++++++++++++++++++ 1 file changed, 111 insertions(+) create mode 100644 cvs/core/agent/http_client.py diff --git a/cvs/core/agent/http_client.py b/cvs/core/agent/http_client.py new file mode 100644 index 000000000..4d54caa5c --- /dev/null +++ b/cvs/core/agent/http_client.py @@ -0,0 +1,111 @@ +''' +Copyright 2025 Advanced Micro Devices, Inc. +All rights reserved. This notice is intended as a precaution against inadvertent publication and does not imply publication or any waiver of confidentiality. +The year included in the foregoing notice is the year of creation of the work. +All code contained here is Property of Advanced Micro Devices, Inc. +''' + +import asyncio +import uuid +from dataclasses import dataclass +from pathlib import Path + +import httpx + +from . import messages + + +@dataclass +class HostOutput: + host: str + stdout: list[str] + stderr: list[str] + exit_code: int | None + exception: Exception | None + + +class ParallelHTTPClientError(Exception): + '''Raised by run_command when stop_on_errors=True and at least one host failed to reach its agent + or returned an unparseable response. Mirrors ParallelSSHClient's raise-on-connection-failure behavior; + a nonzero remote exit_code is not itself a failure here, matching pssh's stop_on_errors semantics.''' + + +class ParallelHTTPClient: + '''ParallelSSHClient-API-compatible client that fans a command out to per-host HTTP agents.''' + + def __init__(self, agent_urls: dict[str, str], token: str, connect_timeout: float | None = None) -> None: + self._agent_urls = agent_urls + self._token = token + self._connect_timeout = connect_timeout + + def _auth_header(self) -> dict[str, str]: + return {messages.AUTH_HEADER: f"{messages.AUTH_SCHEME} {self._token}"} + + def _build_exec_requests( + self, cmd: str, host_args: list | None, read_timeout: float | None + ) -> dict[str, messages.ExecRequest]: + hosts = list(self._agent_urls) + if host_args is not None: + if len(host_args) != len(hosts): + raise ValueError(f"host_args has {len(host_args)} entries but there are {len(hosts)} hosts") + commands = [cmd % args for args in host_args] + else: + commands = [cmd] * len(hosts) + return { + host: messages.ExecRequest( + cmd=command, + env={}, + cwd=Path.cwd(), + timeout=read_timeout, + inactivity_timeout=None, + cmd_id=uuid.uuid4().hex, + out_path=None, + output_mode=messages.ExecOutputMode.INLINE, + ) + for host, command in zip(hosts, commands) + } + + async def _run_one( + self, client: httpx.AsyncClient, host: str, url: str, request: messages.ExecRequest + ) -> HostOutput: + try: + response = await client.post(f"{url}{messages.EXEC_PATH}", content=request.model_dump_json()) + response.raise_for_status() + exec_response = messages.parse_message(messages.ExecResponse, response.text) + except Exception as exc: # noqa: BLE001 - captured per-host so one bad host doesn't sink the others + return HostOutput(host=host, stdout=[], stderr=[], exit_code=None, exception=exc) + return HostOutput( + host=host, + stdout=exec_response.stdout or [], + stderr=exec_response.stderr or [], + exit_code=exec_response.exit_code, + exception=None, + ) + + async def _run_command_async( + self, requests: dict[str, messages.ExecRequest], read_timeout: float | None + ) -> list[HostOutput]: + timeout = httpx.Timeout(read_timeout, connect=self._connect_timeout) + async with httpx.AsyncClient(headers=self._auth_header(), timeout=timeout) as client: + tasks = [self._run_one(client, host, self._agent_urls[host], request) for host, request in requests.items()] + return await asyncio.gather(*tasks) + + def run_command( + self, + cmd: str, + stop_on_errors: bool = True, + read_timeout: float | None = None, + host_args: list | None = None, + ) -> list[HostOutput]: + requests = self._build_exec_requests(cmd, host_args, read_timeout) + outputs = asyncio.run(self._run_command_async(requests, read_timeout)) + if stop_on_errors: + failed = [output for output in outputs if output.exception is not None] + if failed: + details = ", ".join(f"{output.host}: {output.exception}" for output in failed) + raise ParallelHTTPClientError(f"{len(failed)} host(s) failed: {details}") + return outputs + + def join(self) -> None: + '''No-op: kept for API parity with ParallelSSHClient.join(), which waits on SFTP transfers this + client never starts.''' From fae29d1e40c5802bc38d0320a4a53e292b62858e Mon Sep 17 00:00:00 2001 From: speriaswamy-amd Date: Wed, 26 Aug 2026 14:26:38 -0400 Subject: [PATCH 2/9] Lazy connection resuse, health + shutdown API calls --- cvs/core/agent/http_client.py | 99 ++++++++++++++++++++++++++++------- 1 file changed, 79 insertions(+), 20 deletions(-) diff --git a/cvs/core/agent/http_client.py b/cvs/core/agent/http_client.py index 4d54caa5c..1282e1776 100644 --- a/cvs/core/agent/http_client.py +++ b/cvs/core/agent/http_client.py @@ -25,22 +25,49 @@ class HostOutput: class ParallelHTTPClientError(Exception): - '''Raised by run_command when stop_on_errors=True and at least one host failed to reach its agent - or returned an unparseable response. Mirrors ParallelSSHClient's raise-on-connection-failure behavior; - a nonzero remote exit_code is not itself a failure here, matching pssh's stop_on_errors semantics.''' + '''Raised when stop_on_errors=True and at least one host failed to reach its agent or returned an + unparseable response. Mirrors ParallelSSHClient's raise-on-connection-failure behavior; a nonzero + remote exit_code is not itself a failure here, matching pssh's stop_on_errors semantics.''' class ParallelHTTPClient: - '''ParallelSSHClient-API-compatible client that fans a command out to per-host HTTP agents.''' + '''Async, ParallelSSHClient-inspired client that fans a command out to per-host HTTP agents. + + Holds one lazily-created, long-lived httpx.AsyncClient shared across calls so repeated commands + reuse pooled connections instead of paying a new TCP/TLS handshake each time. Callers own the event + loop for the lifetime of the client (there is no internal asyncio.run()); call destroy() or use + `async with` when done to release pooled connections.''' def __init__(self, agent_urls: dict[str, str], token: str, connect_timeout: float | None = None) -> None: - self._agent_urls = agent_urls + self._agent_urls = dict(agent_urls) self._token = token self._connect_timeout = connect_timeout + self._client: httpx.AsyncClient | None = None + + async def __aenter__(self) -> "ParallelHTTPClient": + return self + + async def __aexit__(self, *exc_info) -> None: + await self.destroy() def _auth_header(self) -> dict[str, str]: return {messages.AUTH_HEADER: f"{messages.AUTH_SCHEME} {self._token}"} + def _get_client(self) -> httpx.AsyncClient: + if self._client is None: + self._client = httpx.AsyncClient(headers=self._auth_header()) + return self._client + + def rebuild(self, agent_urls: dict[str, str]) -> None: + '''Replace the host map, e.g. to drop hosts pruned after a failed health check. The shared + client's connection pool needs no action: idle connections to removed hosts simply age out.''' + self._agent_urls = dict(agent_urls) + + async def destroy(self) -> None: + if self._client is not None: + await self._client.aclose() + self._client = None + def _build_exec_requests( self, cmd: str, host_args: list | None, read_timeout: float | None ) -> dict[str, messages.ExecRequest]: @@ -66,10 +93,19 @@ def _build_exec_requests( } async def _run_one( - self, client: httpx.AsyncClient, host: str, url: str, request: messages.ExecRequest + self, + client: httpx.AsyncClient, + host: str, + url: str, + request: messages.ExecRequest, + read_timeout: float | None, ) -> HostOutput: try: - response = await client.post(f"{url}{messages.EXEC_PATH}", content=request.model_dump_json()) + response = await client.post( + f"{url}{messages.EXEC_PATH}", + content=request.model_dump_json(), + timeout=httpx.Timeout(read_timeout, connect=self._connect_timeout), + ) response.raise_for_status() exec_response = messages.parse_message(messages.ExecResponse, response.text) except Exception as exc: # noqa: BLE001 - captured per-host so one bad host doesn't sink the others @@ -82,15 +118,7 @@ async def _run_one( exception=None, ) - async def _run_command_async( - self, requests: dict[str, messages.ExecRequest], read_timeout: float | None - ) -> list[HostOutput]: - timeout = httpx.Timeout(read_timeout, connect=self._connect_timeout) - async with httpx.AsyncClient(headers=self._auth_header(), timeout=timeout) as client: - tasks = [self._run_one(client, host, self._agent_urls[host], request) for host, request in requests.items()] - return await asyncio.gather(*tasks) - - def run_command( + async def run_command( self, cmd: str, stop_on_errors: bool = True, @@ -98,7 +126,13 @@ def run_command( host_args: list | None = None, ) -> list[HostOutput]: requests = self._build_exec_requests(cmd, host_args, read_timeout) - outputs = asyncio.run(self._run_command_async(requests, read_timeout)) + client = self._get_client() + outputs = await asyncio.gather( + *( + self._run_one(client, host, self._agent_urls[host], request, read_timeout) + for host, request in requests.items() + ) + ) if stop_on_errors: failed = [output for output in outputs if output.exception is not None] if failed: @@ -106,6 +140,31 @@ def run_command( raise ParallelHTTPClientError(f"{len(failed)} host(s) failed: {details}") return outputs - def join(self) -> None: - '''No-op: kept for API parity with ParallelSSHClient.join(), which waits on SFTP transfers this - client never starts.''' + async def _fan_out(self, method: str, path: str, stop_on_errors: bool) -> dict[str, bool]: + client = self._get_client() + + async def call(host: str, url: str) -> tuple[str, bool | Exception]: + try: + response = await client.request(method, f"{url}{path}") + response.raise_for_status() + except Exception as exc: # noqa: BLE001 - captured per-host, reported rather than raised + return host, exc + return host, True + + results = await asyncio.gather(*(call(host, url) for host, url in self._agent_urls.items())) + failed = {host: outcome for host, outcome in results if isinstance(outcome, Exception)} + if stop_on_errors and failed: + details = ", ".join(f"{host}: {exc}" for host, exc in failed.items()) + raise ParallelHTTPClientError(f"{len(failed)} host(s) failed: {details}") + return {host: outcome is True for host, outcome in results} + + async def health(self) -> dict[str, bool]: + '''Liveness probe per host; never raises regardless of failures - an unreachable host is the + answer this call exists to produce (feeds rebuild()'s pruning decision), not an error.''' + return await self._fan_out("GET", messages.HEALTH_PATH, stop_on_errors=False) + + async def shutdown(self, stop_on_errors: bool = False) -> dict[str, bool]: + '''Ask every host's agent to terminate its spawned processes and exit. Defaults to best-effort + (stop_on_errors=False), unlike run_command: one already-dead straggler during cleanup shouldn't + stop the rest from being told to shut down.''' + return await self._fan_out("POST", messages.SHUTDOWN_PATH, stop_on_errors) From 939d02f1129d1c772347ff17ebd88db9817d03fb Mon Sep 17 00:00:00 2001 From: speriaswamy-amd Date: Wed, 26 Aug 2026 14:39:06 -0400 Subject: [PATCH 3/9] Fix linter errors --- cvs/core/agent/http_client.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/cvs/core/agent/http_client.py b/cvs/core/agent/http_client.py index 1282e1776..39afd5690 100644 --- a/cvs/core/agent/http_client.py +++ b/cvs/core/agent/http_client.py @@ -69,7 +69,7 @@ async def destroy(self) -> None: self._client = None def _build_exec_requests( - self, cmd: str, host_args: list | None, read_timeout: float | None + self, cmd: str, host_args: list[str] | None, read_timeout: float | None ) -> dict[str, messages.ExecRequest]: hosts = list(self._agent_urls) if host_args is not None: @@ -83,7 +83,7 @@ def _build_exec_requests( cmd=command, env={}, cwd=Path.cwd(), - timeout=read_timeout, + timeout=round(read_timeout) if read_timeout is not None else None, inactivity_timeout=None, cmd_id=uuid.uuid4().hex, out_path=None, @@ -123,7 +123,7 @@ async def run_command( cmd: str, stop_on_errors: bool = True, read_timeout: float | None = None, - host_args: list | None = None, + host_args: list[str] | None = None, ) -> list[HostOutput]: requests = self._build_exec_requests(cmd, host_args, read_timeout) client = self._get_client() From 991e108ba4a0fe445a02ac97a0364531a36ca795 Mon Sep 17 00:00:00 2001 From: speriaswamy-amd Date: Wed, 26 Aug 2026 14:52:34 -0400 Subject: [PATCH 4/9] Handle exception mapping to pssh --- cvs/core/agent/http_client.py | 32 ++++++++++++++++++++++++++++++-- 1 file changed, 30 insertions(+), 2 deletions(-) diff --git a/cvs/core/agent/http_client.py b/cvs/core/agent/http_client.py index 39afd5690..35db0ce74 100644 --- a/cvs/core/agent/http_client.py +++ b/cvs/core/agent/http_client.py @@ -30,6 +30,34 @@ class ParallelHTTPClientError(Exception): remote exit_code is not itself a failure here, matching pssh's stop_on_errors semantics.''' +class HTTPConnectionError(Exception): + '''Host was unreachable at the transport level: DNS failure, connection refused, TLS failure, or a + connect/read/write/pool timeout. Wraps httpx.TransportError and its subclasses. Analogous to + pssh.exceptions.ConnectionError/Timeout/SessionError in cvs/lib/parallel/pssh.py's + prune_unreachable_hosts - a signal the host itself may be down, and a pruning candidate.''' + + +class HTTPProtocolError(Exception): + '''Host was reached but the request failed at the HTTP/application layer: a non-2xx response + (bad auth, exec-already-in-progress conflict, agent-side error) or an unparseable response body. + Wraps httpx.HTTPStatusError and messages.MessageParseError. The host is alive, so this may + succeed on retry - not a pruning candidate, analogous to pssh's non-pruned auth/protocol errors.''' + + +def _classify_exception(exc: Exception) -> Exception: + '''Wrap a raw httpx/messages exception so callers can distinguish "host unreachable" from "host + reached but request failed" via isinstance(exception, HTTPConnectionError), the same split + cvs/lib/parallel/pssh.py's prune_unreachable_hosts draws for SSH via pssh.exceptions.''' + if isinstance(exc, httpx.TransportError): + wrapped: Exception = HTTPConnectionError(str(exc)) + elif isinstance(exc, (httpx.HTTPStatusError, messages.MessageParseError)): + wrapped = HTTPProtocolError(str(exc)) + else: + return exc + wrapped.__cause__ = exc + return wrapped + + class ParallelHTTPClient: '''Async, ParallelSSHClient-inspired client that fans a command out to per-host HTTP agents. @@ -109,7 +137,7 @@ async def _run_one( response.raise_for_status() exec_response = messages.parse_message(messages.ExecResponse, response.text) except Exception as exc: # noqa: BLE001 - captured per-host so one bad host doesn't sink the others - return HostOutput(host=host, stdout=[], stderr=[], exit_code=None, exception=exc) + return HostOutput(host=host, stdout=[], stderr=[], exit_code=None, exception=_classify_exception(exc)) return HostOutput( host=host, stdout=exec_response.stdout or [], @@ -148,7 +176,7 @@ async def call(host: str, url: str) -> tuple[str, bool | Exception]: response = await client.request(method, f"{url}{path}") response.raise_for_status() except Exception as exc: # noqa: BLE001 - captured per-host, reported rather than raised - return host, exc + return host, _classify_exception(exc) return host, True results = await asyncio.gather(*(call(host, url) for host, url in self._agent_urls.items())) From df212a0e57ae53dda0b2186c5b1ff8db8846c8e6 Mon Sep 17 00:00:00 2001 From: speriaswamy-amd Date: Wed, 26 Aug 2026 15:01:08 -0400 Subject: [PATCH 5/9] Unit tests --- cvs/core/agent/http_client.py | 11 +- cvs/core/agent/unittests/test_http_client.py | 297 +++++++++++++++++++ 2 files changed, 306 insertions(+), 2 deletions(-) create mode 100644 cvs/core/agent/unittests/test_http_client.py diff --git a/cvs/core/agent/http_client.py b/cvs/core/agent/http_client.py index 35db0ce74..92c6fe6ba 100644 --- a/cvs/core/agent/http_client.py +++ b/cvs/core/agent/http_client.py @@ -66,10 +66,17 @@ class ParallelHTTPClient: loop for the lifetime of the client (there is no internal asyncio.run()); call destroy() or use `async with` when done to release pooled connections.''' - def __init__(self, agent_urls: dict[str, str], token: str, connect_timeout: float | None = None) -> None: + def __init__( + self, + agent_urls: dict[str, str], + token: str, + connect_timeout: float | None = None, + transport: httpx.AsyncBaseTransport | None = None, + ) -> None: self._agent_urls = dict(agent_urls) self._token = token self._connect_timeout = connect_timeout + self._transport = transport self._client: httpx.AsyncClient | None = None async def __aenter__(self) -> "ParallelHTTPClient": @@ -83,7 +90,7 @@ def _auth_header(self) -> dict[str, str]: def _get_client(self) -> httpx.AsyncClient: if self._client is None: - self._client = httpx.AsyncClient(headers=self._auth_header()) + self._client = httpx.AsyncClient(headers=self._auth_header(), transport=self._transport) return self._client def rebuild(self, agent_urls: dict[str, str]) -> None: diff --git a/cvs/core/agent/unittests/test_http_client.py b/cvs/core/agent/unittests/test_http_client.py new file mode 100644 index 000000000..1fe27b480 --- /dev/null +++ b/cvs/core/agent/unittests/test_http_client.py @@ -0,0 +1,297 @@ +''' +Copyright 2025 Advanced Micro Devices, Inc. +All rights reserved. This notice is intended as a precaution against inadvertent publication and does not imply publication or any waiver of confidentiality. +The year included in the foregoing notice is the year of creation of the work. +All code contained here is Property of Advanced Micro Devices, Inc. +''' + +# Unit tests for cvs/core/agent/http_client.py: ParallelHTTPClient's run_command/health/shutdown +# fan-out, host_args/stop_on_errors semantics, exception classification, and session lifecycle. + +import json +import unittest + +import httpx + +from cvs.core.agent import messages +from cvs.core.agent.http_client import ( + HostOutput, + HTTPConnectionError, + HTTPProtocolError, + ParallelHTTPClient, + ParallelHTTPClientError, +) + +TOKEN = "test-token-123" + + +def _exec_handler(request: httpx.Request) -> httpx.Response: + '''Default /v1/exec responder: echoes the requested cmd back as stdout, exit_code 0.''' + body = json.loads(request.content) + return httpx.Response( + 200, + json={ + "exit_code": 0, + "stdout": [body["cmd"]], + "stderr": [], + "stdout_path": None, + "stderr_path": None, + "truncated": False, + "timed_out": False, + }, + ) + + +class HttpClientTestBase(unittest.IsolatedAsyncioTestCase): + def _make_client(self, agent_urls: dict[str, str], handler, token: str = TOKEN, **kwargs) -> ParallelHTTPClient: + client = ParallelHTTPClient(agent_urls, token, transport=httpx.MockTransport(handler), **kwargs) + self.addAsyncCleanup(client.destroy) + return client + + +class TestRunCommand(HttpClientTestBase): + async def test_returns_host_output_per_host(self): + client = self._make_client({"h1": "http://h1", "h2": "http://h2"}, _exec_handler) + outputs = await client.run_command("echo hi") + self.assertEqual({o.host for o in outputs}, {"h1", "h2"}) + for output in outputs: + self.assertEqual(output.stdout, ["echo hi"]) + self.assertEqual(output.stderr, []) + self.assertEqual(output.exit_code, 0) + self.assertIsNone(output.exception) + + async def test_sends_bearer_auth_header(self): + seen_headers = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_headers.append(request.headers.get("authorization")) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler, token="secret-abc") + await client.run_command("true") + self.assertEqual(seen_headers, [f"{messages.AUTH_SCHEME} secret-abc"]) + + async def test_posts_to_exec_path_on_each_hosts_url(self): + seen_urls = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_urls.append(str(request.url)) + return _exec_handler(request) + + client = self._make_client({"h1": "http://host-one", "h2": "http://host-two"}, handler) + await client.run_command("true") + self.assertEqual( + sorted(seen_urls), sorted([f"http://host-one{messages.EXEC_PATH}", f"http://host-two{messages.EXEC_PATH}"]) + ) + + async def test_host_args_substitutes_a_different_command_per_host(self): + client = self._make_client({"h1": "http://h1", "h2": "http://h2"}, _exec_handler) + outputs = await client.run_command("echo %s", host_args=["one", "two"]) + by_host = {o.host: o.stdout for o in outputs} + self.assertEqual(by_host, {"h1": ["echo one"], "h2": ["echo two"]}) + + async def test_host_args_length_mismatch_raises_value_error(self): + client = self._make_client({"h1": "http://h1", "h2": "http://h2"}, _exec_handler) + with self.assertRaises(ValueError): + await client.run_command("echo %s", host_args=["only-one"]) + + async def test_stop_on_errors_true_raises_when_a_host_fails(self): + def handler(request: httpx.Request) -> httpx.Response: + if "bad" in str(request.url): + return httpx.Response(500, text="boom") + return _exec_handler(request) + + client = self._make_client({"good": "http://good", "bad": "http://bad"}, handler) + with self.assertRaises(ParallelHTTPClientError): + await client.run_command("true", stop_on_errors=True) + + async def test_stop_on_errors_false_returns_partial_results(self): + def handler(request: httpx.Request) -> httpx.Response: + if "bad" in str(request.url): + return httpx.Response(500, text="boom") + return _exec_handler(request) + + client = self._make_client({"good": "http://good", "bad": "http://bad"}, handler) + outputs = await client.run_command("true", stop_on_errors=False) + by_host = {o.host: o for o in outputs} + self.assertIsNone(by_host["good"].exception) + self.assertEqual(by_host["good"].exit_code, 0) + self.assertIsInstance(by_host["bad"].exception, HTTPProtocolError) + + async def test_nonzero_remote_exit_code_is_not_a_stop_on_errors_failure(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + json={ + "exit_code": 1, + "stdout": [], + "stderr": ["failed"], + "stdout_path": None, + "stderr_path": None, + "truncated": False, + "timed_out": False, + }, + ) + + client = self._make_client({"h1": "http://h1"}, handler) + outputs = await client.run_command("false", stop_on_errors=True) + self.assertEqual(outputs[0].exit_code, 1) + self.assertIsNone(outputs[0].exception) + + async def test_read_timeout_is_rounded_to_int_for_the_wire_request(self): + seen_requests: list[messages.ExecRequest] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(messages.parse_message(messages.ExecRequest, request.content.decode())) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.run_command("true", read_timeout=2.7) + self.assertEqual(seen_requests[0].timeout, 3) + + async def test_client_is_reused_across_calls(self): + client = self._make_client({"h1": "http://h1"}, _exec_handler) + await client.run_command("true") + first_client = client._client + await client.run_command("true") + self.assertIs(client._client, first_client) + + +class TestExceptionClassification(HttpClientTestBase): + async def test_connection_failure_is_classified_as_connection_error(self): + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ConnectError("connection refused", request=request) + + client = self._make_client({"h1": "http://h1"}, handler) + outputs = await client.run_command("true", stop_on_errors=False) + self.assertIsInstance(outputs[0].exception, HTTPConnectionError) + + async def test_read_timeout_is_classified_as_connection_error(self): + def handler(request: httpx.Request) -> httpx.Response: + raise httpx.ReadTimeout("timed out", request=request) + + client = self._make_client({"h1": "http://h1"}, handler) + outputs = await client.run_command("true", stop_on_errors=False) + self.assertIsInstance(outputs[0].exception, HTTPConnectionError) + + async def test_bad_http_status_is_classified_as_protocol_error(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(401, text="unauthorized") + + client = self._make_client({"h1": "http://h1"}, handler) + outputs = await client.run_command("true", stop_on_errors=False) + self.assertIsInstance(outputs[0].exception, HTTPProtocolError) + + async def test_unparseable_response_body_is_classified_as_protocol_error(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, text="not json") + + client = self._make_client({"h1": "http://h1"}, handler) + outputs = await client.run_command("true", stop_on_errors=False) + self.assertIsInstance(outputs[0].exception, HTTPProtocolError) + + +class TestHealth(HttpClientTestBase): + async def test_all_hosts_healthy(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response(200, json={"ok": True}) + + client = self._make_client({"h1": "http://h1", "h2": "http://h2"}, handler) + self.assertEqual(await client.health(), {"h1": True, "h2": True}) + + async def test_unreachable_host_reported_as_unhealthy_without_raising(self): + def handler(request: httpx.Request) -> httpx.Response: + if "down" in str(request.url): + raise httpx.ConnectError("connection refused", request=request) + return httpx.Response(200, json={"ok": True}) + + client = self._make_client({"up": "http://up", "down": "http://down"}, handler) + self.assertEqual(await client.health(), {"up": True, "down": False}) + + async def test_hits_health_path(self): + seen_paths = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_paths.append(request.url.path) + return httpx.Response(200, json={"ok": True}) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.health() + self.assertEqual(seen_paths, [messages.HEALTH_PATH]) + + +class TestShutdown(HttpClientTestBase): + async def test_hits_shutdown_path_on_every_host(self): + seen = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen.append((request.method, request.url.path, str(request.url))) + return httpx.Response(200, json={"ok": True}) + + client = self._make_client({"h1": "http://h1", "h2": "http://h2"}, handler) + result = await client.shutdown() + self.assertEqual(result, {"h1": True, "h2": True}) + self.assertEqual({(m, p) for m, p, _ in seen}, {("POST", messages.SHUTDOWN_PATH)}) + self.assertEqual( + {url for _, _, url in seen}, {f"http://h1{messages.SHUTDOWN_PATH}", f"http://h2{messages.SHUTDOWN_PATH}"} + ) + + async def test_default_is_best_effort_and_does_not_raise_on_failure(self): + def handler(request: httpx.Request) -> httpx.Response: + if "bad" in str(request.url): + return httpx.Response(500, text="boom") + return httpx.Response(200, json={"ok": True}) + + client = self._make_client({"good": "http://good", "bad": "http://bad"}, handler) + result = await client.shutdown() + self.assertEqual(result, {"good": True, "bad": False}) + + async def test_stop_on_errors_true_raises_when_a_host_fails(self): + def handler(request: httpx.Request) -> httpx.Response: + if "bad" in str(request.url): + return httpx.Response(500, text="boom") + return httpx.Response(200, json={"ok": True}) + + client = self._make_client({"good": "http://good", "bad": "http://bad"}, handler) + with self.assertRaises(ParallelHTTPClientError): + await client.shutdown(stop_on_errors=True) + + +class TestRebuildAndDestroy(HttpClientTestBase): + async def test_rebuild_replaces_host_map(self): + client = self._make_client({"h1": "http://h1", "h2": "http://h2"}, _exec_handler) + client.rebuild({"h3": "http://h3"}) + outputs = await client.run_command("true") + self.assertEqual([o.host for o in outputs], ["h3"]) + + async def test_destroy_closes_client_and_allows_lazy_recreation(self): + client = self._make_client({"h1": "http://h1"}, _exec_handler) + await client.run_command("true") + self.assertIsNotNone(client._client) + await client.destroy() + self.assertIsNone(client._client) + # a call after destroy() lazily recreates the client rather than failing + outputs = await client.run_command("true") + self.assertEqual(outputs[0].exit_code, 0) + + async def test_async_context_manager_destroys_on_exit(self): + client = ParallelHTTPClient({"h1": "http://h1"}, TOKEN, transport=httpx.MockTransport(_exec_handler)) + async with client as ctx_client: + self.assertIs(ctx_client, client) + await client.run_command("true") + self.assertIsNotNone(client._client) + self.assertIsNone(client._client) + + +class TestHostOutput(unittest.TestCase): + def test_is_a_plain_dataclass_not_pssh_output(self): + output = HostOutput(host="h1", stdout=["a"], stderr=["b"], exit_code=0, exception=None) + self.assertEqual(output.host, "h1") + self.assertEqual(output.stdout, ["a"]) + self.assertEqual(output.stderr, ["b"]) + self.assertEqual(output.exit_code, 0) + self.assertIsNone(output.exception) + + +if __name__ == "__main__": + unittest.main() From bed8ee326a2a3194cdf1bb8719e4fd703432e5f8 Mon Sep 17 00:00:00 2001 From: speriaswamy-amd Date: Wed, 26 Aug 2026 15:28:05 -0400 Subject: [PATCH 6/9] Handle hardcoded output mode, env, cwd, output path --- cvs/core/agent/http_client.py | 52 +++++-- cvs/core/agent/unittests/test_http_client.py | 149 +++++++++++++++++++ 2 files changed, 191 insertions(+), 10 deletions(-) diff --git a/cvs/core/agent/http_client.py b/cvs/core/agent/http_client.py index 92c6fe6ba..b6ec31e3b 100644 --- a/cvs/core/agent/http_client.py +++ b/cvs/core/agent/http_client.py @@ -8,7 +8,6 @@ import asyncio import uuid from dataclasses import dataclass -from pathlib import Path import httpx @@ -104,8 +103,19 @@ async def destroy(self) -> None: self._client = None def _build_exec_requests( - self, cmd: str, host_args: list[str] | None, read_timeout: float | None + self, + cmd: str, + host_args: list[str] | None, + read_timeout: float | None, + env: dict[str, str] | None, + inactivity_timeout: float | None, + output_mode: messages.ExecOutputMode, ) -> dict[str, messages.ExecRequest]: + # Imported here, not at module level: cvs.core.run_layout pulls in cvs/core/__init__.py's + # orchestrator factory, which reaches back into cvs/core/agent/ in ways that risk a cycle + # (same reasoning as cvs/lib/utils_lib.py's lazy import of RunLayout). + from cvs.core.run_layout import RunLayout + hosts = list(self._agent_urls) if host_args is not None: if len(host_args) != len(hosts): @@ -113,20 +123,38 @@ def _build_exec_requests( commands = [cmd % args for args in host_args] else: commands = [cmd] * len(hosts) + run_dir = RunLayout.get().run_dir + out_dir = None + if output_mode == messages.ExecOutputMode.FILE: + out_dir = run_dir / "exec_output" + out_dir.mkdir(parents=True, exist_ok=True) return { host: messages.ExecRequest( cmd=command, - env={}, - cwd=Path.cwd(), + env=env or {}, + cwd=run_dir, timeout=round(read_timeout) if read_timeout is not None else None, - inactivity_timeout=None, + inactivity_timeout=round(inactivity_timeout) if inactivity_timeout is not None else None, cmd_id=uuid.uuid4().hex, - out_path=None, - output_mode=messages.ExecOutputMode.INLINE, + out_path=out_dir, + output_mode=output_mode, ) for host, command in zip(hosts, commands) } + async def _collect_output(self, exec_response: messages.ExecResponse) -> tuple[list[str], list[str]]: + '''FILE mode ships only a tail preview inline; the full output lives on the shared FS at + stdout_path/stderr_path, so read it back here to give callers the same list[str] shape + regardless of which output_mode produced the response (INLINE/EXIT_CODE_ONLY never set + stdout_path, so this falls through to the inline fields for those unchanged).''' + if exec_response.stdout_path is not None and exec_response.stderr_path is not None: + stdout_text, stderr_text = await asyncio.gather( + asyncio.to_thread(exec_response.stdout_path.read_text), + asyncio.to_thread(exec_response.stderr_path.read_text), + ) + return stdout_text.splitlines(), stderr_text.splitlines() + return exec_response.stdout or [], exec_response.stderr or [] + async def _run_one( self, client: httpx.AsyncClient, @@ -143,12 +171,13 @@ async def _run_one( ) response.raise_for_status() exec_response = messages.parse_message(messages.ExecResponse, response.text) + stdout, stderr = await self._collect_output(exec_response) except Exception as exc: # noqa: BLE001 - captured per-host so one bad host doesn't sink the others return HostOutput(host=host, stdout=[], stderr=[], exit_code=None, exception=_classify_exception(exc)) return HostOutput( host=host, - stdout=exec_response.stdout or [], - stderr=exec_response.stderr or [], + stdout=stdout, + stderr=stderr, exit_code=exec_response.exit_code, exception=None, ) @@ -159,8 +188,11 @@ async def run_command( stop_on_errors: bool = True, read_timeout: float | None = None, host_args: list[str] | None = None, + env: dict[str, str] | None = None, + inactivity_timeout: float | None = None, + output_mode: messages.ExecOutputMode = messages.ExecOutputMode.INLINE, ) -> list[HostOutput]: - requests = self._build_exec_requests(cmd, host_args, read_timeout) + requests = self._build_exec_requests(cmd, host_args, read_timeout, env, inactivity_timeout, output_mode) client = self._get_client() outputs = await asyncio.gather( *( diff --git a/cvs/core/agent/unittests/test_http_client.py b/cvs/core/agent/unittests/test_http_client.py index 1fe27b480..8092f021d 100644 --- a/cvs/core/agent/unittests/test_http_client.py +++ b/cvs/core/agent/unittests/test_http_client.py @@ -9,7 +9,9 @@ # fan-out, host_args/stop_on_errors semantics, exception classification, and session lifecycle. import json +import tempfile import unittest +from pathlib import Path import httpx @@ -21,6 +23,7 @@ ParallelHTTPClient, ParallelHTTPClientError, ) +from cvs.core.run_layout import RunLayout TOKEN = "test-token-123" @@ -43,6 +46,16 @@ def _exec_handler(request: httpx.Request) -> httpx.Response: class HttpClientTestBase(unittest.IsolatedAsyncioTestCase): + def setUp(self): + # run_command's cwd/out_path derive from RunLayout; point it at a throwaway tempdir so + # tests don't create real cvs_runs/ directories under the repo (see test_utils_lib.py's + # TestResolveRunDirPlaceholder for the same pattern). + RunLayout._reset() + self.addCleanup(RunLayout._reset) + self.tmp = tempfile.TemporaryDirectory() + self.addCleanup(self.tmp.cleanup) + RunLayout.get(self.tmp.name) + def _make_client(self, agent_urls: dict[str, str], handler, token: str = TOKEN, **kwargs) -> ParallelHTTPClient: client = ParallelHTTPClient(agent_urls, token, transport=httpx.MockTransport(handler), **kwargs) self.addAsyncCleanup(client.destroy) @@ -149,6 +162,61 @@ def handler(request: httpx.Request) -> httpx.Response: await client.run_command("true", read_timeout=2.7) self.assertEqual(seen_requests[0].timeout, 3) + async def test_inactivity_timeout_is_rounded_to_int_for_the_wire_request(self): + seen_requests: list[messages.ExecRequest] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(messages.parse_message(messages.ExecRequest, request.content.decode())) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.run_command("true", inactivity_timeout=4.4) + self.assertEqual(seen_requests[0].inactivity_timeout, 4) + + async def test_inactivity_timeout_defaults_to_none(self): + seen_requests: list[messages.ExecRequest] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(messages.parse_message(messages.ExecRequest, request.content.decode())) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.run_command("true") + self.assertIsNone(seen_requests[0].inactivity_timeout) + + async def test_env_is_passed_through_to_the_exec_request(self): + seen_requests: list[messages.ExecRequest] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(messages.parse_message(messages.ExecRequest, request.content.decode())) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.run_command("true", env={"FOO": "bar"}) + self.assertEqual(seen_requests[0].env, {"FOO": "bar"}) + + async def test_env_defaults_to_empty_dict(self): + seen_requests: list[messages.ExecRequest] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(messages.parse_message(messages.ExecRequest, request.content.decode())) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.run_command("true") + self.assertEqual(seen_requests[0].env, {}) + + async def test_cwd_defaults_to_the_run_layout_run_dir_not_a_local_path(self): + seen_requests: list[messages.ExecRequest] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(messages.parse_message(messages.ExecRequest, request.content.decode())) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.run_command("true") + self.assertEqual(seen_requests[0].cwd, RunLayout.get().run_dir) + async def test_client_is_reused_across_calls(self): client = self._make_client({"h1": "http://h1"}, _exec_handler) await client.run_command("true") @@ -157,6 +225,87 @@ async def test_client_is_reused_across_calls(self): self.assertIs(client._client, first_client) +def _file_mode_handler(request: httpx.Request) -> httpx.Response: + '''Simulates the agent's FILE output_mode: writes the full output to out_path on the (here, + tempdir-backed) shared FS and returns only a short tail preview inline, like http_agent.py does.''' + body = json.loads(request.content) + out_dir = Path(body["out_path"]) + stdout_path = out_dir / f"{body['cmd_id']}.stdout" + stderr_path = out_dir / f"{body['cmd_id']}.stderr" + full_stdout = "\n".join(f"line{i}" for i in range(20)) + stdout_path.write_text(full_stdout) + stderr_path.write_text("err-line") + return httpx.Response( + 200, + json={ + "exit_code": 0, + "stdout": full_stdout.splitlines()[-2:], + "stderr": ["err-line"], + "stdout_path": str(stdout_path), + "stderr_path": str(stderr_path), + "truncated": None, + "timed_out": False, + }, + ) + + +class TestOutputMode(HttpClientTestBase): + async def test_defaults_to_inline_and_does_not_set_out_path(self): + seen_requests: list[messages.ExecRequest] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(messages.parse_message(messages.ExecRequest, request.content.decode())) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.run_command("true") + self.assertEqual(seen_requests[0].output_mode, messages.ExecOutputMode.INLINE) + self.assertIsNone(seen_requests[0].out_path) + + async def test_file_mode_sets_out_path_under_the_run_layout_run_dir(self): + seen_requests: list[messages.ExecRequest] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(messages.parse_message(messages.ExecRequest, request.content.decode())) + return _file_mode_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.run_command("true", output_mode=messages.ExecOutputMode.FILE) + self.assertEqual(seen_requests[0].out_path, RunLayout.get().run_dir / "exec_output") + self.assertTrue(seen_requests[0].out_path.is_dir()) + + async def test_file_mode_returns_full_output_not_just_the_inline_preview(self): + client = self._make_client({"h1": "http://h1"}, _file_mode_handler) + outputs = await client.run_command("true", output_mode=messages.ExecOutputMode.FILE) + self.assertEqual(outputs[0].stdout, [f"line{i}" for i in range(20)]) + self.assertEqual(outputs[0].stderr, ["err-line"]) + + async def test_exit_code_only_mode_is_sent_through(self): + seen_requests: list[messages.ExecRequest] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(messages.parse_message(messages.ExecRequest, request.content.decode())) + return httpx.Response( + 200, + json={ + "exit_code": 0, + "stdout": None, + "stderr": None, + "stdout_path": None, + "stderr_path": None, + "truncated": None, + "timed_out": False, + }, + ) + + client = self._make_client({"h1": "http://h1"}, handler) + outputs = await client.run_command("true", output_mode=messages.ExecOutputMode.EXIT_CODE_ONLY) + self.assertEqual(seen_requests[0].output_mode, messages.ExecOutputMode.EXIT_CODE_ONLY) + self.assertEqual(outputs[0].stdout, []) + self.assertEqual(outputs[0].stderr, []) + self.assertEqual(outputs[0].exit_code, 0) + + class TestExceptionClassification(HttpClientTestBase): async def test_connection_failure_is_classified_as_connection_error(self): def handler(request: httpx.Request) -> httpx.Response: From 8f6c8be143b7cb50a7aca6913351de2fc333451e Mon Sep 17 00:00:00 2001 From: speriaswamy-amd Date: Tue, 1 Sep 2026 15:20:58 -0400 Subject: [PATCH 7/9] Addressing PR comments --- cvs/core/agent/http_agent.py | 10 +- cvs/core/agent/http_client.py | 95 ++++++++-- cvs/core/agent/messages.py | 6 + cvs/core/agent/unittests/test_http_client.py | 188 +++++++++++++++++++ 4 files changed, 280 insertions(+), 19 deletions(-) diff --git a/cvs/core/agent/http_agent.py b/cvs/core/agent/http_agent.py index 44bc74590..f58072ff5 100644 --- a/cvs/core/agent/http_agent.py +++ b/cvs/core/agent/http_agent.py @@ -18,7 +18,6 @@ from . import messages FILE_MODE_PREVIEW_LINES = 20 -TERMINATE_GRACE_PERIOD_SECONDS = 10.0 def _read_secret(file_path: Path) -> str: @@ -154,7 +153,7 @@ async def _communicate_with_timeouts( watchdog.cancel() await asyncio.gather(work, *([watchdog] if watchdog else []), return_exceptions=True) if timed_out: - await _terminate_process_group(process, TERMINATE_GRACE_PERIOD_SECONDS) + await _terminate_process_group(process, messages.TERMINATE_GRACE_PERIOD_SECONDS) return b"".join(stdout_chunks), b"".join(stderr_chunks), timed_out @@ -188,7 +187,7 @@ async def _run_cmd(request: messages.ExecRequest, registry: ProcessRegistry) -> try: await asyncio.wait_for(process.wait(), timeout=request.timeout) except asyncio.TimeoutError: - await _terminate_process_group(process, TERMINATE_GRACE_PERIOD_SECONDS) + await _terminate_process_group(process, messages.TERMINATE_GRACE_PERIOD_SECONDS) timed_out = True finally: await registry.unregister(request.cmd_id) @@ -329,7 +328,10 @@ async def run_shutdown(http_request: Request) -> messages.ShutdownResponse: registry: ProcessRegistry = http_request.app.state.process_registry processes = registry.snapshot() await asyncio.gather( - *(_terminate_process_group(process, TERMINATE_GRACE_PERIOD_SECONDS) for process in processes.values()) + *( + _terminate_process_group(process, messages.TERMINATE_GRACE_PERIOD_SECONDS) + for process in processes.values() + ) ) # Self-signal rather than depending on a Server reference: uvicorn installs a SIGTERM # handler that drains in-flight requests (this one included) before exiting. diff --git a/cvs/core/agent/http_client.py b/cvs/core/agent/http_client.py index b6ec31e3b..5a888a137 100644 --- a/cvs/core/agent/http_client.py +++ b/cvs/core/agent/http_client.py @@ -8,11 +8,41 @@ import asyncio import uuid from dataclasses import dataclass +from pathlib import Path import httpx from . import messages +# /v1/exec returns only after the process finishes. A timed-out process then spends +# TERMINATE_GRACE_PERIOD_SECONDS in SIGTERM-then-SIGKILL before ExecResponse can be sent; +# the extra buffer covers response serialization and FILE-mode writes on the shared FS. +_EXEC_RESPONSE_BUFFER_SECONDS = 1.0 +# Fallback read deadline for requests that carry no execution cost of their own (health, and any +# future endpoint). Matches httpx's implicit default, stated explicitly so it can't drift silently. +_DEFAULT_READ_TIMEOUT_SECONDS = 5.0 +# Shutdown waits out the agent's process-group termination grace before returning. +_SHUTDOWN_READ_TIMEOUT_SECONDS = messages.TERMINATE_GRACE_PERIOD_SECONDS + _EXEC_RESPONSE_BUFFER_SECONDS + + +def _exec_http_read_timeout(read_timeout: float | None) -> float | None: + '''HTTP read deadline for /v1/exec: the agent's process deadline plus termination grace.''' + if read_timeout is None: + return None + return round(read_timeout) + messages.TERMINATE_GRACE_PERIOD_SECONDS + _EXEC_RESPONSE_BUFFER_SECONDS + + +def _validated_exec_output_path(reported: Path | None, out_dir: Path, cmd_id: str, stream: str) -> Path: + '''Accept a FILE-mode path only if it resolves to /.stdout|stderr.''' + expected_name = f"{cmd_id}.{stream}" + if reported is None: + raise HTTPProtocolError(f"FILE-mode response omitted {stream} path") + expected_parent = out_dir.resolve() + resolved = reported.resolve() + if resolved.parent != expected_parent or resolved.name != expected_name: + raise HTTPProtocolError(f"FILE-mode {stream} path {reported} is not {expected_parent / expected_name}") + return resolved + @dataclass class HostOutput: @@ -21,6 +51,8 @@ class HostOutput: stderr: list[str] exit_code: int | None exception: Exception | None + timed_out: bool = False + truncated: bool | None = None class ParallelHTTPClientError(Exception): @@ -87,14 +119,31 @@ async def __aexit__(self, *exc_info) -> None: def _auth_header(self) -> dict[str, str]: return {messages.AUTH_HEADER: f"{messages.AUTH_SCHEME} {self._token}"} + def _http_timeout(self, read_timeout: float | None) -> httpx.Timeout: + return httpx.Timeout(read_timeout, connect=self._connect_timeout) + + def _pool_limits(self) -> httpx.Limits: + # httpx caps the pool at 100 connections by default, which would serialize the tail of a + # fan-out on a larger cluster (and turn into PoolTimeout once a request deadline is set). + # Concurrency here is already bounded by the host count - one connection per agent - so the + # pool needs no bound of its own, and staying unbounded keeps rebuild() to a larger host set + # from having to tear down a live pool and lose its keep-alives. + return httpx.Limits(max_connections=None, max_keepalive_connections=None) + def _get_client(self) -> httpx.AsyncClient: if self._client is None: - self._client = httpx.AsyncClient(headers=self._auth_header(), transport=self._transport) + self._client = httpx.AsyncClient( + headers=self._auth_header(), + transport=self._transport, + timeout=self._http_timeout(_DEFAULT_READ_TIMEOUT_SECONDS), + limits=self._pool_limits(), + ) return self._client def rebuild(self, agent_urls: dict[str, str]) -> None: '''Replace the host map, e.g. to drop hosts pruned after a failed health check. The shared - client's connection pool needs no action: idle connections to removed hosts simply age out.''' + client's connection pool needs no action either way: idle connections to removed hosts age + out, and the pool is unbounded so an added host opens a connection without evicting anyone.''' self._agent_urls = dict(agent_urls) async def destroy(self) -> None: @@ -142,18 +191,27 @@ def _build_exec_requests( for host, command in zip(hosts, commands) } - async def _collect_output(self, exec_response: messages.ExecResponse) -> tuple[list[str], list[str]]: + async def _collect_output( + self, request: messages.ExecRequest, exec_response: messages.ExecResponse + ) -> tuple[list[str], list[str]]: '''FILE mode ships only a tail preview inline; the full output lives on the shared FS at stdout_path/stderr_path, so read it back here to give callers the same list[str] shape regardless of which output_mode produced the response (INLINE/EXIT_CODE_ONLY never set stdout_path, so this falls through to the inline fields for those unchanged).''' - if exec_response.stdout_path is not None and exec_response.stderr_path is not None: + if exec_response.stdout_path is None and exec_response.stderr_path is None: + return exec_response.stdout or [], exec_response.stderr or [] + if request.out_path is None: + raise HTTPProtocolError("agent returned FILE-mode paths but no out_path was requested") + stdout_path = _validated_exec_output_path(exec_response.stdout_path, request.out_path, request.cmd_id, "stdout") + stderr_path = _validated_exec_output_path(exec_response.stderr_path, request.out_path, request.cmd_id, "stderr") + try: stdout_text, stderr_text = await asyncio.gather( - asyncio.to_thread(exec_response.stdout_path.read_text), - asyncio.to_thread(exec_response.stderr_path.read_text), + asyncio.to_thread(stdout_path.read_text), + asyncio.to_thread(stderr_path.read_text), ) - return stdout_text.splitlines(), stderr_text.splitlines() - return exec_response.stdout or [], exec_response.stderr or [] + except OSError as exc: + raise HTTPProtocolError(f"failed to read FILE-mode output: {exc}") from exc + return stdout_text.splitlines(), stderr_text.splitlines() async def _run_one( self, @@ -166,12 +224,12 @@ async def _run_one( try: response = await client.post( f"{url}{messages.EXEC_PATH}", - content=request.model_dump_json(), - timeout=httpx.Timeout(read_timeout, connect=self._connect_timeout), + json=request.model_dump(mode="json"), + timeout=self._http_timeout(_exec_http_read_timeout(read_timeout)), ) response.raise_for_status() exec_response = messages.parse_message(messages.ExecResponse, response.text) - stdout, stderr = await self._collect_output(exec_response) + stdout, stderr = await self._collect_output(request, exec_response) except Exception as exc: # noqa: BLE001 - captured per-host so one bad host doesn't sink the others return HostOutput(host=host, stdout=[], stderr=[], exit_code=None, exception=_classify_exception(exc)) return HostOutput( @@ -180,6 +238,8 @@ async def _run_one( stderr=stderr, exit_code=exec_response.exit_code, exception=None, + timed_out=exec_response.timed_out, + truncated=exec_response.truncated, ) async def run_command( @@ -207,12 +267,13 @@ async def run_command( raise ParallelHTTPClientError(f"{len(failed)} host(s) failed: {details}") return outputs - async def _fan_out(self, method: str, path: str, stop_on_errors: bool) -> dict[str, bool]: + async def _fan_out(self, method: str, path: str, stop_on_errors: bool, read_timeout: float) -> dict[str, bool]: client = self._get_client() + timeout = self._http_timeout(read_timeout) async def call(host: str, url: str) -> tuple[str, bool | Exception]: try: - response = await client.request(method, f"{url}{path}") + response = await client.request(method, f"{url}{path}", timeout=timeout) response.raise_for_status() except Exception as exc: # noqa: BLE001 - captured per-host, reported rather than raised return host, _classify_exception(exc) @@ -228,10 +289,14 @@ async def call(host: str, url: str) -> tuple[str, bool | Exception]: async def health(self) -> dict[str, bool]: '''Liveness probe per host; never raises regardless of failures - an unreachable host is the answer this call exists to produce (feeds rebuild()'s pruning decision), not an error.''' - return await self._fan_out("GET", messages.HEALTH_PATH, stop_on_errors=False) + return await self._fan_out( + "GET", messages.HEALTH_PATH, stop_on_errors=False, read_timeout=_DEFAULT_READ_TIMEOUT_SECONDS + ) async def shutdown(self, stop_on_errors: bool = False) -> dict[str, bool]: '''Ask every host's agent to terminate its spawned processes and exit. Defaults to best-effort (stop_on_errors=False), unlike run_command: one already-dead straggler during cleanup shouldn't stop the rest from being told to shut down.''' - return await self._fan_out("POST", messages.SHUTDOWN_PATH, stop_on_errors) + return await self._fan_out( + "POST", messages.SHUTDOWN_PATH, stop_on_errors, read_timeout=_SHUTDOWN_READ_TIMEOUT_SECONDS + ) diff --git a/cvs/core/agent/messages.py b/cvs/core/agent/messages.py index f6e55d351..26055f2f5 100644 --- a/cvs/core/agent/messages.py +++ b/cvs/core/agent/messages.py @@ -28,6 +28,12 @@ # Limits constants MAX_INLINE_RESPONSE_BYTES = 4 * 1024 * 1024 # INLINE output beyond this is truncated +# Timing constants +# How long the agent gives a process group between SIGTERM and SIGKILL. Part of the protocol, not +# an agent-private detail: /v1/exec only responds once termination finishes, so a caller's HTTP read +# deadline has to allow for this on top of the requested cmd timeout. +TERMINATE_GRACE_PERIOD_SECONDS = 10.0 + class RegisterRequest(BaseModel): '''Data model for a agent to register itself with the server''' diff --git a/cvs/core/agent/unittests/test_http_client.py b/cvs/core/agent/unittests/test_http_client.py index 8092f021d..65708bc54 100644 --- a/cvs/core/agent/unittests/test_http_client.py +++ b/cvs/core/agent/unittests/test_http_client.py @@ -16,12 +16,15 @@ import httpx from cvs.core.agent import messages +from cvs.core.agent.http_agent import create_app from cvs.core.agent.http_client import ( HostOutput, HTTPConnectionError, HTTPProtocolError, ParallelHTTPClient, ParallelHTTPClientError, + _SHUTDOWN_READ_TIMEOUT_SECONDS, + _exec_http_read_timeout, ) from cvs.core.run_layout import RunLayout @@ -72,6 +75,8 @@ async def test_returns_host_output_per_host(self): self.assertEqual(output.stderr, []) self.assertEqual(output.exit_code, 0) self.assertIsNone(output.exception) + self.assertFalse(output.timed_out) + self.assertFalse(output.truncated) async def test_sends_bearer_auth_header(self): seen_headers = [] @@ -224,6 +229,56 @@ async def test_client_is_reused_across_calls(self): await client.run_command("true") self.assertIs(client._client, first_client) + async def test_exec_request_uses_json_content_type(self): + seen_content_types = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_content_types.append(request.headers.get("content-type")) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.run_command("true") + self.assertTrue(seen_content_types) + self.assertIn("application/json", seen_content_types[0]) + + async def test_http_read_timeout_includes_agent_termination_grace(self): + seen_timeouts = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_timeouts.append(request.extensions.get("timeout")) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler, connect_timeout=2.0) + await client.run_command("true", read_timeout=1) + timeout = seen_timeouts[0] + expected_read = _exec_http_read_timeout(1) + self.assertEqual(timeout["connect"], 2.0) + self.assertEqual(timeout["read"], expected_read) + self.assertGreater(timeout["read"], 1) + self.assertGreaterEqual(timeout["read"], 1 + messages.TERMINATE_GRACE_PERIOD_SECONDS) + + async def test_timed_out_and_truncated_are_preserved_on_host_output(self): + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + json={ + "exit_code": -15, + "stdout": ["partial"], + "stderr": [], + "stdout_path": None, + "stderr_path": None, + "truncated": True, + "timed_out": True, + }, + ) + + client = self._make_client({"h1": "http://h1"}, handler) + outputs = await client.run_command("true") + self.assertTrue(outputs[0].timed_out) + self.assertTrue(outputs[0].truncated) + self.assertEqual(outputs[0].exit_code, -15) + self.assertIsNone(outputs[0].exception) + def _file_mode_handler(request: httpx.Request) -> httpx.Response: '''Simulates the agent's FILE output_mode: writes the full output to out_path on the (here, @@ -280,6 +335,55 @@ async def test_file_mode_returns_full_output_not_just_the_inline_preview(self): self.assertEqual(outputs[0].stdout, [f"line{i}" for i in range(20)]) self.assertEqual(outputs[0].stderr, ["err-line"]) + async def test_file_mode_rejects_path_outside_expected_exec_output(self): + secret = Path(self.tmp.name) / "secret.txt" + secret.write_text("classified") + + def handler(request: httpx.Request) -> httpx.Response: + return httpx.Response( + 200, + json={ + "exit_code": 0, + "stdout": ["preview"], + "stderr": [], + "stdout_path": str(secret), + "stderr_path": str(secret), + "truncated": None, + "timed_out": False, + }, + ) + + client = self._make_client({"h1": "http://h1"}, handler) + outputs = await client.run_command("true", output_mode=messages.ExecOutputMode.FILE, stop_on_errors=False) + self.assertIsInstance(outputs[0].exception, HTTPProtocolError) + self.assertEqual(outputs[0].stdout, []) + self.assertEqual(secret.read_text(), "classified") + + async def test_file_mode_rejects_unexpected_filename_in_exec_output(self): + def handler(request: httpx.Request) -> httpx.Response: + out_dir = Path(json.loads(request.content)["out_path"]) + decoy = out_dir / "other.stdout" + decoy.write_text("nope") + stderr_path = out_dir / "other.stderr" + stderr_path.write_text("nope-err") + return httpx.Response( + 200, + json={ + "exit_code": 0, + "stdout": ["preview"], + "stderr": [], + "stdout_path": str(decoy), + "stderr_path": str(stderr_path), + "truncated": None, + "timed_out": False, + }, + ) + + client = self._make_client({"h1": "http://h1"}, handler) + outputs = await client.run_command("true", output_mode=messages.ExecOutputMode.FILE, stop_on_errors=False) + self.assertIsInstance(outputs[0].exception, HTTPProtocolError) + self.assertEqual(outputs[0].stdout, []) + async def test_exit_code_only_mode_is_sent_through(self): seen_requests: list[messages.ExecRequest] = [] @@ -368,6 +472,19 @@ def handler(request: httpx.Request) -> httpx.Response: await client.health() self.assertEqual(seen_paths, [messages.HEALTH_PATH]) + async def test_health_applies_connect_timeout_and_explicit_read_deadline(self): + seen_timeouts = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_timeouts.append(request.extensions.get("timeout")) + return httpx.Response(200, json={"ok": True}) + + client = self._make_client({"h1": "http://h1"}, handler, connect_timeout=2.0) + await client.health() + timeout = seen_timeouts[0] + self.assertEqual(timeout["connect"], 2.0) + self.assertEqual(timeout["read"], 5.0) + class TestShutdown(HttpClientTestBase): async def test_hits_shutdown_path_on_every_host(self): @@ -405,6 +522,20 @@ def handler(request: httpx.Request) -> httpx.Response: with self.assertRaises(ParallelHTTPClientError): await client.shutdown(stop_on_errors=True) + async def test_shutdown_read_timeout_covers_agent_termination_grace(self): + seen_timeouts = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_timeouts.append(request.extensions.get("timeout")) + return httpx.Response(200, json={"ok": True}) + + client = self._make_client({"h1": "http://h1"}, handler, connect_timeout=2.0) + await client.shutdown() + timeout = seen_timeouts[0] + self.assertEqual(timeout["connect"], 2.0) + self.assertEqual(timeout["read"], _SHUTDOWN_READ_TIMEOUT_SECONDS) + self.assertGreaterEqual(timeout["read"], messages.TERMINATE_GRACE_PERIOD_SECONDS) + class TestRebuildAndDestroy(HttpClientTestBase): async def test_rebuild_replaces_host_map(self): @@ -432,6 +563,61 @@ async def test_async_context_manager_destroys_on_exit(self): self.assertIsNone(client._client) +class TestConnectionPool(HttpClientTestBase): + def test_pool_is_unbounded_rather_than_capped_at_the_httpx_default_of_100(self): + client = ParallelHTTPClient({f"h{i}": f"http://h{i}" for i in range(101)}, TOKEN) + limits = client._pool_limits() + self.assertIsNone(limits.max_connections) + self.assertIsNone(limits.max_keepalive_connections) + + async def test_fan_out_reaches_every_host_past_the_httpx_default_of_100(self): + seen_urls = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_urls.append(str(request.url)) + return _exec_handler(request) + + client = self._make_client({f"h{i}": f"http://h{i}" for i in range(101)}, handler) + outputs = await client.run_command("true") + self.assertEqual(len(outputs), 101) + self.assertEqual(len(set(seen_urls)), 101) + + async def test_growing_the_host_set_keeps_the_existing_pooled_client(self): + client = self._make_client({f"h{i}": f"http://h{i}" for i in range(2)}, _exec_handler) + await client.run_command("true") + first_client = client._client + client.rebuild({f"h{i}": f"http://h{i}" for i in range(101)}) + outputs = await client.run_command("true") + self.assertIs(client._client, first_client) + self.assertEqual(len(outputs), 101) + + +class TestCreateAppIntegration(HttpClientTestBase): + async def test_run_command_posts_json_accepted_by_create_app(self): + agent_dir = Path(self.tmp.name) / "agent" + agent_dir.mkdir() + (agent_dir / messages.AUTH_TOKEN_FILENAME).write_text(TOKEN + "\n") + app = create_app( + agent_dir=agent_dir, + world_rank=0, + world_size=1, + own_hostname="rank0-host", + own_port=9000, + register_timeout=5.0, + ) + async with app.router.lifespan_context(app): + client = ParallelHTTPClient({"local": "http://testserver"}, TOKEN, transport=httpx.ASGITransport(app=app)) + try: + outputs = await client.run_command("echo hello") + finally: + await client.destroy() + self.assertEqual(len(outputs), 1) + self.assertIsNone(outputs[0].exception) + self.assertEqual(outputs[0].exit_code, 0) + self.assertEqual(outputs[0].stdout, ["hello"]) + self.assertFalse(outputs[0].timed_out) + + class TestHostOutput(unittest.TestCase): def test_is_a_plain_dataclass_not_pssh_output(self): output = HostOutput(host="h1", stdout=["a"], stderr=["b"], exit_code=0, exception=None) @@ -440,6 +626,8 @@ def test_is_a_plain_dataclass_not_pssh_output(self): self.assertEqual(output.stderr, ["b"]) self.assertEqual(output.exit_code, 0) self.assertIsNone(output.exception) + self.assertFalse(output.timed_out) + self.assertIsNone(output.truncated) if __name__ == "__main__": From 67ca1cc2ac835093f5c9b2069116c3191f686974 Mon Sep 17 00:00:00 2001 From: speriaswamy-amd Date: Tue, 1 Sep 2026 21:56:49 -0400 Subject: [PATCH 8/9] Keep a finite connect deadline when none is given, and avoid rounding sub-second timeouts to zero. Co-authored-by: Cursor --- cvs/core/agent/http_client.py | 46 +++++++++---- cvs/core/agent/unittests/test_http_client.py | 72 ++++++++++++++++++++ 2 files changed, 103 insertions(+), 15 deletions(-) diff --git a/cvs/core/agent/http_client.py b/cvs/core/agent/http_client.py index 5a888a137..8799f0694 100644 --- a/cvs/core/agent/http_client.py +++ b/cvs/core/agent/http_client.py @@ -18,18 +18,33 @@ # TERMINATE_GRACE_PERIOD_SECONDS in SIGTERM-then-SIGKILL before ExecResponse can be sent; # the extra buffer covers response serialization and FILE-mode writes on the shared FS. _EXEC_RESPONSE_BUFFER_SECONDS = 1.0 -# Fallback read deadline for requests that carry no execution cost of their own (health, and any -# future endpoint). Matches httpx's implicit default, stated explicitly so it can't drift silently. +# Read deadline for requests that carry no execution cost of their own (health, and any future +# endpoint). Matches httpx's implicit default, stated explicitly so it can't drift silently. _DEFAULT_READ_TIMEOUT_SECONDS = 5.0 +# Connect deadline when the caller supplies none. Deliberately independent of the read deadline: +# reaching an agent costs the same whether the command it runs takes a second or an hour. +_DEFAULT_CONNECT_TIMEOUT_SECONDS = 5.0 # Shutdown waits out the agent's process-group termination grace before returning. _SHUTDOWN_READ_TIMEOUT_SECONDS = messages.TERMINATE_GRACE_PERIOD_SECONDS + _EXEC_RESPONSE_BUFFER_SECONDS -def _exec_http_read_timeout(read_timeout: float | None) -> float | None: - '''HTTP read deadline for /v1/exec: the agent's process deadline plus termination grace.''' - if read_timeout is None: +def _agent_timeout_seconds(timeout: float | None, name: str) -> int | None: + '''Map a caller timeout onto ExecRequest's int seconds. round() alone turns (0, 0.5) into 0, + which the agent treats as an immediate kill; keep at least 1s for any positive value.''' + if timeout is None: return None - return round(read_timeout) + messages.TERMINATE_GRACE_PERIOD_SECONDS + _EXEC_RESPONSE_BUFFER_SECONDS + if timeout <= 0: + raise ValueError(f"{name} must be positive, got {timeout}") + return max(1, round(timeout)) + + +def _exec_http_read_timeout(agent_timeout: int | None) -> float | None: + '''HTTP read deadline for /v1/exec, derived from the deadline the agent itself was given: the + agent only answers once the process is done, and a timed-out one needs its termination grace + first. Takes the already-normalized ExecRequest.timeout so the two can't drift apart.''' + if agent_timeout is None: + return None + return agent_timeout + messages.TERMINATE_GRACE_PERIOD_SECONDS + _EXEC_RESPONSE_BUFFER_SECONDS def _validated_exec_output_path(reported: Path | None, out_dir: Path, cmd_id: str, stream: str) -> Path: @@ -120,7 +135,12 @@ def _auth_header(self) -> dict[str, str]: return {messages.AUTH_HEADER: f"{messages.AUTH_SCHEME} {self._token}"} def _http_timeout(self, read_timeout: float | None) -> httpx.Timeout: - return httpx.Timeout(read_timeout, connect=self._connect_timeout) + # httpx.Timeout(read, connect=None) disables the TCP deadline, leaving a black-holed host to + # the OS SYN timeout. Letting connect track read instead would hand a long-running command's + # deadline to the connect phase, which is back to waiting out the OS, so an omitted + # connect_timeout gets a fixed finite deadline of its own. + connect = self._connect_timeout if self._connect_timeout is not None else _DEFAULT_CONNECT_TIMEOUT_SECONDS + return httpx.Timeout(read_timeout, connect=connect) def _pool_limits(self) -> httpx.Limits: # httpx caps the pool at 100 connections by default, which would serialize the tail of a @@ -182,8 +202,8 @@ def _build_exec_requests( cmd=command, env=env or {}, cwd=run_dir, - timeout=round(read_timeout) if read_timeout is not None else None, - inactivity_timeout=round(inactivity_timeout) if inactivity_timeout is not None else None, + timeout=_agent_timeout_seconds(read_timeout, "read_timeout"), + inactivity_timeout=_agent_timeout_seconds(inactivity_timeout, "inactivity_timeout"), cmd_id=uuid.uuid4().hex, out_path=out_dir, output_mode=output_mode, @@ -219,13 +239,12 @@ async def _run_one( host: str, url: str, request: messages.ExecRequest, - read_timeout: float | None, ) -> HostOutput: try: response = await client.post( f"{url}{messages.EXEC_PATH}", json=request.model_dump(mode="json"), - timeout=self._http_timeout(_exec_http_read_timeout(read_timeout)), + timeout=self._http_timeout(_exec_http_read_timeout(request.timeout)), ) response.raise_for_status() exec_response = messages.parse_message(messages.ExecResponse, response.text) @@ -255,10 +274,7 @@ async def run_command( requests = self._build_exec_requests(cmd, host_args, read_timeout, env, inactivity_timeout, output_mode) client = self._get_client() outputs = await asyncio.gather( - *( - self._run_one(client, host, self._agent_urls[host], request, read_timeout) - for host, request in requests.items() - ) + *(self._run_one(client, host, self._agent_urls[host], request) for host, request in requests.items()) ) if stop_on_errors: failed = [output for output in outputs if output.exception is not None] diff --git a/cvs/core/agent/unittests/test_http_client.py b/cvs/core/agent/unittests/test_http_client.py index 65708bc54..b10004702 100644 --- a/cvs/core/agent/unittests/test_http_client.py +++ b/cvs/core/agent/unittests/test_http_client.py @@ -167,6 +167,24 @@ def handler(request: httpx.Request) -> httpx.Response: await client.run_command("true", read_timeout=2.7) self.assertEqual(seen_requests[0].timeout, 3) + async def test_subsecond_read_timeout_is_at_least_one_second_on_the_wire(self): + seen_requests: list[messages.ExecRequest] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(messages.parse_message(messages.ExecRequest, request.content.decode())) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.run_command("true", read_timeout=0.4) + self.assertEqual(seen_requests[0].timeout, 1) + + async def test_non_positive_read_timeout_raises_value_error(self): + client = self._make_client({"h1": "http://h1"}, _exec_handler) + with self.assertRaises(ValueError): + await client.run_command("true", read_timeout=0) + with self.assertRaises(ValueError): + await client.run_command("true", read_timeout=-1) + async def test_inactivity_timeout_is_rounded_to_int_for_the_wire_request(self): seen_requests: list[messages.ExecRequest] = [] @@ -178,6 +196,17 @@ def handler(request: httpx.Request) -> httpx.Response: await client.run_command("true", inactivity_timeout=4.4) self.assertEqual(seen_requests[0].inactivity_timeout, 4) + async def test_subsecond_inactivity_timeout_is_at_least_one_second_on_the_wire(self): + seen_requests: list[messages.ExecRequest] = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_requests.append(messages.parse_message(messages.ExecRequest, request.content.decode())) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.run_command("true", inactivity_timeout=0.4) + self.assertEqual(seen_requests[0].inactivity_timeout, 1) + async def test_inactivity_timeout_defaults_to_none(self): seen_requests: list[messages.ExecRequest] = [] @@ -257,6 +286,36 @@ def handler(request: httpx.Request) -> httpx.Response: self.assertGreater(timeout["read"], 1) self.assertGreaterEqual(timeout["read"], 1 + messages.TERMINATE_GRACE_PERIOD_SECONDS) + async def test_connect_deadline_does_not_scale_with_a_long_running_commands_read_deadline(self): + # A black-holed host has to fail at the connect deadline, not at whatever deadline the + # command it would have run happens to carry. + seen_timeouts = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_timeouts.append(request.extensions.get("timeout")) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.run_command("true", read_timeout=1) + await client.run_command("true", read_timeout=3600) + short, long_running = seen_timeouts + self.assertEqual(short["connect"], 5.0) + self.assertEqual(long_running["connect"], 5.0) + self.assertGreater(long_running["read"], 3600) + + async def test_unbounded_exec_still_uses_a_finite_connect_deadline(self): + seen_timeouts = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_timeouts.append(request.extensions.get("timeout")) + return _exec_handler(request) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.run_command("true") + timeout = seen_timeouts[0] + self.assertIsNone(timeout["read"]) + self.assertEqual(timeout["connect"], 5.0) + async def test_timed_out_and_truncated_are_preserved_on_host_output(self): def handler(request: httpx.Request) -> httpx.Response: return httpx.Response( @@ -485,6 +544,19 @@ def handler(request: httpx.Request) -> httpx.Response: self.assertEqual(timeout["connect"], 2.0) self.assertEqual(timeout["read"], 5.0) + async def test_omitted_connect_timeout_does_not_disable_health_connect_deadline(self): + seen_timeouts = [] + + def handler(request: httpx.Request) -> httpx.Response: + seen_timeouts.append(request.extensions.get("timeout")) + return httpx.Response(200, json={"ok": True}) + + client = self._make_client({"h1": "http://h1"}, handler) + await client.health() + timeout = seen_timeouts[0] + self.assertEqual(timeout["connect"], 5.0) + self.assertEqual(timeout["read"], 5.0) + class TestShutdown(HttpClientTestBase): async def test_hits_shutdown_path_on_every_host(self): From def2424c40e0ba3d65f0a18a39bc0f9fbd07cc2f Mon Sep 17 00:00:00 2001 From: speriaswamy-amd Date: Wed, 2 Sep 2026 14:08:51 -0400 Subject: [PATCH 9/9] Stop passing register_timeout into create_app so CI's merge with main succeeds. Co-authored-by: Cursor --- cvs/core/agent/unittests/test_http_client.py | 1 - 1 file changed, 1 deletion(-) diff --git a/cvs/core/agent/unittests/test_http_client.py b/cvs/core/agent/unittests/test_http_client.py index b10004702..ab377052e 100644 --- a/cvs/core/agent/unittests/test_http_client.py +++ b/cvs/core/agent/unittests/test_http_client.py @@ -675,7 +675,6 @@ async def test_run_command_posts_json_accepted_by_create_app(self): world_size=1, own_hostname="rank0-host", own_port=9000, - register_timeout=5.0, ) async with app.router.lifespan_context(app): client = ParallelHTTPClient({"local": "http://testserver"}, TOKEN, transport=httpx.ASGITransport(app=app))