Track direct subordinate execution with its own context task

Direct delegation awaited the child's monologue without assigning a task to
its separate context. Child chats therefore appeared idle while working and
could not be stopped independently through the normal context controls.

Run direct delegation through the child's existing AgentContext.run_task and
await its result. Propagate caller cancellation to the child, while treating
a child-only Stop as a recoverable error for the parent. Preserve inline
execution for legacy shared-context agents and existing parallel job ownership.

Add lifecycle coverage for direct and nested completion, failure, parent/child
Stop, intervention, and legacy execution. Update profile test doubles and the
tool's ownership documentation.

Validation: 121 focused checks passed; the nine lifecycle cases also passed
in the container framework runtime. Live UI checks verified Working/Paused
transitions, child reuse, completion, and parent recovery after child-only Stop.

tests/test_subagent_profiles.py

@@ -9,6 +9,7 @@ import pytest
 
 from agent import Agent, AgentConfig, AgentContext
 from helpers import persist_chat, projects, settings
+from helpers.defer import DeferredTask
 from helpers.errors import RepairableException
 
 
@@ -20,6 +21,7 @@ class _FakeContext:
         self.output_data = {}
         self.created_at = datetime.now(timezone.utc)
         self.agent0 = None
+        self.task = None
 
     def get_data(self, key: str, recursive: bool = True):
         return self.data.get(key)
@@ -34,7 +36,11 @@ class _FakeContext:
         self.output_data[key] = value
 
     def is_running(self) -> bool:
-        return False
+        return bool(self.task and self.task.is_alive())
+
+    def run_task(self, func, *args):
+        self.task = DeferredTask().start_task(func, *args)
+        return self.task
 
 
 class _FakeParentAgent:

tests/test_subordinate_lifecycle.py

@@ -0,0 +1,129 @@
+import asyncio
+import threading
+
+import pytest
+
+from agent import Agent, AgentConfig, AgentContext, UserMessage
+from api.stop import stop_context
+from helpers import extension
+from helpers.errors import RepairableException
+from tools import call_subordinate
+
+
+@pytest.mark.parametrize("outcome", ["success", "error", "stop_child", "stop_parent"])
+@pytest.mark.parametrize("depth", [1, 2])
+def test_direct_subordinate_task_lifecycle(monkeypatch, outcome, depth):
+    monkeypatch.setattr(extension, "call_extensions_sync", lambda *args, **kwargs: None)
+    monkeypatch.setattr(AgentContext, "_contexts", {})
+    monkeypatch.setattr(Agent, "hist_add_user_message", lambda *args: None)
+    monkeypatch.setattr(call_subordinate.message_queue, "log_user_message", lambda *args, **kwargs: None)
+    saved = []
+    monkeypatch.setattr(call_subordinate.persist_chat, "save_tmp_chat", lambda ctx: saved.append(ctx.id))
+    agents = [Agent(number, AgentConfig(mcp_servers="")) for number in range(depth + 1)]
+    monkeypatch.setattr(call_subordinate, "get_or_create_subordinate", lambda parent, **kwargs: agents[parent.number + 1])
+    started = threading.Event()
+    leaf_finished = threading.Event()
+    parent_resumed = threading.Event()
+    parent_finished = threading.Event()
+    release_leaf = asyncio.Event()
+    release_parent = asyncio.Event()
+    results = []
+
+    async def delegate(parent):
+        tool = call_subordinate.Delegation(parent, "call_subordinate", None, {}, "", None)
+        return await tool.execute(message="work")
+
+    async def leaf():
+        started.set()
+        try:
+            await release_leaf.wait()
+            if outcome == "error":
+                raise ValueError("test failure")
+            return "done"
+        finally:
+            leaf_finished.set()
+
+    async def nested():
+        return (await delegate(agents[1])).message
+
+    async def parent_run():
+        try:
+            try:
+                results.append((await delegate(agents[0])).message)
+            except (RepairableException, ValueError) as error:
+                results.append(error)
+            parent_resumed.set()
+            await release_parent.wait()
+        finally:
+            parent_finished.set()
+
+    monkeypatch.setattr(agents[-1], "monologue", leaf)
+    if depth == 2:
+        monkeypatch.setattr(agents[1], "monologue", nested)
+    parent_task = agents[0].context.run_task(parent_run)
+    loop = parent_task.event_loop_thread.loop
+    try:
+        assert started.wait(2)
+        assert all(agent.context.output()["running"] for agent in agents)
+        assert len({id(agent.context.task) for agent in agents}) == len(agents)
+
+        if outcome == "success":
+            context = agents[-1].context
+            task = context.task
+            context.paused = True
+            assert context.output()["paused"] is True
+            assert context.output()["running"] is True
+            assert context.communicate(UserMessage("intervention")) is task
+            assert agents[-1].intervention.message == "intervention"
+            assert context.paused is False
+
+        if outcome == "stop_parent":
+            assert stop_context(agents[0].context)["stopped"] is True
+            assert parent_finished.wait(2)
+            assert leaf_finished.wait(2)
+            assert not parent_resumed.is_set()
+        else:
+            if outcome == "stop_child":
+                assert stop_context(agents[-1].context)["stopped"] is True
+            else:
+                loop.call_soon_threadsafe(release_leaf.set)
+            assert parent_resumed.wait(2)
+            assert leaf_finished.wait(2)
+            assert agents[0].context.is_running()
+            if outcome == "success":
+                assert results == ["done"]
+            else:
+                assert isinstance(results[0], RepairableException if outcome == "stop_child" else ValueError)
+            assert all(not agent.context.is_running() for agent in agents[1:])
+            loop.call_soon_threadsafe(release_parent.set)
+            assert parent_task.result_sync(timeout=2) is None
+        asyncio.run_coroutine_threadsafe(asyncio.sleep(0), loop).result(2)
+        assert all(not agent.context.is_running() for agent in agents)
+        assert all(saved.count(agent.context.id) == 2 for agent in agents[1:])
+    finally:
+        loop.call_soon_threadsafe(release_leaf.set)
+        loop.call_soon_threadsafe(release_parent.set)
+        for agent in reversed(agents):
+            agent.context.kill_process()
+        assert parent_finished.wait(2)
+        assert leaf_finished.wait(2)
+
+
+@pytest.mark.asyncio
+async def test_legacy_subordinate_keeps_shared_task(monkeypatch):
+    monkeypatch.setattr(extension, "call_extensions_sync", lambda *args, **kwargs: None)
+    monkeypatch.setattr(AgentContext, "_contexts", {})
+    monkeypatch.setattr(Agent, "hist_add_user_message", lambda *args: None)
+    parent = Agent(0, AgentConfig(mcp_servers=""))
+    child = Agent(1, parent.config, parent.context)
+    monkeypatch.setattr(call_subordinate, "get_or_create_subordinate", lambda *args, **kwargs: child)
+
+    async def monologue():
+        assert asyncio.current_task() is caller
+        return "legacy"
+
+    monkeypatch.setattr(child, "monologue", monologue)
+    caller = asyncio.current_task()
+    tool = call_subordinate.Delegation(parent, "call_subordinate", None, {}, "", None)
+    assert (await tool.execute(message="work")).message == "legacy"
+    assert parent.context.task is None

