Merge pull request '[core-be-agent] fix(#354): wire delegation-results consumer into a2a executor' (#358) from fix/354-a2a-delegation-auto-resume into main

Co-Authored-By: infra-sre

.gitea/workflows/publish-runtime.yml

@@ -139,6 +139,14 @@ jobs:
           /tmp/smoke/bin/python "$GITHUB_WORKSPACE/scripts/wheel_smoke.py"
 
       - name: Publish to PyPI
+        # working-directory matches the preceding Build/Verify steps. Without
+        # this, twine runs from the default workspace checkout dir where
+        # `dist/` doesn't exist and fails with:
+        #   ERROR InvalidDistribution: Cannot find file (or expand pattern): 'dist/*'
+        # Caught on the first-ever successful dispatch of this workflow
+        # (run 5097, 2026-05-11 02:08Z) — every other step in the publish
+        # job already had this working-directory; Publish was missing it.
+        working-directory: ${{ runner.temp }}/runtime-build
         env:
           # PYPI_TOKEN: repository secret scoped to molecule-ai-workspace-runtime.
           # Set via: Settings → Actions → Variables and Secrets → New Secret.

.github/workflows/ci.yml

@@ -365,7 +365,7 @@ jobs:
           cache: pip
           cache-dependency-path: workspace/requirements.txt
       - if: needs.changes.outputs.python == 'true'
-        run: pip install -r requirements.txt pytest pytest-asyncio pytest-cov
+        run: pip install -r requirements.txt pytest pytest-asyncio pytest-cov sqlalchemy>=2.0.0
       # Coverage flags + fail-under floor moved into workspace/pytest.ini
       # (issue #1817) so local `pytest` and CI use identical config.
       - if: needs.changes.outputs.python == 'true'

scripts/build_runtime_package.py

@@ -50,6 +50,7 @@ from pathlib import Path
 # without updating this set), which broke every workspace startup with
 # `ModuleNotFoundError: No module named 'transcript_auth'`.
 TOP_LEVEL_MODULES = {
+    "_sanitize_a2a",
     "a2a_cli",
     "a2a_client",
     "a2a_executor",

workspace/a2a_cli.py

@@ -25,11 +25,10 @@ _WORKSPACE_ID_raw = os.environ.get("WORKSPACE_ID")
 if not _WORKSPACE_ID_raw:
     raise RuntimeError("WORKSPACE_ID environment variable is required but not set")
 WORKSPACE_ID = _WORKSPACE_ID_raw
-# Platform URL: always host.docker.internal inside containers (Docker or not).
-# The if/else is kept structurally for historical context; both paths now
-# use the same default — the platform API is only reachable via the Docker
-# network mesh from inside a workspace container regardless of runtime env.
-PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
+if os.path.exists("/.dockerenv") or os.environ.get("DOCKER_VERSION"):
+    PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
+else:
+    PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://localhost:8080")
 
 
 async def discover(target_id: str) -> dict | None:

workspace/a2a_client.py

@@ -26,11 +26,10 @@ _WORKSPACE_ID_raw = os.environ.get("WORKSPACE_ID")
 if not _WORKSPACE_ID_raw:
     raise RuntimeError("WORKSPACE_ID environment variable is required but not set")
 WORKSPACE_ID = _WORKSPACE_ID_raw
-# Platform URL: always host.docker.internal inside containers (Docker or not).
-# The if/else is kept structurally for historical context; both paths now
-# use the same default — the platform API is only reachable via the Docker
-# network mesh from inside a workspace container regardless of runtime env.
-PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
+if os.path.exists("/.dockerenv") or os.environ.get("DOCKER_VERSION"):
+    PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
+else:
+    PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://localhost:8080")
 
 # Cache workspace ID → name mappings (populated by list_peers calls)
 _peer_names: dict[str, str] = {}

workspace/a2a_executor.py

@@ -51,6 +51,7 @@ from shared_runtime import (
 from executor_helpers import (
     collect_outbound_files,
     extract_attached_files,
+    read_delegation_results,
 )
 from builtin_tools.telemetry import (
     A2A_TASK_ID,
@@ -215,6 +216,17 @@ class LangGraphA2AExecutor(AgentExecutor):
           3. Message(final_text)                      — terminal event
         """
         user_input = extract_message_text(context)
+        # Inject delegation results from prior turns. Heartbeat writes
+        # completed delegation rows to DELEGATION_RESULTS_FILE and sends
+        # a self-message to wake the agent; this consumes the file and
+        # surfaces the results as context so the agent can act on them
+        # without needing an explicit check_task_status call.
+        # Results are prepended so they are visible even when the
+        # self-message text is overwritten by a subsequent user message.
+        pending_results = read_delegation_results()
+        if pending_results:
+            logger.info("A2A execute: injecting %d delegation result(s)", pending_results.count("\n") + 1)
+            user_input = f"[Delegation results available]\n{pending_results}\n\n{user_input}"
         # Pull attached files from A2A message parts (kind: "file") and
         # append a manifest to the prompt so the agent knows they exist.
         # LangGraph tools (filesystem, bash, skills) can then open the

workspace/builtin_tools/temporal_workflow.py

@@ -54,19 +54,6 @@ import httpx
 
 logger = logging.getLogger(__name__)
 
-
-def _platform_url() -> str:
-    """Return the platform URL, defaulting to host.docker.internal.
-
-    The workspace runtime always runs inside a Docker container, so
-    ``localhost`` refers to the container itself, not the platform host.
-    The platform API is only reachable via ``host.docker.internal`` from
-    within a workspace container, regardless of how the container was started.
-    The legacy non-Docker branch is removed (it would have returned
-    ``localhost:8080`` which is unreachable from inside the container).
-    """
-    return os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
-
 # ─────────────────────────────────────────────────────────────────────────────
 # Constants
 # ─────────────────────────────────────────────────────────────────────────────
@@ -92,12 +79,12 @@ async def _fetch_latest_checkpoint(workspace_id: str) -> Optional[dict]:
         workspace_id: The workspace to query.
 
     Reads:
-        PLATFORM_URL  Platform base URL (default ``http://host.docker.internal:8080``).
+        PLATFORM_URL  Platform base URL (default ``http://localhost:8080``).
     """
     try:
         from platform_auth import auth_headers as _auth_headers  # type: ignore[import]
 
-        platform_url = _platform_url()
+        platform_url = os.environ.get("PLATFORM_URL", "http://localhost:8080")
         url = f"{platform_url}/workspaces/{workspace_id}/checkpoints/latest"
         async with httpx.AsyncClient(timeout=5.0) as client:
             resp = await client.get(url, headers=_auth_headers())
@@ -138,12 +125,12 @@ async def _save_checkpoint(
         payload:       Optional JSON-serialisable dict stored as JSONB.
 
     Reads:
-        PLATFORM_URL   Platform base URL (default ``http://host.docker.internal:8080``).
+        PLATFORM_URL   Platform base URL (default ``http://localhost:8080``).
     """
     try:
         from platform_auth import auth_headers as _auth_headers  # type: ignore[import]
 
-        platform_url = _platform_url()
+        platform_url = os.environ.get("PLATFORM_URL", "http://localhost:8080")
         url = f"{platform_url}/workspaces/{workspace_id}/checkpoints"
         body: dict = {
             "workflow_id": workflow_id,

workspace/consolidation.py

@@ -18,11 +18,10 @@ from platform_auth import auth_headers
 
 logger = logging.getLogger(__name__)
 
-# Platform URL: always host.docker.internal inside containers (Docker or not).
-# The if/else is kept structurally for historical context; both paths now
-# use the same default — the platform API is only reachable via the Docker
-# network mesh from inside a workspace container regardless of runtime env.
-PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
+if os.path.exists("/.dockerenv") or os.environ.get("DOCKER_VERSION"):
+    PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
+else:
+    PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://localhost:8080")
 _WORKSPACE_ID_raw = os.environ.get("WORKSPACE_ID")
 if not _WORKSPACE_ID_raw:
     raise RuntimeError("WORKSPACE_ID environment variable is required but not set")

workspace/coordinator.py

@@ -22,11 +22,10 @@ from policies.routing import build_team_routing_payload
 
 logger = logging.getLogger(__name__)
 
-# Platform URL: always host.docker.internal inside containers (Docker or not).
-# The if/else is kept structurally for historical context; both paths now
-# use the same default — the platform API is only reachable via the Docker
-# network mesh from inside a workspace container regardless of runtime env.
-PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
+if os.path.exists("/.dockerenv") or os.environ.get("DOCKER_VERSION"):
+    PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
+else:
+    PLATFORM_URL = os.environ.get("PLATFORM_URL", "http://localhost:8080")
 _WORKSPACE_ID_raw = os.environ.get("WORKSPACE_ID")
 if not _WORKSPACE_ID_raw:
     raise RuntimeError("WORKSPACE_ID environment variable is required but not set")

workspace/main.py

@@ -60,10 +60,10 @@ async def main():  # pragma: no cover
     config_path = os.environ.get("WORKSPACE_CONFIG_PATH", "/configs")
     # Docker-aware default — host.docker.internal resolves the platform service
     # from inside the Docker network mesh; falls back to localhost for local dev.
-    # Both branches now use the same default (architectural decision: the platform
-    # API is only reachable via host.docker.internal from within a workspace
-    # container, regardless of how the container was started).
-    platform_url = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
+    if os.path.exists("/.dockerenv") or os.environ.get("DOCKER_VERSION"):
+        platform_url = os.environ.get("PLATFORM_URL", "http://host.docker.internal:8080")
+    else:
+        platform_url = os.environ.get("PLATFORM_URL", "http://localhost:8080")
     awareness_config = get_awareness_config()
 
     # 0. Initialise OpenTelemetry (no-op if packages not installed)

workspace/plugins_registry/__init__.py

@@ -51,32 +51,6 @@ class AdaptorSource:
 
 def _load_module_from_path(module_name: str, path: Path):
     """Import a Python file by absolute path. Returns the module or None on failure."""
-
-    # KI-296: Before exec'ing plugin-adapter files (which import
-    # ``from plugins_registry import ...`` as a top-level name), register
-    # the molecule-runtime subpackage as ``plugins_registry`` in sys.modules.
-    # In the molecule-core workspace source this is already a top-level package,
-    # so the setdefault is a no-op. In the PyPI-installed runtime wheel
-    # (molecule-ai-workspace-runtime 0.1.129+), the package ships as
-    # ``molecule_runtime.plugins_registry`` and without this shim every
-    # plugin adapter would fail with ModuleNotFoundError.
-    import sys as _sys
-
-    if "plugins_registry" not in _sys.modules:
-        try:
-            _mr_pr = __import__("molecule_runtime.plugins_registry", fromlist=[""])
-            _sys.modules["plugins_registry"] = _mr_pr
-            # Also register submodules the adapters commonly import directly.
-            for _sub in ("builtins", "protocol", "raw_drop"):
-                _submod = getattr(_mr_pr, _sub, None)
-                if _submod is not None:
-                    _sys.modules[f"plugins_registry.{_sub}"] = _submod
-        except ImportError:
-            # molecule-runtime not installed (e.g. test environment with
-            # workspace/ on sys.path directly) — skip shim; the top-level
-            # workspace/plugins_registry package is already findable.
-            pass
-
     spec = importlib.util.spec_from_file_location(module_name, path)
     if spec is None or spec.loader is None:
         return None

workspace/tests/test_a2a_executor.py

@@ -1201,3 +1201,94 @@ async def test_terminal_error_routes_via_updater_failed():
     assert not eq._complete_calls, (
         "complete() should not fire when execute() raises"
     )
+
+
+# ---------------------------------------------------------------------------
+# Issue #354 — delegation results auto-resume gap
+# ---------------------------------------------------------------------------
+# heartbeat.py's _check_delegations writes completed delegation rows to
+# DELEGATION_RESULTS_FILE and sends a self-message to wake the agent.
+# read_delegation_results() in executor_helpers.py atomically reads+consumes
+# that file. The fix wires this consumer into _core_execute so the agent
+# receives delegation results as context in the next turn — closing the gap
+# where parallel delegate_task calls return after the SDK turn ends and the
+# agent has no way to discover the results.
+
+@pytest.mark.asyncio
+async def test_delegation_results_injected_into_user_input(monkeypatch):
+    """When delegation results exist, they are prepended to the user input
+    passed to the agent so the agent can act on them without an explicit
+    check_task_status call."""
+    import a2a_executor
+    from unittest.mock import patch
+
+    pending_results = (
+        "- [completed] Delegation abc123: Checked 3 issues\n"
+        "  Response: 3 open, 0 critical\n"
+        "- [failed] Delegation def456: Scan PR #352\n"
+        "  Error: peer workspace offline"
+    )
+
+    # Patch read_delegation_results at the module level where a2a_executor
+    # imported it so the _core_execute call picks it up.
+    with patch.object(a2a_executor, "read_delegation_results", return_value=pending_results):
+        agent = MagicMock()
+        agent.astream_events = MagicMock(return_value=_stream(_text_chunk("Got it")))
+        executor = LangGraphA2AExecutor(agent)
+
+        part = MagicMock()
+        part.text = "What's the status?"
+        context = _make_context([part], "ctx-deleg", task_id="task-deleg")
+        eq = _make_event_queue()
+        eq._complete_calls = []
+        eq._failed_calls = []
+
+        await executor.execute(context, eq)
+
+        # Verify the agent received the injected context
+        agent.astream_events.assert_called_once()
+        call_args = agent.astream_events.call_args
+        messages = call_args[0][0]["messages"]
+
+        # The last message should be a human turn with the injected context
+        human_turn = messages[-1]
+        assert human_turn[0] == "human"
+        # Must contain the delegation results marker
+        assert "[Delegation results available]" in human_turn[1]
+        # Must contain the completed delegation
+        assert "abc123" in human_turn[1]
+        assert "3 open" in human_turn[1]
+        # Must contain the failed delegation
+        assert "def456" in human_turn[1]
+        # Must contain the original user message
+        assert "What's the status?" in human_turn[1]
+
+
+@pytest.mark.asyncio
+async def test_no_delegation_results_no_injection(monkeypatch):
+    """When no delegation results exist, user input is passed through unchanged."""
+    import a2a_executor
+    from unittest.mock import patch
+
+    with patch.object(a2a_executor, "read_delegation_results", return_value=""):
+        agent = MagicMock()
+        agent.astream_events = MagicMock(return_value=_stream(_text_chunk("ok")))
+        executor = LangGraphA2AExecutor(agent)
+
+        part = MagicMock()
+        part.text = "Hello"
+        context = _make_context([part], "ctx-clean", task_id="task-clean")
+        eq = _make_event_queue()
+        eq._complete_calls = []
+        eq._failed_calls = []
+
+        await executor.execute(context, eq)
+
+        agent.astream_events.assert_called_once()
+        call_args = agent.astream_events.call_args
+        messages = call_args[0][0]["messages"]
+        human_turn = messages[-1]
+        assert human_turn[0] == "human"
+        # Must NOT contain the injection marker
+        assert "[Delegation results available]" not in human_turn[1]
+        assert human_turn[1] == "Hello"

workspace/tests/test_plugins_registry.py

@@ -325,44 +325,3 @@ def test_resolve_registry_missing_module_falls_through(monkeypatch, tmp_path: Pa
     monkeypatch.setattr(pr, "_REGISTRY_ROOT", tmp_path / "empty-registry")
     _, source = pr.resolve("demo-plugin", "test_runtime", plugin_root)
     assert source == AdaptorSource.RAW_DROP
-
-
-def test_load_module_from_path_registers_plugins_registry_sys_modules(tmp_path: Path):
-    """KI-296: _load_module_from_path registers ``plugins_registry`` in sys.modules
-    before exec'ing the adapter, so adapter files that do
-    ``from plugins_registry import ...`` resolve correctly when the runtime is
-    installed from the PyPI wheel (where the package ships as
-    ``molecule_runtime.plugins_registry`` rather than a top-level ``plugins_registry``).
-    """
-    import sys as _sys
-    import plugins_registry as pr
-
-    # Create a fake adapter that imports plugins_registry at top level.
-    adapter_file = tmp_path / "fake_runtime_adapter.py"
-    adapter_file.write_text(
-        "from plugins_registry import InstallContext  # noqa: F401\n"
-        "from plugins_registry.builtins import AgentskillsAdaptor as Adaptor  # noqa: F401\n"
-    )
-
-    # Evict any pre-existing sys.modules entries for the shim keys so the
-    # import inside _load_module_from_path actually runs.
-    _saved = {
-        k: _sys.modules.pop(k, None)
-        for k in (
-            "plugins_registry", "plugins_registry.builtins",
-            "plugins_registry.protocol", "plugins_registry.raw_drop",
-            "_plugin_adaptor.test.fake_runtime",
-        )
-    }
-
-    try:
-        result = pr._load_module_from_path("_plugin_adaptor.test.fake_runtime", adapter_file)
-        assert result is not None, "module should load without ImportError"
-        assert hasattr(result, "Adaptor"), "AgentskillsAdaptor alias should be in namespace"
-    finally:
-        # Restore sys.modules state.
-        for k, v in _saved.items():
-            if v is None:
-                _sys.modules.pop(k, None)
-            else:
-                _sys.modules[k] = v