fix(agent): reclaim background processes by execution owner

Track raw task identities across an agent's turns and match them against
process owner_task_id during close. Session IDs and shared terminal keys
are not process ownership, so the old bulk cleanup missed delegated work.
Preserve parent/sibling processes and consume teardown notifications.

Move task-resource cleanup into the lifecycle mixin, add real-process
isolation regressions, and document background process lifetime.

agent/client_lifecycle.py

@@ -1,6 +1,8 @@
-"""OpenAI/Anthropic wire-client lifecycle + credential refresh for ``AIAgent`` (``ClientLifecycleMixin``):
-shared primary client, single-slot per-request client caches (owner-thread close vs stranger-thread abort),
-credential refresh/rotation, route-derived default headers. Extracted from ``run_agent.py``, MRO unchanged."""
+"""Tool-resource teardown, wire-client lifecycle and credential refresh for ``AIAgent``.
+
+``ClientLifecycleMixin`` owns task cleanup, the shared primary client, per-request client caches
+(owner-thread close vs stranger-thread abort), credential rotation and route-derived headers.
+"""
 import logging
 import threading
 import time
@@ -63,6 +65,28 @@ def _valid_credential_pair(api_key: Any, base_url: Any) -> bool:
 
 
 class ClientLifecycleMixin:
