Unify subordinate lifecycle across parallel calls

Route direct and parallel call_subordinate execution through one reusable, persisted A1/A2 child-context lifecycle.

Keep failed children resumable with reset=false and cover sibling reuse, recursive numbering, persistence, and nested task ownership.

helpers/parallel_tools.py

@@ -23,6 +23,7 @@ PARALLEL_WORKER_JOB_KEY = "_parallel_job_id"
 PARALLEL_WORKER_KIND_KEY = "_parallel_worker_kind"
 
 CHILD_PARENT_CONTEXT_ID_KEY = "parent_context_id"
+CHILD_PARENT_AGENT_NUMBER_KEY = "parent_agent_number"
 CHILD_PARENT_CONTEXT_KIND_KEY = "parent_context_kind"
 CHILD_PARENT_CONTEXT_LABEL_KEY = "parent_context_label"
 CHILD_PARALLEL_JOB_ID_KEY = "parallel_job_id"
@@ -53,6 +54,7 @@ class ParallelJob:
     tool_name: str
     tool_args: dict[str, Any]
     kind: JobKind
+    parent_agent: "Agent | None" = field(default=None, repr=False)
     state: JobState = "pending"
     created_at: float = field(default_factory=time.time)
     started_at: float | None = None
@@ -208,6 +210,7 @@ async def start_parallel_jobs(
             tool_name=call.tool_name,
             tool_args=call.tool_args,
             kind=kind,
+            parent_agent=agent,
         )
         job_store[job.id] = job
         jobs.append(job)
