feat(subagent): derive child run context

coding-agent/src/coding_agent/app.py

@@ -123,6 +123,7 @@ def create_child_pipeline(
     max_steps_override: int | None = None,
     approval_mode_override: str | None = None,
     session_id_override: str | None = None,
+    run_id_override: str | None = None,
     agent_id_override: str | None = None,
     parent_run_id_override: str | None = None,
     context_budget: ContextBudget | None = None,
@@ -318,9 +319,15 @@ async def _memory_handler(directive: Any) -> None:
         if not session_id_override:
             raise ValueError("session_id_override must be None or a non-empty string")
         session_id = session_id_override
+    if run_id_override is None:
+        run_id = uuid.uuid4().hex
+    else:
+        if not run_id_override:
+            raise ValueError("run_id_override must be None or a non-empty string")
+        run_id = run_id_override
     run_context = AgentRunContext(
         session_id=session_id,
-        run_id=uuid.uuid4().hex,
+        run_id=run_id,
         agent_id=agent_id_override,
         parent_run_id=parent_run_id_override,
         environment=environment,
@@ -374,6 +381,7 @@ def create_agent(
     max_steps_override: int | None = None,
     approval_mode_override: str | None = None,
     session_id_override: str | None = None,
+    run_id_override: str | None = None,
     agent_id_override: str | None = None,
     parent_run_id_override: str | None = None,
     context_budget: ContextBudget | None = None,
@@ -395,6 +403,7 @@ def create_agent(
         max_steps_override=max_steps_override,
         approval_mode_override=approval_mode_override,
         session_id_override=session_id_override,
+        run_id_override=run_id_override,
         agent_id_override=agent_id_override,
         parent_run_id_override=parent_run_id_override,
         context_budget=context_budget,

coding-agent/src/coding_agent/tools/subagent.py

@@ -2,6 +2,7 @@
 
 import asyncio
 import contextlib
+import uuid
 from collections.abc import Callable
 from inspect import isawaitable
 from typing import Any, AsyncContextManager, cast
@@ -201,33 +202,45 @@ async def subagent_dispatch(
                 "subagent requires a non-empty parent session_id; "
                 "ensure PipelineContext.session_id or run_context is populated"
             )
-        merged_trace_metadata: dict[str, Any] = (
-            dict(parent_run_context.trace_metadata)
+        child_trace_metadata = {
+            _TRACE_KEY_PARENT_AGENT_ID: parent_agent_id,
+            _TRACE_KEY_CHILD_AGENT_ID: child_agent_id,
+        }
+        child_run_context = (
+            parent_run_context.derive_child(
+                run_id=uuid.uuid4().hex,
+                agent_id=child_agent_id,
+                trace_metadata=child_trace_metadata,
+            )
             if parent_run_context is not None
-            else {}
-        )
-        merged_trace_metadata.update(
-            {
-                _TRACE_KEY_PARENT_AGENT_ID: parent_agent_id,
-                _TRACE_KEY_CHILD_AGENT_ID: child_agent_id,
-            }
+            else None
         )
-        child_pipeline, child_ctx = child_pipeline_builder(
-            parent_provider=__pipeline_ctx__.llm_provider,
-            tape_fork=child_tape,
-            tool_filter=lambda tool_name: tool_name != "subagent",
-            session_id_override=parent_session_id,
-            agent_id_override=child_agent_id,
-            parent_run_id_override=(
-                parent_run_context.run_id if parent_run_context is not None else None
+        child_pipeline_kwargs: dict[str, Any] = {
+            "parent_provider": __pipeline_ctx__.llm_provider,
+            "tape_fork": child_tape,
+            "tool_filter": lambda tool_name: tool_name != "subagent",
+            "session_id_override": parent_session_id,
+            "agent_id_override": child_agent_id,
+            "parent_run_id_override": (
+                child_run_context.parent_run_id
+                if child_run_context is not None
+                else None
             ),
-            context_budget=(
-                parent_run_context.context_budget
-                if parent_run_context is not None
+            "context_budget": (
+                child_run_context.context_budget
+                if child_run_context is not None
                 else None
             ),
-            trace_metadata=merged_trace_metadata,
-        )
+            "trace_metadata": (
+                child_run_context.trace_metadata
+                if child_run_context is not None
+                else child_trace_metadata
+            ),
+        }
+        if child_run_context is not None:
+            child_pipeline_kwargs["environment"] = child_run_context.environment
+            child_pipeline_kwargs["run_id_override"] = child_run_context.run_id
+        child_pipeline, child_ctx = child_pipeline_builder(**child_pipeline_kwargs)
         child_ctx.config["agent_id"] = legacy_agent_id_str(
             effective_agent_id(child_ctx) or child_agent_id
         )

coding-agent/tests/coding_agent/test_bootstrap.py

@@ -2,7 +2,8 @@
 from pathlib import Path
 
 from agentkit.runtime import AgentRunContext, ContextBudget
-from coding_agent.__main__ import create_agent
+from agentkit.tape.tape import Tape
+from coding_agent.__main__ import create_agent, create_child_pipeline
 from coding_agent.environment import LocalEnvironment
 
 
@@ -246,6 +247,100 @@ def test_create_agent_rejects_empty_session_id_override(self, tmp_path):
                 environment=environment,
             )
 
+    def test_create_agent_applies_run_id_override(self, tmp_path):
+        config_path = (
+            Path(__file__).parent.parent.parent / "src" / "coding_agent" / "agent.toml"
+        )
+        if not config_path.exists():
+            pytest.skip("agent.toml not found")
+
+        environment = LocalEnvironment(workspace_root=tmp_path / "env-workspace")
+
+        _pipeline, ctx = create_agent(
+            config_path=config_path,
+            data_dir=tmp_path / "data",
+            api_key="[REDACTED:api-key]",
+            model_override="gpt-test",
+            provider_override="openai",
+            base_url_override="http://localhost:1234/v1",
+            session_id_override="session-1",
+            run_id_override="run-1",
+            environment=environment,
+        )
+
+        assert ctx.run_context.run_id == "run-1"
+
+    def test_create_child_pipeline_applies_run_id_override(self, tmp_path):
+        config_path = (
+            Path(__file__).parent.parent.parent / "src" / "coding_agent" / "agent.toml"
+        )
+        if not config_path.exists():
+            pytest.skip("agent.toml not found")
+
+        environment = LocalEnvironment(workspace_root=tmp_path / "env-workspace")
+
+        _pipeline, ctx = create_child_pipeline(
+            parent_provider=None,
+            tape_fork=Tape(),
+            config_path=config_path,
+            data_dir=tmp_path / "data",
+            api_key="[REDACTED:api-key]",
+            model_override="gpt-test",
+            provider_override="openai",
+            base_url_override="http://localhost:1234/v1",
+            session_id_override="session-1",
+            run_id_override="run-1",
+            environment=environment,
+        )
+
+        assert ctx.run_context.run_id == "run-1"
+
+    def test_create_agent_rejects_empty_run_id_override(self, tmp_path):
+        config_path = (
+            Path(__file__).parent.parent.parent / "src" / "coding_agent" / "agent.toml"
+        )
+        if not config_path.exists():
+            pytest.skip("agent.toml not found")
+
+        environment = LocalEnvironment(workspace_root=tmp_path / "env-workspace")
+
+        with pytest.raises(ValueError, match="run_id_override"):
+            create_agent(
+                config_path=config_path,
+                data_dir=tmp_path / "data",
+                api_key="[REDACTED:api-key]",
+                model_override="gpt-test",
+                provider_override="openai",
+                base_url_override="http://localhost:1234/v1",
+                session_id_override="session-1",
+                run_id_override="",
+                environment=environment,
+            )
+
+    def test_create_child_pipeline_rejects_empty_run_id_override(self, tmp_path):
+        config_path = (
+            Path(__file__).parent.parent.parent / "src" / "coding_agent" / "agent.toml"
+        )
+        if not config_path.exists():
+            pytest.skip("agent.toml not found")
+
+        environment = LocalEnvironment(workspace_root=tmp_path / "env-workspace")
+
+        with pytest.raises(ValueError, match="run_id_override"):
+            create_child_pipeline(
+                parent_provider=None,
+                tape_fork=Tape(),
+                config_path=config_path,
+                data_dir=tmp_path / "data",
+                api_key="[REDACTED:api-key]",
+                model_override="gpt-test",
+                provider_override="openai",
+                base_url_override="http://localhost:1234/v1",
+                session_id_override="session-1",
+                run_id_override="",
+                environment=environment,
+            )
+
     def test_create_agent_reads_subagent_timeout_from_config(self, tmp_path):
         config_path = tmp_path / "agent.toml"
         config_path.write_text(

coding-agent/tests/coding_agent/tools/test_subagent.py

@@ -300,6 +300,8 @@ async def run_turn(self, _goal: str) -> TurnOutcome:
 
     assert result == "Subagent completed: Child finished"
     assert captured_kwargs["session_id_override"] == "parent-session"
+    assert captured_kwargs["run_id_override"]
+    assert captured_kwargs["run_id_override"] != "parent-run"
     assert captured_kwargs["agent_id_override"] == "parent-agent.child-1"
     assert captured_kwargs["parent_run_id_override"] == "parent-run"
     assert captured_kwargs["trace_metadata"] == {
@@ -382,6 +384,40 @@ async def test_subagent_tool_forwards_parent_context_budget_and_merges_trace_met
     }
 
 
+@pytest.mark.asyncio
+async def test_subagent_tool_forwards_parent_environment_to_child_builder(
+    monkeypatch: pytest.MonkeyPatch, tmp_path
+):
+    captured: dict[str, Any] = {}
+    environment = LocalEnvironment(tmp_path)
+
+    monkeypatch.setattr(
+        "coding_agent.tools.subagent.PipelineAdapter", _make_immediate_adapter()
+    )
+
+    parent_ctx = PipelineContext(
+        tape=Tape(),
+        session_id="parent-session",
+        run_context=AgentRunContext(
+            session_id="parent-session",
+            run_id="parent-run",
+            agent_id="parent-agent",
+            environment=environment,
+        ),
+        config={
+            "agent_id": "parent-agent",
+            "subagent_timeout": 30.0,
+            "child_worker_coordinator": _StubCoordinator(),
+        },
+    )
+
+    tool_fn = build_subagent_tool(_capture_kwargs_builder(captured))
+
+    await tool_fn(goal="Inspect", __pipeline_ctx__=parent_ctx)
+
+    assert captured["environment"] is environment
+
+
 @pytest.mark.asyncio
 async def test_subagent_tool_overwrites_caller_supplied_reserved_trace_keys(
     monkeypatch: pytest.MonkeyPatch, tmp_path