feat(harness): seal a stopped call with its partial evidence, flag worker channel bypass and record shared records (TH-10189)

.github/workflows/sdk-smoke.yml

@@ -63,6 +63,7 @@ jobs:
           tests/test_call_originator_factory.py
           tests/test_retell_evidence.py
           tests/harness/test_process_runtime.py
+          tests/harness/test_multi_agent_compatibility_references.py
           tests/test_harness_sandbox_server.py
           --deselect tests/test_harness_sandbox_server.py::test_private_github_submission_becomes_hosted_job
           --deselect tests/test_harness_sandbox_server.py::test_public_github_branch_url_populates_job_ref

src/fi/alk/harness/bundle_author_v2.py

@@ -1983,6 +1983,97 @@ def _adk_interface_for(command: list[str], app_name: str) -> dict[str, Any]:
     }
 
 
+def _is_langgraph_server_command(command: list[str]) -> bool:
+    """Return whether argv starts the LangGraph API server (``langgraph dev`` or ``up``)."""
+
+    normalized = [str(item).strip() for item in command]
+    return any(
+        normalized[index] == "langgraph"
+        and normalized[index + 1] in {"dev", "up"}
+        for index in range(max(0, len(normalized) - 1))
+    )
+
+
+def _langgraph_server_interface(
+    root: Path, runtime: dict[str, Any]
+) -> dict[str, Any] | None:
+    """Compile the LangGraph server's threads and runs API into the generic JSON HTTP interface.
+
+    ``langgraph dev`` serves the graphs its ``langgraph.json`` declares through threads and runs,
+    never an OpenAI chat route. The assistant is the declared graph the written interface names
+    as its model, else the only graph, else the first one declared (the server's default).
+    """
+
+    command = [str(item) for item in (runtime.get("command") or [])]
+    if not _is_langgraph_server_command(command):
+        return None
+    configured = str(runtime.get("workdir") or "").strip()
+    workdir = (root / configured) if configured not in {"", ".", "/"} else root
+    config_name = "langgraph.json"
+    for index, token in enumerate(command):
+        if token == "--config" and index + 1 < len(command):
+            config_name = command[index + 1]
+        elif token.startswith("--config="):
+            config_name = token.partition("=")[2]
+    config = (workdir / config_name).resolve()
+    if not config.is_relative_to(root.resolve()) or not config.is_file():
+        return None
+    try:
+        graphs = json.loads(config.read_text(encoding="utf-8")).get("graphs")
+    except (OSError, ValueError, AttributeError):
+        return None
+    if not isinstance(graphs, dict) or not graphs:
+        return None
+    interface = runtime.get("interface")
+    named = interface.get("model") if isinstance(interface, dict) else None
+    assistant = named if named in graphs else next(iter(graphs))
+    return {
+        "kind": "http",
+        "protocol": "json_template",
+        "port": _command_port(command, default=2024),
+        "path": "/threads/{{langgraph_thread_id}}/runs/wait",
+        "health_path": "/ok",
+        "include_tools": False,
+        "setup_requests": [
+            {
+                "method": "POST",
+                "path": "/threads",
+                "body_template": {},
+                "accepted_statuses": [200],
+                "capture": {"langgraph_thread_id": "thread_id"},
+            }
+        ],
+        "request_template": {
+            "assistant_id": assistant,
+            "input": {
+                "messages": [{"role": "user", "content": "{{new_message_content}}"}]
+            },
+        },
+        "response_path": "messages.-1.content",
+    }
+
+
+def _native_server_runtime(root: Path, runtime: dict[str, Any]) -> dict[str, Any] | None:
+    """The runtime with the documented API of a source-owned framework server sealed as its
+    interface (``adk web``, ``langgraph dev``), or None when the command starts neither.
+
+    A LangGraph command without ``--port`` gets the server's default port written out, so each
+    world's port can replace it.
+    """
+
+    interface = _adk_web_interface(root, runtime) or _langgraph_server_interface(
+        root, runtime
+    )
+    if interface is None:
+        return None
+    command = [str(item) for item in (runtime.get("command") or [])]
+    if _is_langgraph_server_command(command) and not any(
+        token == "--port" or token.startswith("--port=") for token in command
+    ):
+        command = [*command, "--port", str(interface["port"])]
+    return {**runtime, "command": command, "interface": interface}
+
+
 def _submitted_command(process: SourceProcess, command: list[str]) -> list[str]:
     """Run a contract argv inside the dependency environment selected for the source."""
 
@@ -2320,6 +2411,25 @@ def _compose_environment(service: dict[str, Any]) -> dict[str, str]:
     return {}
 
 
