feat(run): 子智能体状态实时推送 + 主 run 错误失败不级联取消子 run

backend/package/yuxi/services/agent_request_service.py

@@ -20,25 +20,25 @@
 from yuxi.repositories.conversation_repository import ConversationRepository
 from yuxi.repositories.project_repository import ProjectRepository
 from yuxi.services.agent_request_queue_service import (
+    DELIVERY_STATUS_QUEUED,
+    DELIVERY_STATUS_REJECTED,
+    REQUEST_STATUS_QUEUED,
+    REQUEST_STATUS_REJECTED,
     DispatchResult,
-    request_view,
-    validate_queue_policy,
+    dispatch_ready_head,
     get_thread_conversation,
     is_steerable_message_run,
-    dispatch_ready_head,
     queue_conflict,
-    REQUEST_STATUS_QUEUED,
-    REQUEST_STATUS_REJECTED,
-    DELIVERY_STATUS_QUEUED,
-    DELIVERY_STATUS_REJECTED,
+    request_view,
+    validate_queue_policy,
 )
 from yuxi.services.agent_run_service import create_agent_run_input_message, enqueue_agent_run, resolve_agent_run_config
-from yuxi.utils.datetime_utils import utc_now_naive
-from yuxi.workspace.paths import ensure_bound_user_workdir
 from yuxi.services.input_message_service import AgentRunInputMessage
 from yuxi.services.project_service import create_implicit_project
 from yuxi.services.workdir_service import WorkdirBinding, resolve_conversation_workdir_binding
 from yuxi.storage.postgres.models_business import AgentRunRequest, User
+from yuxi.utils.datetime_utils import utc_now_naive
+from yuxi.workspace.paths import ensure_bound_user_workdir
 
 
 @dataclass(frozen=True)

backend/package/yuxi/services/agent_run_manifest_service.py

@@ -15,12 +15,12 @@
 from typing import Any
 
 from sqlalchemy.ext.asyncio import AsyncSession
+from yuxi.agents.backends.paths import runtime_workdir_path
 from yuxi.agents.buildin import agent_manager
 from yuxi.agents.context import BaseContext, prepare_agent_runtime_context
-from yuxi.agents.backends.paths import runtime_workdir_path
-from yuxi.services.workdir_service import AuthorizedWorkdir
 from yuxi.agents.skills.service import PERSONAL_SKILL_SOURCE_TYPE
 from yuxi.repositories.agent_repository import AgentRepository
+from yuxi.services.workdir_service import AuthorizedWorkdir
 from yuxi.storage.postgres.models_business import AgentRun, User
 
 MANIFEST_SCHEMA_VERSION = 2

backend/package/yuxi/services/chat_service.py

@@ -26,7 +26,6 @@
 from yuxi.agents.buildin import agent_manager
 from yuxi.agents.callbacks.model_request_timing import FirstModelRequestRecorder
 from yuxi.agents.context import BaseContext
-from yuxi.services.agent_run_manifest_service import PreparedRunExecution
 from yuxi.agents.state import AgentStatePayload
 from yuxi.models.utils import parse_assistant_message_body
 from yuxi.repositories.agent_repository import AgentRepository
@@ -35,6 +34,7 @@
 from yuxi.repositories.model_message_audit_repository import ModelMessageAuditRepository
 from yuxi.repositories.subagent_thread_repository import SubagentThreadRepository
 from yuxi.repositories.tool_message_audit_repository import ToolMessageAuditRepository
+from yuxi.services.agent_run_manifest_service import PreparedRunExecution
 from yuxi.services.attachment_service import serialize_attachment
 from yuxi.services.input_message_service import AgentRunInputMessage
 from yuxi.services.langfuse_service import (
@@ -832,7 +832,11 @@ async def save_messages_from_langgraph_state(
                 )
                 if terminal_run is None or not changed:
                     raise ValueError(f"AgentRun 输出已写入但 {terminal_status} 终态未能在同一事务提交")
-                cancelled_descendants = await run_repo.cancel_active_execution_tree_descendants(terminal_run)
+                # 与 run_worker.CASCADE_CANCEL_STATUSES 对齐：completed 不级联取消子 Run，
+                # 子 Run 继续执行落库、主 Run 续跑时收割。此处终态仅 completed/interrupted，
+                # 只有 interrupted 命中取消类终态，才收敛 execution tree。
+                if terminal_status == "interrupted":
+                    cancelled_descendants = await run_repo.cancel_active_execution_tree_descendants(terminal_run)
             await conv_repo.db.commit()
             await publish_cancel_signals([run_id for run_id, _thread_id in cancelled_descendants])
             return terminal_status is not None

backend/package/yuxi/services/run_worker.py

@@ -26,8 +26,8 @@
 )
 from yuxi.services.agent_run_manifest_service import (
     PreparedRunExecution,
-    prepare_run_execution,
     compute_manifest_fingerprint,
+    prepare_run_execution,
 )
 from yuxi.services.chat_service import get_agent_state_view, stream_agent_chat, stream_agent_resume
 from yuxi.services.input_message_service import restore_chat_input_message
