From 231573fdd5face1ae7eeab83e639d95c7feb46b2 Mon Sep 17 00:00:00 2001 From: Edgar Arakelyan Date: Mon, 26 Oct 2026 10:57:29 -0700 Subject: [PATCH] fix(validate): report the env writer's progress on the event loop, which a Temporal heartbeat needs Co-Authored-By: Claude Opus 5.5 (1M context) --- src/agent_env/a2a_agent/validator.py | 7 +++++-- tst/unit/cli/a2a_agent_validate_test.py | 25 +++++++++++++++++++++++++ 2 files changed, 30 insertions(+), 2 deletions(-) diff --git a/src/agent_env/a2a_agent/validator.py b/src/agent_env/a2a_agent/validator.py index 6687debad..42ccac326 100644 --- a/src/agent_env/a2a_agent/validator.py +++ b/src/agent_env/a2a_agent/validator.py @@ -157,10 +157,13 @@ async def validate( Namespace routing is on throughout, so the env it deploys, ``A2AAgent.VALIDATION_ENV_ID``, can be the ``@local`` env an installed bundle's alias names, written first (``write_validation_env``) in a thread so a - service's event loop isn't held up. Returns the merged `verifications` dict from the run's task context. + service's event loop isn't held up. ``on_progress`` is called on this event loop throughout, the writer's lines + included. Returns the merged `verifications` dict from the run's task context. """ + loop = asyncio.get_running_loop() + report = on_progress or (lambda line: None) with namespace_routing(): - await asyncio.to_thread(write_validation_env, on_progress or (lambda line: None)) + await asyncio.to_thread(write_validation_env, lambda line: loop.call_soon_threadsafe(report, line)) return await A2AAgentValidator._run(agent, on_progress, litellm_api_key) @staticmethod diff --git a/tst/unit/cli/a2a_agent_validate_test.py b/tst/unit/cli/a2a_agent_validate_test.py index c44d7fafc..53d61bde7 100644 --- a/tst/unit/cli/a2a_agent_validate_test.py +++ b/tst/unit/cli/a2a_agent_validate_test.py @@ -1,7 +1,10 @@ """The A2A agent validator deploys the mail-mcp example's env by its alias, `mail`, writing it first when the store has no env of that id, with namespace routing on throughout so the alias's @local env resolves, in the CLI or a service.""" +import asyncio +import contextvars import importlib +import threading import pytest from click.testing import CliRunner @@ -73,6 +76,28 @@ async def run(agent, on_progress, litellm_api_key): assert seen == [("write", True), ("run", True)] and not namespace_routing_enabled() +@pytest.mark.asyncio +async def test_the_writers_progress_reaches_the_callback_on_the_event_loop_in_the_callers_context(monkeypatch): + """A service's callback can need both, as a Temporal activity's heartbeat does.""" + caller = contextvars.ContextVar("caller") + caller.set("activity") + loop_thread, lines = threading.get_ident(), [] + monkeypatch.setattr(validator_module, "write_validation_env", lambda on_progress: on_progress("writing it")) + + async def run(agent, on_progress, litellm_api_key): + return {} + + monkeypatch.setattr(validator_module.A2AAgentValidator, "_run", staticmethod(run)) + + def on_progress(line): + asyncio.get_running_loop() + lines.append((line, threading.get_ident() == loop_thread, caller.get(None))) + + await validator_module.A2AAgentValidator.validate(object(), on_progress=on_progress) + + assert lines == [("writing it", True, "activity")] + + def test_the_cli_says_why_the_env_couldnt_be_written(monkeypatch, local_stores, cli_routing): image = get_artifact_store().put_document(DockerImageArtifact( id="agent-image", description="d", image_name="ghcr.io/acme/agent:1", tar_gz_s3_url="s3://b/x"))