@@ -218,6 +221,8 @@ async def start_parallel_jobs(
             job.started_at = time.time()
             task = DeferredTask(thread_name=THREAD_BACKGROUND)
             job.deferred_task = task
+            if _parallel_worker_kind(agent) == "subordinate" and context.task:
+                context.task.add_child_task(task)
             task.start_task(_run_parallel_job, context.id, job.id)
         except Exception as exc:
             _finish_job(job, "error", error=str(exc))
@@ -410,34 +415,39 @@ async def _run_parallel_job(parent_context_id: str, job_id: str) -> None:
 
 
 async def _run_subordinate_context_job(parent_context_id: str, job: ParallelJob) -> str:
-    from agent import AgentContext, AgentContextType, UserMessage
-    from helpers import message_queue, persist_chat
+    from agent import AgentContext
     from helpers.tool_policy import ensure_tool_allowed
-    from tools.call_subordinate import _validate_subordinate_profile
+    from tools.call_subordinate import get_or_create_subordinate, run_subordinate
 
     parent_context = AgentContext.get(parent_context_id)
     if not parent_context:
         raise ValueError("Parent context not found.")
-    ensure_tool_allowed(parent_context.agent0, "call_subordinate")
+    parent_agent = job.parent_agent or parent_context.agent0
+    ensure_tool_allowed(parent_agent, "call_subordinate")
 
     args = job.tool_args
     message = str(args.get("message") or "").strip()
     if not message:
         raise ValueError("call_subordinate requires `tool_args.message`.")
 
-    profile = _validate_subordinate_profile(
-        parent_context.agent0,
-        str(args.get("profile") or args.get("agent_profile") or ""),
+    context_id = str(args.get("context_id") or args.get("agent_id") or "").strip()
+    reset = args.get("reset", False)
+    slot = (
+        job.id
+        if coerce_bool(reset, False) and not context_id
+        else "default"
     )
     attachments = args.get("attachments") if isinstance(args.get("attachments"), list) else []
-    attachments = [str(item) for item in attachments]
-
-    child_name = _subordinate_context_name(job)
-    worker_context = AgentContext(
-        config=_clone_config(parent_context.config, profile=profile),
-        name=child_name,
-        type=AgentContextType.USER,
+    subordinate = get_or_create_subordinate(
+        parent_agent,
+        profile=str(args.get("profile") or args.get("agent_profile") or ""),
+        reset=reset,
+        context_id=context_id,
+        name=str(args.get("name") or ""),
+        message=message,
+        slot=slot,
     )
+    worker_context = subordinate.context
     job.worker_context_id = worker_context.id
     if job.deferred_task:
         worker_context.task = job.deferred_task
@@ -445,30 +455,9 @@ async def _run_subordinate_context_job(parent_context_id: str, job: ParallelJob)
     worker_context.set_data(PARALLEL_WORKER_PARENT_CONTEXT_KEY, parent_context.id)
     worker_context.set_data(PARALLEL_WORKER_JOB_KEY, job.id)
     worker_context.set_data(PARALLEL_WORKER_KIND_KEY, job.kind)
-    worker_context.set_output_data(CHILD_PARENT_CONTEXT_ID_KEY, parent_context.id)
-    worker_context.set_output_data(CHILD_PARENT_CONTEXT_KIND_KEY, "parallel")
-    worker_context.set_output_data(CHILD_PARENT_CONTEXT_LABEL_KEY, child_name)
     worker_context.set_output_data(CHILD_PARALLEL_JOB_ID_KEY, job.id)
     worker_context.set_output_data(CHILD_PARALLEL_TOOL_NAME_KEY, job.tool_name)
-    _copy_project(parent_context, worker_context)
-
-    system_prompt = _subordinate_worker_system_prompt(profile)
-    message_queue.log_user_message(worker_context, message, attachments, source=" (parallel)")
-    worker_context.agent0.hist_add_user_message(
-        UserMessage(
-            message=message,
-            attachments=attachments,
-            system_message=[system_prompt],
-        )
-    )
-    persist_chat.save_tmp_chat(worker_context)
-
-    try:
-        result = await worker_context.agent0.monologue()
-        worker_context.agent0.history.new_topic()
-        return result
-    finally:
-        persist_chat.save_tmp_chat(worker_context)
+    return await run_subordinate(parent_agent, subordinate, message, attachments)
 
 
 async def _run_direct_tool_job(parent_context_id: str, job: ParallelJob) -> str:
@@ -711,16 +700,13 @@ def _job_snapshot(job: ParallelJob, *, include_result: bool) -> dict[str, Any]:
     return data
 
 
-def _clone_config(config: "AgentConfig", *, profile: str = "") -> "AgentConfig":
+def _clone_config(config: "AgentConfig") -> "AgentConfig":
     try:
-        cloned = replace(
+        return replace(
             config,
             knowledge_subdirs=list(config.knowledge_subdirs),
             additional=dict(config.additional),
         )
-        if profile:
-            cloned.profile = profile
-        return cloned
     except Exception:
         return config
 
@@ -734,27 +720,3 @@ def _copy_project(parent_context: "AgentContext", worker_context: "AgentContext"
             projects.activate_project(worker_context.id, project_name, mark_dirty=False)
     except Exception:
         pass
-
-
-def _subordinate_worker_system_prompt(profile: str) -> str:
-    lines = [
-        "You are running as an isolated parallel worker for a parent Agent Zero chat.",
-        "Return a concise final textual summary for the parent. Artifacts and files are supplementary, not a substitute for the textual result.",
-    ]
-    if profile:
-        lines.append(f"Act with the `{profile}` profile's expertise and priorities.")
-    return "\n".join(lines)
-
-
-def _subordinate_context_name(job: ParallelJob) -> str:
-    name = str(job.tool_args.get("name") or "").strip()
-    if name:
-        return name
-    message = str(job.tool_args.get("message") or "").strip()
-    label = _short_label(message)
-    return label or f"Parallel subordinate {job.index + 1}"
-
-
-def _short_label(text: str, limit: int = 80) -> str:
-    compact = " ".join(text.split())
-    return compact[:limit].rstrip()

helpers/parallel_tools.py.dox.md

@@ -26,11 +26,11 @@
 - Normalization accepts full agent-reply-shaped objects when `tool_name` and `tool_args` are present; non-contract planning fields such as `thoughts` or `headline` are ignored.
 - `tool_calls` should be an array, but normalization also accepts a valid JSON string encoding of that array to recover provider/model stringification.
 - Normalization rejects `document_query` and `response` inside `parallel`: document parsing and Q&A must run sequentially, while `response` must remain top-level so it can end the message loop.
-- `call_subordinate` jobs first enforce the parent profile's delegation policy
-  and validate the requested profile through the sequential delegation owner,
-  then run in isolated child chat contexts tagged with parent-chat metadata;
-  they must not be added to the scheduler task list and may use normal child-chat
-  tools, including `parallel`.
+- `call_subordinate` jobs first enforce the actual calling agent's delegation policy, then call the same creation and execution functions as direct delegation in `tools/call_subordinate.py`; this helper does not construct or prompt a second kind of subordinate.
+- Fresh parallel sibling calls create distinct `parent.number + 1` child agents. Their job snapshots expose stable `context_id` values that direct or parallel `reset=false` calls can continue after success or failure.
+- Jobs retain their actual parent agent so parallel calls made by A1 create A2 rather than falling back to a context's A0.
+- Subordinate child chats are tagged with job metadata, remain outside the scheduler task list, and may use normal child-chat tools including `parallel`.
+- Nested parallel jobs started by a parallel subordinate are registered as child `DeferredTask` instances so stopping the ancestor also stops its descendants.
 - Direct tool jobs run in isolated background contexts and are blocked from recursively invoking `parallel`.
 - Direct tool background context cleanup removes both the in-memory context and any transient chat folder left on disk.
 - Parent-visible child log items are created for each wrapped call so the WebUI can inspect concurrent children separately while the wrapper result remains model-history-only.

prompts/agent.system.tool.call_sub.md

@@ -1,9 +1,11 @@
 ### call_subordinate
 delegate research or complex subtasks to a specialized agent.
-args: `message`, optional `profile`, `reset`
+args: `message`, optional `profile`, `reset`, `context_id`
 - `profile`: optional prompt profile key for the subordinate; when provided, it must exactly match an available profile; leave empty for the default profile
-- `reset`: use json boolean `true` for the first message or when changing profile; use `false` to continue
+- `reset`: use json boolean `true` to create a fresh child; use `false` to continue the default child or the supplied `context_id`
+- `context_id`: stable child ID returned by an earlier direct or parallel call; use it with `reset: false` to continue that exact child
 - `message`: define role, goal, and the concrete task
+each caller creates its next agent level: A0 creates A1 children, A1 creates A2 children, and so on
 after the subordinate returns, answer from its result directly when it satisfies the user request
 do not repeat the same solving work or call extra tools after a sufficient subordinate result
 example:

prompts/agent.system.tool.parallel.md

@@ -10,7 +10,7 @@ Rules:
 - never nest `parallel`
 - Never include `document_query` in `tool_calls`; it is too heavy for parallel workers, so call it sequentially.
 - Call `response` only as a top-level tool so it ends the message loop; never wrap it inside `parallel.tool_calls`.
-- `call_subordinate` inside `parallel` starts an isolated child chat under the parent chat, not a scheduler task
+- `call_subordinate` uses the same child lifecycle here as it does top-level; fresh siblings are next-level agents, and each job's `context_id` can be continued later with `reset: false`
 - use `wait: false` only when you will collect results later with `job_ids`
 - if extras list running or ready parallel jobs, collect them before final synthesis
 - `timeout` only limits how long this call waits; running jobs continue and can be awaited again by `job_ids`

tests/test_parallel_tool.py

@@ -48,6 +48,7 @@ class _FakeContext:
         self.id = "ctx"
         self.data = {}
         self.log = _FakeLog()
+        self.task = None
 
     def get_data(self, key: str, recursive: bool = True):
         return self.data.get(key)
@@ -60,14 +61,29 @@ class _FakeAgent:
     def __init__(self) -> None:
         self.context = _FakeContext()
         self.agent_name = "A0"
+        self.number = 0
 
 
 class _FakeDeferredTask:
-    def __init__(self, *, ready: bool = False, alive: bool = True, result=None) -> None:
+    def __init__(
+        self,
+        *,
+        ready: bool = False,
+        alive: bool = True,
+        result=None,
+        thread_name=None,
+    ) -> None:
         self.ready = ready
         self.alive = alive
         self._result = result
         self.killed = 0
+        self.thread_name = thread_name
+        self.started = None
+        self.children = []
+
+    def start_task(self, func, *args):
+        self.started = (func, args)
+        return self
 
     def is_ready(self):
         return self.ready
@@ -81,6 +97,12 @@ class _FakeDeferredTask:
     def kill(self):
         self.killed += 1
         self.alive = False
+        for child in self.children:
+            child.kill()
+        self.children = []
+
+    def add_child_task(self, task, terminate_thread=False):
+        self.children.append(task)
 
 
 def test_normalize_parallel_tool_calls_accepts_normal_tool_request_shapes() -> None:
@@ -138,6 +160,20 @@ def test_normalize_parallel_tool_calls_accepts_json_string_array() -> None:
     assert calls[1].tool_args["message"] == "Research nuclear fusion news in Italian."
 
 
+def test_subordinate_prompts_share_reusable_tree_contract() -> None:
+    call_prompt = (PROJECT_ROOT / "prompts/agent.system.tool.call_sub.md").read_text(
+        encoding="utf-8"
+    )
+    parallel_prompt = (PROJECT_ROOT / "prompts/agent.system.tool.parallel.md").read_text(
+        encoding="utf-8"
+    )
+
+    assert "A0 creates A1 children, A1 creates A2 children" in call_prompt
+    assert "stable child ID" in call_prompt
+    assert "same child lifecycle here as it does top-level" in parallel_prompt
+    assert "each job's `context_id`" in parallel_prompt
+
+
 def test_normalize_parallel_tool_calls_rejects_nested_parallel() -> None:
     with pytest.raises(ValueError, match="cannot be nested"):
         parallel_tools.normalize_parallel_tool_calls(
@@ -475,6 +511,246 @@ async def test_parallel_subordinate_reuses_profile_validation(monkeypatch) -> No
         await parallel_tools._run_subordinate_context_job("ctx", job)
 
 
+@pytest.mark.asyncio
+async def test_parallel_subordinates_are_distinct_reusable_a1_children(monkeypatch) -> None:
+    from agent import Agent, AgentConfig, AgentContext
+    from helpers import message_queue, persist_chat, tool_policy
+
+    parent_id = "ctx-parallel-a1-tree"
+    AgentContext.remove(parent_id)
+    parent = AgentContext(
+        AgentConfig(mcp_servers="", profile="agent0"),
+        id=parent_id,
+        set_current=False,
+    )
+
+    async def fake_monologue(agent):
+        return agent.agent_name
+
+    monkeypatch.setattr(Agent, "monologue", fake_monologue)
+    monkeypatch.setattr(tool_policy, "ensure_tool_allowed", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(message_queue, "log_user_message", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(persist_chat, "save_tmp_chat", lambda _context: None)
+
+    child_ids = []
+    try:
+        jobs = await parallel_tools.start_parallel_jobs(
+            parent.agent0,
+            [
+                parallel_tools.NormalizedToolCall(
+                    index=0,
+                    tool_name="call_subordinate",
+                    tool_args={"message": "left branch", "reset": True},
+                ),
+                parallel_tools.NormalizedToolCall(
+                    index=1,
+                    tool_name="call_subordinate",
+                    tool_args={"message": "right branch", "reset": True},
+                ),
+            ],
+        )
+        results = await parallel_tools.await_parallel_jobs(
+            parent.agent0,
+            [job.id for job in jobs],
+            timeout=10,
+        )
+        child_ids = [result["context_id"] for result in results]
+
+        assert [result["state"] for result in results] == ["success", "success"]
+        assert [result["result"] for result in results] == ["A1", "A1"]
+        assert len(set(child_ids)) == 2
+        assert set(parent.agent0.get_data("_subordinates")) == set(child_ids)
+        for child_id in child_ids:
+            child = AgentContext.get(child_id)
+            assert child is not None
+            assert child.agent0.number == 1
+            assert child.get_output_data("parent_context_id") == parent.id
+            assert child.get_output_data("parent_agent_number") == 0
+            assert child.get_output_data("parent_context_kind") == "subordinate"
+    finally:
+        for child_id in child_ids:
+            AgentContext.remove(child_id)
+        AgentContext.remove(parent_id)
+
+
+@pytest.mark.asyncio
+async def test_failed_parallel_subordinate_continues_directly_or_in_parallel(
+    monkeypatch,
+) -> None:
+    from agent import Agent, AgentConfig, AgentContext
+    from helpers import message_queue, persist_chat, tool_policy
+    from tools.call_subordinate import Delegation
+
+    parent_id = "ctx-parallel-resume-tree"
+    AgentContext.remove(parent_id)
+    parent = AgentContext(
+        AgentConfig(mcp_servers="", profile="agent0"),
+        id=parent_id,
+        set_current=False,
+    )
+    calls = {}
+
+    async def flaky_monologue(agent):
+        count = calls.get(agent.context.id, 0) + 1
+        calls[agent.context.id] = count
+        if count == 1:
+            raise RuntimeError("simulated API failure")
+        return f"{agent.agent_name} continuation {count}"
+
+    monkeypatch.setattr(Agent, "monologue", flaky_monologue)
+    monkeypatch.setattr(tool_policy, "ensure_tool_allowed", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(message_queue, "log_user_message", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(persist_chat, "save_tmp_chat", lambda _context: None)
+
+    child_id = ""
+    try:
+        failed = parallel_tools.ParallelJob(
+            id="callsubordin-failed",
+            parent_context_id=parent.id,
+            index=0,
+            tool_name="call_subordinate",
+            tool_args={"message": "remember ALPHA", "reset": True},
+            kind="subordinate",
+            parent_agent=parent.agent0,
+        )
+        parallel_tools._jobs_for_context(parent)[failed.id] = failed
+        await parallel_tools._run_parallel_job(parent.id, failed.id)
+        child_id = failed.worker_context_id or ""
+
+        assert failed.state == "error"
+        assert failed.error == "simulated API failure"
+        assert child_id
+        assert AgentContext.get(child_id).agent0.number == 1  # type: ignore[union-attr]
+
+        direct = Delegation(
+            parent.agent0,
+            "call_subordinate",
+            None,
+            {},
+            "",
+            None,
+        )
+        direct_result = await direct.execute(
+            message="continue after the API failure",
+            context_id=child_id,
+            reset=False,
+        )
+        assert direct_result.message == "A1 continuation 2"
+        assert direct_result.additional == {"context_id": child_id}
+
+        continued = parallel_tools.ParallelJob(
+            id="callsubordin-continued",
+            parent_context_id=parent.id,
+            index=0,
+            tool_name="call_subordinate",
+            tool_args={
+                "message": "continue once more",
+                "context_id": child_id,
+                "reset": False,
+            },
+            kind="subordinate",
+            parent_agent=parent.agent0,
+        )
+        parallel_tools._jobs_for_context(parent)[continued.id] = continued
+        await parallel_tools._run_parallel_job(parent.id, continued.id)
+
+        assert continued.state == "success"
+        assert continued.worker_context_id == child_id
+        assert continued.result == "A1 continuation 3"
+        assert calls == {child_id: 3}
+    finally:
+        if child_id:
+            AgentContext.remove(child_id)
+        AgentContext.remove(parent_id)
+
+
+@pytest.mark.asyncio
+async def test_parallel_a1_spawns_a2_with_same_lifecycle(monkeypatch) -> None:
+    from agent import Agent, AgentConfig, AgentContext
+    from helpers import message_queue, persist_chat, tool_policy
+
+    parent_id = "ctx-parallel-a2-tree"
+    AgentContext.remove(parent_id)
+    parent = AgentContext(
+        AgentConfig(mcp_servers="", profile="agent0"),
+        id=parent_id,
+        set_current=False,
+    )
+
+    async def fake_monologue(agent):
+        return agent.agent_name
+
+    monkeypatch.setattr(Agent, "monologue", fake_monologue)
+    monkeypatch.setattr(tool_policy, "ensure_tool_allowed", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(message_queue, "log_user_message", lambda *_args, **_kwargs: None)
+    monkeypatch.setattr(persist_chat, "save_tmp_chat", lambda _context: None)
+
+    child_ids = []
+    try:
+        a1_job = parallel_tools.ParallelJob(
+            id="callsubordin-a1",
+            parent_context_id=parent.id,
+            index=0,
+            tool_name="call_subordinate",
+            tool_args={"message": "be A1", "reset": True},
+            kind="subordinate",
+            parent_agent=parent.agent0,
+        )
+        a1_result = await parallel_tools._run_subordinate_context_job(parent.id, a1_job)
+        a1 = AgentContext.get(a1_job.worker_context_id or "").agent0  # type: ignore[union-attr]
+        child_ids.append(a1.context.id)
+
+        a2_job = parallel_tools.ParallelJob(
+            id="callsubordin-a2",
+            parent_context_id=a1.context.id,
+            index=0,
+            tool_name="call_subordinate",
+            tool_args={"message": "be A2", "reset": True},
+            kind="subordinate",
+            parent_agent=a1,
+        )
+        a2_result = await parallel_tools._run_subordinate_context_job(
+            a1.context.id, a2_job
+        )
+        a2_context = AgentContext.get(a2_job.worker_context_id or "")
+        child_ids.append(a2_context.id)  # type: ignore[union-attr]
+
+        assert a1_result == "A1"
+        assert a1.number == 1
+        assert a2_result == "A2"
+        assert a2_context.agent0.number == 2  # type: ignore[union-attr]
+        assert a2_context.get_output_data("parent_context_id") == a1.context.id  # type: ignore[union-attr]
+        assert a2_context.get_output_data("parent_agent_number") == 1  # type: ignore[union-attr]
+    finally:
+        for child_id in reversed(child_ids):
+            AgentContext.remove(child_id)
+        AgentContext.remove(parent_id)
+
+
+@pytest.mark.asyncio
+async def test_parallel_subordinate_owns_nested_parallel_tasks(monkeypatch) -> None:
+    monkeypatch.setattr(parallel_tools, "DeferredTask", _FakeDeferredTask)
+    agent = _FakeAgent()
+    parent_task = _FakeDeferredTask()
+    agent.context.task = parent_task
+    agent.context.set_data(parallel_tools.PARALLEL_WORKER_KIND_KEY, "subordinate")
+
+    jobs = await parallel_tools.start_parallel_jobs(
+        agent,  # type: ignore[arg-type]
+        [
+            parallel_tools.NormalizedToolCall(
+                index=0,
+                tool_name="call_subordinate",
+                tool_args={"message": "nested", "reset": True},
+            )
+        ],
+    )
+
+    assert parent_task.children == [jobs[0].deferred_task]
+    parent_task.kill()
+    assert jobs[0].deferred_task.killed == 1  # type: ignore[union-attr]
+
+
 @pytest.mark.asyncio
 async def test_parallel_direct_tool_jobs_fallback_to_generic_tool_log_type(monkeypatch) -> None:
     class FakeDeferredTask:

tests/test_subagent_profiles.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+from datetime import datetime, timezone
 from types import SimpleNamespace
 
 import pytest
@@ -10,10 +11,28 @@ from helpers.errors import RepairableException
 
 
 class _FakeContext:
-    id = "ctx"
+    def __init__(self, id: str = "ctx") -> None:
+        self.id = id
+        self.name = None
+        self.data = {}
+        self.output_data = {}
+        self.created_at = datetime.now(timezone.utc)
+        self.agent0 = None
 
     def get_data(self, key: str, recursive: bool = True):
-        return None
+        return self.data.get(key)
+
+    def set_data(self, key: str, value, recursive: bool = True):
+        self.data[key] = value
+
+    def get_output_data(self, key: str, recursive: bool = True):
+        return self.output_data.get(key)
+
+    def set_output_data(self, key: str, value, recursive: bool = True):
+        self.output_data[key] = value
+
+    def is_running(self) -> bool:
+        return False
 
 
 class _FakeParentAgent:
@@ -45,10 +64,17 @@ class _FakeSubAgent:
     DATA_NAME_SUPERIOR = "_superior"
     DATA_NAME_SUBORDINATE = "_subordinate"
 
-    def __init__(self, number: int, config: AgentConfig, context) -> None:
+    _counter = 0
+
+    def __init__(self, number: int, config: AgentConfig, context=None) -> None:
+        if context is None:
+            self.__class__._counter += 1
+            context = _FakeContext(f"child-{self.__class__._counter}")
         self.number = number
+        self.agent_name = f"A{number}"
         self.config = config
         self.context = context
+        self.context.agent0 = self
         self.data = {}
         self.history = SimpleNamespace(new_topic=lambda: None)
         self.messages = []
@@ -56,6 +82,9 @@ class _FakeSubAgent:
     def set_data(self, key: str, value):
         self.data[key] = value
 
+    def get_data(self, key: str):
+        return self.data.get(key)
+
     def hist_add_user_message(self, message):
         self.messages.append(message)
 
@@ -106,6 +135,12 @@ async def test_call_subordinate_uses_valid_profile(monkeypatch) -> None:
             profile=(override_settings or {}).get("agent_profile", "agent0"),
         ),
     )
+    monkeypatch.setattr(
+        call_subordinate.message_queue, "log_user_message", lambda *_args, **_kwargs: None
+    )
+    monkeypatch.setattr(
+        call_subordinate.persist_chat, "save_tmp_chat", lambda _context: None
+    )
 
     parent = _FakeParentAgent()
     tool = call_subordinate.Delegation(
@@ -118,13 +153,122 @@ async def test_call_subordinate_uses_valid_profile(monkeypatch) -> None:
     )
 
     response = await tool.execute(message="work", profile="developer", reset=True)
-    child = parent.get_data(_FakeSubAgent.DATA_NAME_SUBORDINATE)
+    children = parent.get_data(call_subordinate.SUBORDINATES_DATA_KEY)
+    child = next(iter(children.values()))
 
     assert response.message == "delegated"
+    assert response.additional == {"context_id": child.context.id}
+    assert child.number == 1
     assert child.config.profile == "developer"
     assert child.messages[0].message == "work"
 
 
+@pytest.mark.asyncio
+async def test_call_subordinate_reset_false_reuses_numbered_child(monkeypatch) -> None:
+    import tools.call_subordinate as call_subordinate
+
+    monkeypatch.setattr(call_subordinate, "Agent", _FakeSubAgent)
+    monkeypatch.setattr(
+        call_subordinate, "_subordinate_profile_labels", lambda _agent: {}
+    )
+    monkeypatch.setattr(
+        call_subordinate,
+        "initialize_agent",
+        lambda override_settings=None: AgentConfig(mcp_servers="", profile="agent0"),
+    )
+    monkeypatch.setattr(
+        call_subordinate.message_queue, "log_user_message", lambda *_args, **_kwargs: None
+    )
+    monkeypatch.setattr(
+        call_subordinate.persist_chat, "save_tmp_chat", lambda _context: None
+    )
+
+    parent = _FakeParentAgent()
+    tool = call_subordinate.Delegation(
+        parent,  # type: ignore[arg-type]
+        "call_subordinate",
+        None,
+        {},
+        "",
+        None,
+    )
+    first = await tool.execute(message="first", reset=True)
+    second = await tool.execute(
+        message="continue",
+        context_id=first.additional["context_id"],  # type: ignore[index]
+        reset=False,
+    )
+
+    children = parent.get_data(call_subordinate.SUBORDINATES_DATA_KEY)
+    child = next(iter(children.values()))
+    assert len(children) == 1
+    assert child.number == 1
+    assert [message.message for message in child.messages] == ["first", "continue"]
+    assert second.additional == first.additional
+
+
+def test_subordinate_tree_numbers_each_generation(monkeypatch) -> None:
+    import tools.call_subordinate as call_subordinate
+
+    monkeypatch.setattr(call_subordinate, "Agent", _FakeSubAgent)
+    monkeypatch.setattr(
+        call_subordinate, "_subordinate_profile_labels", lambda _agent: {}
+    )
+    monkeypatch.setattr(
+        call_subordinate,
+        "initialize_agent",
+        lambda override_settings=None: AgentConfig(mcp_servers="", profile="agent0"),
+    )
+
+    parent = _FakeParentAgent()
+    child = call_subordinate.get_or_create_subordinate(
+        parent,  # type: ignore[arg-type]
+        reset=True,
+        message="A1 work",
+    )
+    grandchild = call_subordinate.get_or_create_subordinate(
+        child,  # type: ignore[arg-type]
+        reset=True,
+        message="A2 work",
+    )
+
+    assert child.number == 1
+    assert grandchild.number == 2
+    assert child.context.get_output_data("parent_context_id") == parent.context.id
+    assert grandchild.context.get_output_data("parent_context_id") == child.context.id
+    assert grandchild.get_data(Agent.DATA_NAME_SUPERIOR) is child
+
+
+def test_subordinate_context_id_is_scoped_to_its_parent(monkeypatch) -> None:
+    import tools.call_subordinate as call_subordinate
+
+    monkeypatch.setattr(call_subordinate, "Agent", _FakeSubAgent)
+    monkeypatch.setattr(
+        call_subordinate, "_subordinate_profile_labels", lambda _agent: {}
+    )
+    monkeypatch.setattr(
+        call_subordinate,
+        "initialize_agent",
+        lambda override_settings=None: AgentConfig(mcp_servers="", profile="agent0"),
+    )
+
+    owner = _FakeParentAgent()
+    other = _FakeParentAgent()
+    other.context = _FakeContext("other-parent")
+    child = call_subordinate.get_or_create_subordinate(
+        owner,  # type: ignore[arg-type]
+        reset=True,
+        message="private branch",
+    )
+
+    with pytest.raises(RepairableException, match="was not found under A0"):
+        call_subordinate.get_or_create_subordinate(
+            other,  # type: ignore[arg-type]
+            context_id=child.context.id,
+            reset=False,
+        )
+
+
 @pytest.mark.asyncio
 async def test_call_subordinate_requires_reset_to_change_existing_profile(monkeypatch) -> None:
     import tools.call_subordinate as call_subordinate
@@ -189,6 +333,56 @@ def test_persist_chat_roundtrip_preserves_each_agent_profile(monkeypatch) -> Non
         AgentContext.remove(context_id)
 
 
+def test_persisted_numbered_child_is_reusable_after_reload(monkeypatch) -> None:
+    import tools.call_subordinate as call_subordinate
+
+    config_factory = lambda override_settings=None: AgentConfig(
+        mcp_servers="",
+        profile=(override_settings or {}).get("agent_profile", "agent0"),
+    )
+    monkeypatch.setattr(
+        persist_chat,
+        "initialize_agent",
+        config_factory,
+    )
+    monkeypatch.setattr(call_subordinate, "initialize_agent", config_factory)
+
+    parent_id = "ctx-persisted-agent-tree-parent"
+    AgentContext.remove(parent_id)
+    parent = AgentContext(
+        AgentConfig(mcp_servers="", profile="agent0"),
+        id=parent_id,
+        set_current=False,
+    )
+    child = call_subordinate.get_or_create_subordinate(
+        parent.agent0,
+        reset=True,
+        message="persist me",
+    )
+    context_id = child.context.id
+    try:
+        assert len(persist_chat._serialize_context(parent)["agents"]) == 1
+        serialized = persist_chat._serialize_context(child.context)
+        AgentContext.remove(context_id)
+        parent.agent0.data.pop(call_subordinate.SUBORDINATES_DATA_KEY, None)
+        restored = persist_chat._deserialize_context(serialized)
+        resumed = call_subordinate.get_or_create_subordinate(
+            parent.agent0,
+            context_id=context_id,
+            reset=False,
+        )
+
+        assert restored.agent0.number == 1
+        assert restored.agent0.agent_name == "A1"
+        assert restored.get_output_data("parent_context_id") == parent.id
+        assert restored.get_output_data("parent_agent_number") == 0
+        assert resumed is restored.agent0
+        assert resumed.get_data(Agent.DATA_NAME_SUPERIOR) is parent.agent0
+    finally:
+        AgentContext.remove(context_id)
+        AgentContext.remove(parent_id)
+
+
 @pytest.mark.parametrize("project_name", [None, "demo"], ids=["global", "project"])
 @pytest.mark.asyncio
 async def test_agent_profile_set_uses_scope_and_preserves_subagent_profile(

tools/call_subordinate.py

@@ -1,11 +1,20 @@
-from agent import Agent, UserMessage
-from helpers import projects, subagents
+from agent import Agent, AgentContext, UserMessage
+from helpers import message_queue, persist_chat, projects, subagents
 from helpers.errors import RepairableException
 from helpers.tool import Tool, Response
 from initialize import initialize_agent
 from extensions.python.hist_add_tool_result import _90_save_tool_call_file as save_tool_call_file
 
 
+SUBORDINATES_DATA_KEY = "_subordinates"
+CHILD_PARENT_CONTEXT_ID_KEY = "parent_context_id"
+CHILD_PARENT_AGENT_NUMBER_KEY = "parent_agent_number"
+CHILD_PARENT_CONTEXT_KIND_KEY = "parent_context_kind"
+CHILD_PARENT_CONTEXT_LABEL_KEY = "parent_context_label"
+CHILD_SUBORDINATE_SLOT_KEY = "subordinate_slot"
+DEFAULT_SUBORDINATE_SLOT = "default"
+
+
 def _subordinate_profile_labels(agent: Agent) -> dict[str, str]:
     project = projects.get_context_project_name(agent.context) if agent.context else None
     return {
@@ -34,59 +43,196 @@ def _validate_subordinate_profile(agent: Agent, profile: str) -> str:
     )
 
 
+def _register_subordinate(parent: Agent, subordinate: Agent, slot: str) -> None:
+    subordinates = parent.get_data(SUBORDINATES_DATA_KEY)
+    if not isinstance(subordinates, dict):
+        subordinates = {}
+        parent.set_data(SUBORDINATES_DATA_KEY, subordinates)
+    subordinates[subordinate.context.id] = subordinate
+    subordinate.set_data(Agent.DATA_NAME_SUPERIOR, parent)
+    if slot == DEFAULT_SUBORDINATE_SLOT and subordinate.context is parent.context:
+        parent.set_data(Agent.DATA_NAME_SUBORDINATE, subordinate)
+
+
+def _is_child_context(context: AgentContext, parent: Agent, slot: str | None = None) -> bool:
+    if context.get_output_data(CHILD_PARENT_CONTEXT_ID_KEY) != parent.context.id:
+        return False
+    if context.get_output_data(CHILD_PARENT_AGENT_NUMBER_KEY) != parent.number:
+        return False
+    if context.agent0.number != parent.number + 1:
+        return False
+    return slot is None or context.get_output_data(CHILD_SUBORDINATE_SLOT_KEY) == slot
+
+
+def _is_live_context(context: AgentContext) -> bool:
+    return not isinstance(context, AgentContext) or AgentContext.get(context.id) is context
+
+
+def _find_subordinate(parent: Agent, context_id: str, slot: str) -> Agent | None:
+    registered = parent.get_data(SUBORDINATES_DATA_KEY)
+    registered = registered if isinstance(registered, dict) else {}
+    if context_id:
+        subordinate = registered.get(context_id)
+        if (
+            subordinate
+            and _is_live_context(subordinate.context)
+            and _is_child_context(subordinate.context, parent)
+        ):
+            return subordinate
+        context = AgentContext.get(context_id)
+        if not context or not _is_child_context(context, parent):
+            raise RepairableException(
+                f"Subordinate context '{context_id}' was not found under {parent.agent_name}."
+            )
+        subordinate = context.agent0
+        _register_subordinate(parent, subordinate, slot)
+        return subordinate
+
+    existing = parent.get_data(Agent.DATA_NAME_SUBORDINATE)
+    if slot == DEFAULT_SUBORDINATE_SLOT and existing is not None:
+        return existing
+
+    registered_matches = [
+        subordinate
+        for subordinate in registered.values()
+        if _is_live_context(subordinate.context)
+        and _is_child_context(subordinate.context, parent, slot)
+    ]
+    if registered_matches:
+        return max(
+            registered_matches,
+            key=lambda subordinate: subordinate.context.created_at,
+        )
+
+    matches = [
+        context
+        for context in AgentContext.all()
+        if _is_child_context(context, parent, slot)
+    ]
+    if not matches:
+        return None
+    subordinate = max(matches, key=lambda context: context.created_at).agent0
+    _register_subordinate(parent, subordinate, slot)
+    return subordinate
+
+
+def get_or_create_subordinate(
+    parent: Agent,
+    *,
+    profile: str = "",
+    reset: bool | str = False,
+    context_id: str = "",
+    name: str = "",
+    message: str = "",
+    slot: str = DEFAULT_SUBORDINATE_SLOT,
+) -> Agent:
+    requested_profile = _validate_subordinate_profile(parent, profile)
+    target_context_id = str(context_id or "").strip()
+    reset_requested = str(reset).lower().strip() == "true"
+    if target_context_id and reset_requested:
+        raise RepairableException(
+            "`context_id` continues an existing subordinate and requires reset=false. "
+            "Omit `context_id` to create a fresh subordinate."
+        )
+
+    subordinate = (
+        None
+        if reset_requested
+        else _find_subordinate(parent, target_context_id, slot)
+    )
+    if subordinate:
+        current_profile = str(getattr(subordinate.config, "profile", "") or "")
+        if requested_profile and current_profile != requested_profile:
+            raise RepairableException(
+                f"Subordinate already uses profile '{current_profile or 'default'}'. "
+                f"Set reset=true and omit `context_id` to switch to '{requested_profile}'."
+            )
+        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."
+            )
+        return subordinate
+
+    override_settings = {"agent_profile": requested_profile} if requested_profile else None
+    subordinate = Agent(parent.number + 1, initialize_agent(override_settings=override_settings))
+    context = subordinate.context
+    context.name = str(name or "").strip() or _short_label(message) or subordinate.agent_name
+    context.set_output_data(CHILD_PARENT_CONTEXT_ID_KEY, parent.context.id)
+    context.set_output_data(CHILD_PARENT_AGENT_NUMBER_KEY, parent.number)
+    context.set_output_data(CHILD_PARENT_CONTEXT_KIND_KEY, "subordinate")
+    context.set_output_data(CHILD_PARENT_CONTEXT_LABEL_KEY, context.name)
+    context.set_output_data(CHILD_SUBORDINATE_SLOT_KEY, slot)
+
+    project = projects.get_context_project_name(parent.context)
+    if project:
+        projects.activate_project(context.id, project, mark_dirty=False)
+    model_override = parent.context.get_data("chat_model_override")
+    if model_override:
+        context.set_data("chat_model_override", model_override)
+
+    _register_subordinate(parent, subordinate, slot)
+    return subordinate
+
+
+async def run_subordinate(
+    parent: Agent,
+    subordinate: Agent,
+    message: str,
+    attachments: list[str] | None = None,
+) -> str:
+    assignment = str(message or "").strip()
+    if not assignment:
+        raise RepairableException("call_subordinate requires a non-empty `message`.")
+
+    attachment_paths = [str(item) for item in attachments or []]
+    if subordinate.context is not parent.context:
+        message_queue.log_user_message(
+            subordinate.context,
+            assignment,
+            attachment_paths,
+            source=" (subordinate)",
+        )
+    subordinate.hist_add_user_message(
+        UserMessage(message=assignment, attachments=attachment_paths)
+    )
+    if subordinate.context is not parent.context:
+        persist_chat.save_tmp_chat(subordinate.context)
+
+    try:
+        result = await subordinate.monologue()
+        subordinate.history.new_topic()
+        return result
+    finally:
+        if subordinate.context is not parent.context:
+            persist_chat.save_tmp_chat(subordinate.context)
+
+
+def _short_label(text: str, limit: int = 80) -> str:
+    return " ".join(str(text or "").split())[:limit].rstrip()
+
+
 class Delegation(Tool):
 
-    async def execute(self, message="", reset="", **kwargs):
-        requested_profile = _validate_subordinate_profile(
-            self.agent, kwargs.get("profile", kwargs.get("agent_profile", ""))
+    async def execute(self, message="", reset="", context_id="", **kwargs):
+        attachments = kwargs.get("attachments")
+        attachments = attachments if isinstance(attachments, list) else []
+        subordinate = get_or_create_subordinate(
+            self.agent,
+            profile=kwargs.get("profile", kwargs.get("agent_profile", "")),
+            reset=reset,
+            context_id=context_id or kwargs.get("agent_id", ""),
+            name=kwargs.get("name", ""),
+            message=message,
         )
-        existing_subordinate = self.agent.get_data(Agent.DATA_NAME_SUBORDINATE)
-        reset_requested = str(reset).lower().strip() == "true"
-
-        if existing_subordinate and requested_profile and not reset_requested:
-            current_profile = str(
-                getattr(getattr(existing_subordinate, "config", None), "profile", "")
-                or ""
-            )
-            if current_profile != requested_profile:
-                raise RepairableException(
-                    f"Subordinate already uses profile '{current_profile or 'default'}'. "
-                    f"Set reset=true to switch to '{requested_profile}'."
-                )
-
-        # create subordinate agent using the data object on this agent and set superior agent to his data object
-        if (
-            existing_subordinate is None
-            or reset_requested
-        ):
-            # set subordinate prompt profile if provided, otherwise use the default profile
-            override_settings = (
-                {"agent_profile": requested_profile} if requested_profile else None
-            )
-            config = initialize_agent(override_settings=override_settings)
-
-            # create agent
-            sub = Agent(self.agent.number + 1, config, self.agent.context)
-            # register superior/subordinate
-            sub.set_data(Agent.DATA_NAME_SUPERIOR, self.agent)
-            self.agent.set_data(Agent.DATA_NAME_SUBORDINATE, sub)
-
-        # add user message to subordinate agent
-        subordinate: Agent = self.agent.get_data(Agent.DATA_NAME_SUBORDINATE)  # type: ignore
-        subordinate.hist_add_user_message(UserMessage(message=message, attachments=[]))
-
-        # run subordinate monologue
-        result = await subordinate.monologue()
-
-        # seal the subordinate's current topic so messages move to `topics` for compression
-        subordinate.history.new_topic()
+        result = await run_subordinate(self.agent, subordinate, message, attachments)
 
         # hint to use includes for long responses
-        additional = None
+        additional = {"context_id": subordinate.context.id}
         if len(result) >= save_tool_call_file.LEN_MIN:
             hint = self.agent.read_prompt("fw.hint.call_sub.md")
             if hint:
-                additional = {"hint": hint}
+                additional["hint"] = hint
 
         # result
         return Response(message=result, break_loop=False, additional=additional)

tools/call_subordinate.py.dox.md

@@ -12,11 +12,13 @@
 - `call_subordinate.py.dox.md` owns durable notes about responsibilities, contracts, side effects, and verification for that implementation.
 - Classes:
 - `Delegation` (`Tool`)
-  - `async execute(self, message=..., reset=..., **kwargs)`
+  - `async execute(self, message=..., reset=..., context_id=..., **kwargs)`
   - `get_log_object(self)`
 - Top-level functions:
 - `_subordinate_profile_labels(agent: Agent) -> dict[str, str]`
 - `_validate_subordinate_profile(agent: Agent, profile: str) -> str`
+- `get_or_create_subordinate(...) -> Agent`
+- `run_subordinate(...) -> str`
 
 ## Runtime Contracts
 
@@ -26,12 +28,20 @@
 - `Delegation` defines `execute(...)`.
 - 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.
-- Supplying a different profile for an existing subordinate without `reset=true` raises `RepairableException` instead of silently continuing the old subordinate.
+- Direct and parallel calls use the same creation, continuation, message, history, and persistence functions in this module.
+- 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.
+- Child contexts inherit the caller's project and selected chat-model override, 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.
+- Existing same-context linear subordinates remain reusable for saved-chat compatibility, but new children use child contexts and a private per-parent registry.
 - Imported dependency areas include: `agent`, `extensions.python.hist_add_tool_result`, `helpers`, `helpers.errors`, `helpers.tool`.
 
 ## Key Concepts
 
-- Important called helpers/classes observed in the source: `self.agent.get_data`, `projects.get_context_project_name`, `subagents.get_available_agents_dict`, `RepairableException`, `initialize_agent`, `subordinate.hist_add_user_message`, `subordinate.history.new_topic`, `Response`, `self.agent.context.log.log`, `Agent`, `sub.set_data`, `self.agent.set_data`, `UserMessage`, `subordinate.monologue`, `self.agent.read_prompt`, `str.lower.strip`, `str.lower`.
+- Important called helpers/classes observed in the source: `AgentContext.all`, `projects.get_context_project_name`, `projects.activate_project`, `subagents.get_available_agents_dict`, `RepairableException`, `initialize_agent`, `message_queue.log_user_message`, `persist_chat.save_tmp_chat`, `UserMessage`, `subordinate.monologue`, and `subordinate.history.new_topic`.
 - Keep request/response, tool, or helper semantics documented here at the same time as source changes.
 
 ## Work Guidance
@@ -46,6 +56,7 @@
 - Related tests observed by source search:
   - `tests/test_default_prompt_budget.py`
   - `tests/test_subagent_profiles.py`
+  - `tests/test_parallel_tool.py`
 
 ## Child DOX Index
 

tools/parallel.py.dox.md

@@ -28,7 +28,8 @@
 - `action="await"` waits for requested job IDs until completion or `timeout`; timeout returns running job handles without canceling them.
 - `action="collect"` returns completed job results without waiting.
 - `action="cancel"` requests cancellation for requested job IDs.
-- Recursive use of `parallel` from inside a direct background tool worker is blocked before execution; subordinate child chats started by `call_subordinate` can use normal child-chat tools, including `parallel`.
+- Recursive use of `parallel` from inside a direct background tool worker is blocked before execution; numbered subordinate child chats can use normal child-chat tools, including `parallel`, to create their next-level descendants.
+- Wrapped `call_subordinate` uses the same lifecycle as a top-level call. `job_id` identifies one parallel invocation, while its returned `context_id` identifies the reusable child agent for later `reset=false` calls.
 - The wrapper tool does not create its own visible process-step log; each wrapped child call owns the visible log row, and the wrapper result is recorded only in model history.
 
 ## Key Concepts