+def _compose_service_running(root: Path, command: list[str]) -> dict[str, str]:
+    """The source's one Compose file and its one source-built service that runs exactly `command`, else nothing; a
+    file that publishes no ports names its agent this way only."""
+    found = [path for name in ("compose.yaml", "compose.yml", "docker-compose.yaml", "docker-compose.yml")
+             if (path := root / name).is_file()]
+    if len(found) != 1 or not command:
+        return {}
+    try:
+        services = (yaml.safe_load(found[0].read_text(encoding="utf-8")) or {}).get("services") or {}
+        matches = [
+            name
+            for name, service in services.items()
+            if isinstance(service, dict) and service.get("build") and _compose_command(service.get("command")) == command
+        ]
+    except (yaml.YAMLError, AttributeError, BundleAuthorError, OSError):
+        return {}
+    return {"compose_file": found[0].name, "service": matches[0]} if len(matches) == 1 else {}
+
+
 def resolve_environment_plan(
     source: str | Path,
     job: HarnessJob,
@@ -2374,11 +2484,19 @@ def resolve_environment_plan(
             contract_runtime or {},
             public_environment=declared_runtime_environment,
         )
+        # A tool service the agent calls over HTTP, which the source's own Compose file may ship as `tools-api`.
+        calls_tool_service = any(
+            dependency.get("kind", "").lower() == "service"
+            and dependency.get("engine", "").lower() == "http"
+            and dependency.get("used_by")
+            for dependency in (runtime_dependencies or [])
+        )
         if (
             not isinstance(selected, ComposeSelection)
             and runtime_model.execution == "native"
             and (
                 runtime_model.required_environment
+                or calls_tool_service
                 or any(
                     dependency.get("kind", "").lower()
                     in {"datastore", "database", "file", "queue"}
@@ -2392,6 +2510,7 @@ def resolve_environment_plan(
                 "execution": "auto",
                 "compose_file": "",
                 "service": "",
+                **(_compose_service_running(root, runtime_model.command) if calls_tool_service else {}),
             }
             try:
                 topology = select_runtime(
@@ -2401,7 +2520,13 @@ def resolve_environment_plan(
                 )
             except ValueError:
                 topology = None
-            if isinstance(topology, ComposeSelection):
+            # Only a tool service shipped as `tools-api` puts the world's door in front of the tools; for any other
+            # name the native choice stands, as before.
+            if isinstance(topology, ComposeSelection) and (
+                runtime_model.required_environment
+                or "tools-api" in topology.services
+                or not calls_tool_service
+            ):
                 required = runtime_model.required_environment
                 requirements_are_wired = all(
                     all(
@@ -2943,8 +3068,11 @@ def resolve_environment_plan(
                     "started_check": StartedCheck(port=True, timeout_seconds=180),
                 }
             )
-        if is_http_runtime and _is_adk_web_command(runtime_command):
-            # ADK exposes a normal ``--port`` seam. Wire it to the world's allocated port so a
+        if is_http_runtime and (
+            _is_adk_web_command(runtime_command)
+            or _is_langgraph_server_command(runtime_command)
+        ):
+            # ADK and LangGraph expose a normal ``--port`` seam. Wire it to the world's allocated port so a
             # multi-world chat run does not collapse to serial execution or collide on 8080.
             process = _consumable_source_process(
                 process, "PORT", f"{{{{PORT_{control_name}}}}}"
@@ -3165,16 +3293,15 @@ def resolve_environment_plan(
         for process in normalized_processes
     ]
 
-    # A voice agent's tool calls are traced by its LiveKit worker, never twice.
-    if (
-        not isinstance(selected, ComposeSelection)
-        and runtime_model.command
-        and str(contract_modality or "").strip().lower() != "voice"
-    ):
+    # A voice agent's tool calls are traced by its LiveKit worker, never twice. A Compose-selected agent gets the
+    # framework observation too (its hand-offs and per-worker tool calls), but not import-mode tool bindings.
+    if runtime_model.command and str(contract_modality or "").strip().lower() != "voice":
         trace_environment = _source_tool_trace_environment(
             root,
             runtime_model,
-            (tool_entrypoints or []) if contract_interface_kind == "command" else [],
+            (tool_entrypoints or [])
+            if contract_interface_kind == "command" and not isinstance(selected, ComposeSelection)
+            else [],
         )
         if trace_environment:
             processes = [
@@ -3588,16 +3715,15 @@ def author_bundle_v2(
         interface = runtime.get("interface") if isinstance(runtime, dict) else None
         if isinstance(interface, dict):
             contract_interface_kind = str(interface.get("kind") or "").strip().lower()
-            adk_interface = _adk_web_interface(source_root, runtime)
+            native_runtime = _native_server_runtime(source_root, runtime)
             if (
                 contract_modality == "chat"
-                and adk_interface is not None
+                and native_runtime is not None
                 and contract_interface_kind in {"", "http"}
             ):
-                # ADK's source-owned web API is deterministic; do not retain stale model-authored
-                # request shapes from older ADK releases.
-                runtime = dict(runtime)
-                runtime["interface"] = adk_interface
+                # A source-owned framework server's API is deterministic; do not retain stale
+                # model-authored request shapes (an OpenAI chat route for ``langgraph dev``).
+                runtime = native_runtime
                 contract_body = {**contract_body, "runtime": runtime}
                 contract_interface_kind = "http"
         elif (
@@ -3639,13 +3765,12 @@ def author_bundle_v2(
             and not isinstance(interface, dict)
             and runtime.get("command")
         ):
-            runtime = dict(runtime)
-            adk_interface = _adk_web_interface(source_root, runtime)
-            if adk_interface is not None:
-                # ``adk web`` is a persistent source-owned HTTP server, not a one-shot command.
-                # Launch it directly and describe its existing API rather than nesting it inside
-                # the subprocess bridge (which would make both processes bind the same port).
-                runtime["interface"] = adk_interface
+            native_runtime = _native_server_runtime(source_root, runtime)
+            if native_runtime is not None:
+                # ``adk web`` and ``langgraph dev`` are persistent source-owned HTTP servers, not
+                # one-shot commands. Launch them directly and describe their existing API rather
+                # than nesting them inside the subprocess bridge (both would bind the same port).
+                runtime = native_runtime
                 contract_interface_kind = "http"
             else:
                 raise BundleAuthorError(