@@ -74,6 +74,11 @@
 RUN_LEASE_SECONDS = 120
 RUN_HEARTBEAT_SECONDS = 30
 SUPPORTED_RUN_TYPES = {"chat", "resume", "subagent"}
+# 只有「取消类」终态才级联取消后代 Run。错误/正常终态(failed/completed)刻意保留
+# 活跃子 Run 继续执行：断网恢复后主 Run 从 checkpoint 续跑时仍可收割子 Run 成果。
+# 该策略同时约束 mark_run_terminal、worker finally 与重试跳过路径——任何一处漏掉，
+# 都会让「失败不连坐子 Run」的承诺在真实执行链路上失效。
+CASCADE_CANCEL_STATUSES = frozenset({"cancelled", "cancel_requested", "interrupted"})
 WORKER_ID = f"worker-{uuid.uuid4().hex}"
 _RECONCILIATION_TASK_KEY = "agent_run_reconciliation_task"
 _TASK_RECONCILIATION_TASK_KEY = "durable_task_reconciliation_task"
@@ -332,10 +337,26 @@ async def _release_runtime_if_idle(run: AgentRun) -> bool:
 
 
 async def _release_runtime_before_terminal_event(run: AgentRun | None) -> None:
-    """在终态事件可见前收敛 runtime，避免客户端撞上随后发生的删除。"""
+    """在终态事件可见前收敛 runtime，避免客户端撞上随后发生的删除。
+
+    取消类终态已级联取消后代，execution tree 必须能立即收敛；否则抛
+    RuntimeCleanupPendingError 让 ARQ 重试。
+
+    非取消类终态(failed/completed)刻意保留活跃子 Run 继续执行，runtime 仍被子 Run
+    占用、cleanup 天然无法收敛：此时不强求立即清理，保持 runtime_cleanup_pending=True，
+    由 reconcile_pending_runtime_cleanups 在子 Run 收敛后完成清理。
+    """
     if run is None or run.run_type == "subagent":
         return
-    await _require_runtime_cleanup(run, f"Run {run.id} 的 execution tree 尚未完成 runtime cleanup")
+    if run.status in CASCADE_CANCEL_STATUSES:
+        await _require_runtime_cleanup(run, f"Run {run.id} 的 execution tree 尚未完成 runtime cleanup")
+        return
+    # failed/completed 刻意保留活跃子 Run：tree 未收敛是预期结果，保持 runtime_cleanup_pending
+    # 交由 reconcile 完成。但 provisioner/并发清理自身失败仍属基础设施故障，必须重试。
+    try:
+        await _release_runtime_if_idle(run)
+    except Exception as exc:
+        raise RuntimeCleanupPendingError(f"Run {run.id} 的 execution tree 尚未完成 runtime cleanup") from exc
 
 
 async def _require_runtime_cleanup(run: AgentRun, message: str) -> None:
@@ -396,15 +417,52 @@ async def _flush_writer_best_effort(writer: ChunkedEventWriter) -> None:
         logger.warning(f"Failed to flush non-authoritative AgentRun events: run={writer.run_id}", exc_info=True)
 
 
+async def _clear_cancel_signal_best_effort(run_id: str) -> None:
+    """取消键清理失败不能覆盖已经提交的 Run 终态。"""
+
+    try:
+        await clear_cancel_signal(run_id)
+    except Exception:
+        logger.warning(f"Failed to clear non-authoritative AgentRun cancel signal: run={run_id}", exc_info=True)
+
+
+async def _publish_subagent_run_update(run: AgentRun | None) -> None:
+    """子 run 生命周期变化时向父 run 事件流推送增量。
+
+    父 graph 在并行 task 阻塞期间不产生 values 事件，agent_state 冻结；
+    该增量让前端在子 run 启动/终结的瞬间更新面板，而不是等全部 task 一起返回。
+    """
+    if run is None or run.run_type != "subagent" or not run.created_by_run_id:
+        return
+    try:
+        from yuxi.services.subagent_run_service import serialize_subagent_run_state
+
+        payload = serialize_subagent_run_state(run)
+    except Exception:
+        logger.warning(f"Failed to serialize subagent run {run.id} for parent update", exc_info=True)
+        return
+    # 事件必须挂到父 Run 的线程上：订阅方按父线程归属消费该增量。
+    # run.conversation_thread_id 是子会话线程 ID，不是父线程锚点。
+    parent_run = await _get_run(run.created_by_run_id)
+    await _append_run_event_best_effort(
+        run.created_by_run_id,
+        "subagent_run_update",
+        {"subagent_run": payload},
+        thread_id=parent_run.conversation_thread_id if parent_run is not None else None,
+    )
+
+
 async def mark_run_running(run_id: str, worker_id: str) -> bool:
     async with pg_manager.get_async_session_context() as db:
         repo = AgentRunRepository(db)