+    def _close_task_resources(self, task_id: str) -> None:
+        """Release task resources without treating a shared environment as process ownership."""
+        from run_agent import _quietly, cleanup_browser, cleanup_vm
+
+        def kill_processes() -> None:
+            from tools.process_registry import process_registry
+            # A session can run several task IDs; delegated IDs also differ from session_id.
+            # Never match the environment key (e.g. "default"), shared by parent and siblings.
+            owners = getattr(self, "_process_owner_task_ids", ())
+            for process in process_registry.list_sessions():
+                if process["owner_task_id"] in owners and process["status"] == "running":
+                    process_registry.kill_process(
+                        process["session_id"], source="agent_close", consume_output=True,
+                    )
+
+        def release_computer_use() -> None:
+            from tools.computer_use.tool import release_computer_use_session
+            release_computer_use_session(task_id)
+
+        for step in (kill_processes, lambda: cleanup_vm(task_id), lambda: cleanup_browser(task_id), release_computer_use):
+            _quietly(step)
+
     def _client_log_context(self) -> str:
         thread = threading.current_thread()
         return (

agent/turn_context.py

@@ -419,6 +419,7 @@ def _bind_turn_identity(
     # Unique task_id when not provided isolates VMs between tasks.
     effective_task_id = task_id or str(uuid.uuid4())
     agent._current_task_id = effective_task_id
+    agent._process_owner_task_ids = {*getattr(agent, "_process_owner_task_ids", ()), effective_task_id}
     turn_id = str(getattr(agent, "_relay_pending_turn_id", "") or "") or (
         f"{agent.session_id or 'session'}:{effective_task_id}:{uuid.uuid4().hex[:8]}"
     )

run_agent.py

@@ -926,20 +926,6 @@ class AIAgent(
 
     # -- close()/release_clients() phases -------------------------------------------------------------
 
-    def _close_task_resources(self, task_id: str) -> None:
-        """Kill this task's background processes, then its terminal sandbox, browser daemon and computer-use
-        backend (lazy imports keep the core footprint narrow)."""
-        def kill_processes() -> None:
-            from tools.process_registry import process_registry
-            process_registry.kill_all(task_id=task_id)
-
-        def release_computer_use() -> None:
-            from tools.computer_use.tool import release_computer_use_session
-            release_computer_use_session(task_id)
-
-        for step in (kill_processes, lambda: cleanup_vm(task_id), lambda: cleanup_browser(task_id), release_computer_use):
-            _quietly(step)
-
     def _close_active_children(self, *, soft: bool) -> None:
         """Detach and close per-turn child agents; ``soft`` releases their clients first, falling back to close()."""
         try:

tests/agent/test_owned_process_cleanup.py

@@ -0,0 +1,63 @@
+"""Agent teardown owns processes, not their shared terminal environment."""
+import json
+import shlex
+import sys
+
+from agent.turn_context import _bind_turn_identity
+from run_agent import AIAgent
+from tools.process_registry import ProcessRegistry
+from tools.terminal_tool import terminal_tool
+
+
+def _agent():
+    return AIAgent(api_key="test", base_url="http://127.0.0.1:9/v1",
+                   provider="openai-compat", model="test", enabled_toolsets=[],
+                   quiet_mode=True, skip_context_files=True, skip_memory=True)
+
+
+def _spawn(agent, task_id, tmp_path):
+    _bind_turn_identity(agent, task_id, None, None, None, None)
+    command = shlex.quote(sys.executable) + " -c " + shlex.quote("import time; time.sleep(60)")
+    result = json.loads(terminal_tool(command, background=True, task_id=task_id,
+                                     workdir=str(tmp_path), notify_on_complete=True))
+    return result["session_id"]
+
+
+def test_child_close_kills_only_its_processes(tmp_path, monkeypatch):
+    import tools.process_registry as processes
+    registry = ProcessRegistry()
+    monkeypatch.setattr(processes, "process_registry", registry)
+    parent, child, sibling, unstarted = [_agent() for _ in range(4)]
+    try:
+        parent_id = _spawn(parent, "parent-owner", tmp_path)
+        child_id = _spawn(child, "sa-child-owner", tmp_path)
+        sibling_id = _spawn(sibling, "sa-sibling-owner", tmp_path)
+        assert len({registry._running[s].task_id for s in (parent_id, child_id, sibling_id)}) == 1
+        unstarted.close()
+        assert all(registry.poll(s)["status"] == "running" for s in (parent_id, child_id, sibling_id))
+        child.close()
+        assert registry.poll(child_id)["status"] != "running"
+        assert registry.poll(parent_id)["status"] == "running"
+        assert registry.poll(sibling_id)["status"] == "running"
+        assert child_id in registry._completion_consumed
+        child.close()
+        assert registry.poll(parent_id)["status"] == "running"
+    finally:
+        registry.kill_all()
+        for agent in (parent, child, sibling, unstarted):
+            agent.close()
+
+
+def test_close_reclaims_processes_from_previous_turns(tmp_path, monkeypatch):
+    import tools.process_registry as processes
+    registry = ProcessRegistry()
+    monkeypatch.setattr(processes, "process_registry", registry)
+    agent = _agent()
+    try:
+        first = _spawn(agent, "turn-one-owner", tmp_path)
+        second = _spawn(agent, "turn-two-owner", tmp_path)
+        agent.close()
+        assert all(registry.poll(s)["status"] != "running" for s in (first, second))
+    finally:
+        registry.kill_all()
+        agent.close()

website/docs/user-guide/features/delegation.md

@@ -10,6 +10,12 @@ The `delegate_task` tool spawns child AIAgent instances with isolated context, i
 
 Top-level model calls run in the background automatically. Hermes returns a handle immediately so the conversation can continue, then posts the result back as a new message. An orchestrator subagent waits for its own workers so it can synthesize their results before returning.
 
+## Background process lifetime
+
+Background terminal processes belong to the agent that starts them. Closing a child during delegation teardown terminates its remaining processes, including work started in earlier turns, without stopping processes owned by the parent or sibling agents. Sharing a terminal environment does not transfer process ownership.
+
+A child should wait for its builds, tests, and other bounded background commands before returning its final summary. Start a CI watcher or server in the parent session if it must continue after the child finishes; returning a process ID does not transfer ownership to the parent.
+
 ## Single Task
 
 ```python

website/i18n/zh-Hans/docusaurus-plugin-content-docs/current/user-guide/features/delegation.md

@@ -10,6 +10,12 @@ description: "使用 delegate_task 为并行工作流生成隔离的子智能体
 
 顶层模型调用会自动在后台运行。Hermes 会立即返回句柄，使对话可以继续，并在任务完成后将结果作为新消息发送回来。编排者子智能体会等待自己的工作线程完成，以便在返回前综合结果。
 
+## 后台进程的生命周期
+
+后台终端进程属于启动它的智能体。委派结束并关闭子智能体时，系统会终止它仍在运行的进程，包括先前轮次启动的任务，但不会停止父智能体或其他子智能体拥有的进程。共享终端环境并不意味着共享进程所有权。
+
+子智能体应等待构建、测试等有明确结束条件的后台命令完成，再返回最终摘要。如果 CI 监视器或服务器需要在子智能体结束后继续运行，应由父会话启动；返回进程 ID 不会将所有权转移给父智能体。
+
 ## 单任务
 
 ```python