fix(validate): report the env writer's progress on the event loop, which a Temporal heartbeat needs

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

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"))