tools/call_subordinate.py

@@ -1,3 +1,5 @@
+import asyncio
+
 from agent import Agent, AgentContext, UserMessage
 from helpers import message_queue, persist_chat, projects, subagents
 from helpers.errors import RepairableException
@@ -151,7 +153,7 @@ def get_or_create_subordinate(
         if subordinate.context is not parent.context and subordinate.context.is_running():
             raise RepairableException(
                 f"Subordinate context '{subordinate.context.id}' is still running. "
-                "Await or cancel its parallel job before continuing it."
+                "Wait for it to finish or stop it before continuing it."
             )
         label = str(name or "").strip()
         if label and subordinate.context is not parent.context:
@@ -230,7 +232,22 @@ class Delegation(Tool):
             name=kwargs.get("name", ""),
             message=message,
         )
-        result = await run_subordinate(self.agent, subordinate, message, attachments)
+        if subordinate.context is self.agent.context:
+            result = await run_subordinate(self.agent, subordinate, message, attachments)
+        else:
+            task = subordinate.context.run_task(
+                run_subordinate, self.agent, subordinate, message, attachments
+            )
+            try:
+                result = await task.result()
+            except asyncio.CancelledError:
+                caller = asyncio.current_task()
+                if caller and caller.cancelling():
+                    task.kill()
+                    raise
+                raise RepairableException(
+                    f"Subordinate context '{subordinate.context.id}' was stopped."
+                ) from None
 
         # hint to use includes for long responses
         additional = {"context_id": subordinate.context.id}

tools/call_subordinate.py.dox.md

@@ -29,12 +29,13 @@
 - Observed side-effect areas: filesystem writes, settings/state persistence.
 - `profile`/`agent_profile` values are validated against available profile keys before use; unknown profiles raise `RepairableException` so the agent can retry with a real profile.
 - Direct and parallel calls use the same creation, continuation, message, history, and persistence functions in this module.
+- Direct delegation runs that helper through the separate child's `AgentContext.run_task`, so running status, intervention, Stop, and completion notifications use the normal context lifecycle. Cancelling the caller cancels its direct child; stopping only the child raises a recoverable error to the caller. Parallel jobs retain their own task ownership, and legacy same-context children remain inline.
 - A non-empty `name` updates both the child context name and sidebar label on creation or continuation. Omission preserves an existing name; legacy same-context children must not rename the parent chat.
 - Every fresh child is `Agent(parent.number + 1, ...)` in its own persisted child-chat context, so sibling A1 agents can each create their own A2 descendants without sharing streaming state.
 - `reset=true` creates a fresh child. `reset=false` continues the caller's default child or the exact child named by `context_id`.
 - Child context IDs are accepted only when their persisted parent context, parent agent number, and child depth match the caller.
 - Supplying a different profile for an existing child without creating a fresh child raises `RepairableException` instead of silently changing its profile.
-- Active parallel children cannot be continued concurrently; await or cancel their job first.
+- Active children cannot be continued concurrently; wait for completion or stop them first.
 - Child contexts inherit the caller's project. After project activation, copy the caller's chat-model override only when the child's resolved scoped preset is `Default` (including missing/invalid selections); non-default scoped presets keep their configured models. Continuing an existing child preserves its chat override.
 - Child contexts are saved before execution and again on exit, and remain reusable after model/API failures.
 - The direct tool result includes `context_id`; parallel job snapshots expose the same stable child ID separately from their per-invocation job ID.
@@ -61,6 +62,7 @@
   - `tests/test_default_prompt_budget.py`
   - `tests/test_subagent_profiles.py`
   - `tests/test_parallel_tool.py`
+  - `tests/test_subordinate_lifecycle.py` (direct and nested completion, failure, parent/child stop, intervention, and legacy shared-context execution)
 
 ## Child DOX Index