-        _, acquired = await repo.mark_running(
+        run, acquired = await repo.mark_running(
             run_id,
             worker_id=worker_id,
             lease_seconds=RUN_LEASE_SECONDS,
         )
-        return acquired
+    if acquired:
+        await _publish_subagent_run_update(run)
+    return acquired
 
 
 async def renew_run_lease(run_id: str, worker_id: str) -> bool:
@@ -457,6 +515,8 @@ async def mark_run_terminal(
     error_message: str | None = None,
     token_usage: dict | None = None,
     worker_id: str | None = None,
+    *,
+    cascade_cancel_descendants: bool = True,
 ):
     cancelled_descendants: list[tuple[str, str]] = []
     async with pg_manager.get_async_session_context() as db:
@@ -469,10 +529,14 @@ async def mark_run_terminal(
             token_usage=token_usage,
             worker_id=worker_id,
         )
-        if changed and run is not None:
+        # 仅用户主动取消才级联取消子 run。主 run 因模型/网络错误失败时子 run 不取消：
+        # 断网恢复后主 run 从 checkpoint 续跑，仍可收割子 run 已完成/继续执行中的成果。
+        if changed and run is not None and cascade_cancel_descendants:
             cancelled_descendants = await repo.cancel_active_execution_tree_descendants(run)
         persisted_status = run.status if run else None
     await publish_cancel_signals([child_id for child_id, _thread_id in cancelled_descendants])
+    if changed:
+        await _publish_subagent_run_update(run)
     return TerminalTransition(status=persisted_status, changed=changed)
 
 
@@ -752,6 +816,9 @@ async def _finish_run(
         error_message=error_message,
         token_usage=token_usage,
         worker_id=worker_id,
+        # 主 run 失败(failed)不级联取消子 run：子 run 继续执行落库，
+        # 主 run 之后从 checkpoint 续跑时仍可收割；用户取消走 _finish_user_cancel 仍级联。
+        cascade_cancel_descendants=status in CASCADE_CANCEL_STATUSES,
     )
     if transition.status in TERMINAL_RUN_STATUSES:
         committed_run = await _get_run(run_id)
@@ -848,10 +915,16 @@ async def process_agent_run(ctx, run_id: str):
         return
 
     if run.status in TERMINAL_RUN_STATUSES:
-        await _finish_execution_tree_children(run)
+        if run.status in CASCADE_CANCEL_STATUSES:
+            await _finish_execution_tree_children(run)
         cleanup_was_pending = bool(getattr(run, "runtime_cleanup_pending", False))
         if cleanup_was_pending:
-            await _require_runtime_cleanup(run, f"Run {run_id} 的 execution tree 尚未完成 runtime cleanup")
+            if run.status in CASCADE_CANCEL_STATUSES:
+                await _require_runtime_cleanup(run, f"Run {run_id} 的 execution tree 尚未完成 runtime cleanup")
+            else:
+                # failed/completed 可能仍有活跃子 Run 占用 runtime，cleanup 交由
+                # reconcile_pending_runtime_cleanups 在子 Run 收敛后完成。
+                await _release_runtime_before_terminal_event(run)
             await _append_end_event(run_id, run.status, thread_id=run.conversation_thread_id)
         if run.status == "completed":
             await dispatch_next_request(
@@ -1158,7 +1231,10 @@ async def record_prepared() -> None:
                         if status == "finished":
                             if chunk.get("terminal_committed") is True:
                                 committed_run = await _get_run(run_id)
-                                if committed_run is not None:
+                                # completed 与 failed 同策略：不取消活跃后代(见
+                                # CASCADE_CANCEL_STATUSES)。正常完成时子 Run 已收敛，
+                                # 该分支是幂等兜底；异常残留由 reconcile 按 lease 收敛。
+                                if committed_run is not None and committed_run.status in CASCADE_CANCEL_STATUSES:
                                     await _finish_execution_tree_children(committed_run)
                                 await _release_runtime_before_terminal_event(committed_run)
                                 await _append_end_event(
@@ -1474,7 +1550,9 @@ async def record_prepared() -> None:
         except Exception:
             logger.error(f"Failed to load AgentRun during lifecycle cleanup: run={run_id}", exc_info=True)
             final_run = None
-        if final_run and final_run.status in TERMINAL_RUN_STATUSES:
+        # 只有取消类终态才在收尾时取消活跃后代；failed/completed 保留子 Run 继续执行
+        # (与 mark_run_terminal 的 cascade 策略同一语义，见 CASCADE_CANCEL_STATUSES)。
+        if final_run and final_run.status in CASCADE_CANCEL_STATUSES:
             await _finish_execution_tree_children(final_run)
         if final_run and final_run.status == "cancelled":
             await clear_cancel_signal(run_id)

backend/package/yuxi/services/scheduled_agent_service.py

@@ -19,8 +19,8 @@
 from yuxi.repositories.agent_repository import AgentRepository
 from yuxi.repositories.project_repository import ProjectRepository
 from yuxi.repositories.scheduled_agent_repository import ScheduledAgentRepository
+from yuxi.services.agent_request_service import AgentRequestInput, RunOrigin, submit_agent_request
 from yuxi.services.input_message_service import build_chat_input_message
-from yuxi.services.agent_request_service import RunOrigin, AgentRequestInput, submit_agent_request
 from yuxi.storage.postgres.manager import pg_manager
 from yuxi.storage.postgres.models_business import ScheduledAgentJob, ScheduledAgentRun, User
 from yuxi.utils.datetime_utils import format_utc_datetime, utc_now_naive

backend/package/yuxi/services/subagent_run_service.py

@@ -167,6 +167,20 @@ async def start(
         if created:
             await self.db.commit()
             await agent_run_service.enqueue_agent_run(run.id)
+            # 创建即推送：父面板立刻出现该子智能体条目，不等 worker 领取(running)后的推送。
+            try:
+                from yuxi.services.run_queue_service import append_run_stream_event
+
+                # 事件必须挂到父 Run 的线程上：订阅方按父线程归属消费该增量；
+                # run.conversation_thread_id 是子会话线程 ID，不是父线程锚点。
+                await append_run_stream_event(
+                    created_by_run_id,
+                    "subagent_run_update",
+                    {"subagent_run": serialize_subagent_run_state(run)},
+                    thread_id=creator_run.conversation_thread_id,
+                )
+            except Exception:
+                pass  # 推送失败不影响 run 创建；后续 mark_run_running 仍会推送
 
         return SubagentStartResult(
             run=run,

backend/test/unit/services/test_chat_service_sync.py

@@ -920,6 +920,63 @@ async def cancel_active_execution_tree_descendants(self, _run):
     }
 
 
+@pytest.mark.asyncio
+async def test_complete_run_does_not_cascade_cancel_descendants(
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    events: list[tuple] = []
+
+    class FakeDB:
+        async def commit(self):
+            events.append(("commit",))
+
+        async def rollback(self):
+            events.append(("rollback",))
+
+    class FakeGraph:
+        async def aget_state(self, _config):
+            return SimpleNamespace(values={"messages": [AIMessage(content="done")]})
+
+    class FakeRunRepo:
+        def __init__(self, _db):
+            pass
+
+        async def lock_output_persistence(self, *_args, **_kwargs):
+            events.append(("lock",))
+            return object()
+
+        async def set_output_message(self, run_id, message_id, *, worker_id):
+            events.append(("message", run_id, message_id, worker_id))
+
+        async def set_terminal_status(self, run_id, **kwargs):
+            events.append(("terminal", run_id, kwargs))
+            return SimpleNamespace(status="completed"), True
+
+        async def cancel_active_execution_tree_descendants(self, _run):
+            events.append(("descendants",))
+            return []
+
+    fake_db = FakeDB()
+    monkeypatch.setattr(svc, "AgentRunRepository", FakeRunRepo)
+    monkeypatch.setattr(svc, "ModelMessageAuditRepository", _EmptyModelAuditRepo)
+    monkeypatch.setattr(svc, "ToolMessageAuditRepository", _EmptyToolAuditRepo)
+
+    terminal_committed = await svc.save_messages_from_langgraph_state(
+        state=await FakeGraph().aget_state({}),
+        thread_id="thread-1",
+        conv_repo=_FakeConvRepo(fake_db),
+        run_id="run-1",
+        request_id="request-1",
+        worker_id="worker-1",
+        complete_run=True,
+        token_usage={"available": False},
+    )
+
+    assert terminal_committed is True
+    # completed 不级联取消子 Run：事件序列中不得出现 descendants。
+    assert [event[0] for event in events] == ["lock", "message", "terminal", "commit"]
+
+
 @pytest.mark.asyncio
 async def test_workspace_prompt_excludes_memory_from_shared_context(
     tmp_path: Path,