diff --git a/src/agent_core/tools/bash_job_notify.py b/src/agent_core/tools/bash_job_notify.py index ca42093..a1703ad 100644 --- a/src/agent_core/tools/bash_job_notify.py +++ b/src/agent_core/tools/bash_job_notify.py @@ -445,6 +445,13 @@ def deliver_via_inject( "source": note.get("source") if note else "cli", "user_id": note.get("user_id") if note else "root", } + if note is not None: + jid = str(note.get("job_id") or "").strip() + if jid: + inject_md["_bash_job_notify"] = { + "job_id": jid, + "remote": bool(note.get("remote", False)), + } chat_hint = None if note and isinstance(note.get("metadata"), dict): chat_hint = str(note["metadata"].get("feishu_chat_id") or "").strip() or None @@ -473,13 +480,15 @@ def deliver_via_inject( logger.warning("bash_job_notify: inject_turn failed session=%s: %s", sid, exc) return False - # 标记该 job 已通知并从注册表移除 if note is not None: - _mark_notified( - sid, - str(note.get("job_id") or ""), - remote=bool(note.get("remote", False)), - ) + jid = str(note.get("job_id") or "").strip() + if jid: + with _LOCK: + jobs = _TRACKED_BY_SESSION.get(sid, {}) + key = _job_key(jid, remote=bool(note.get("remote", False))) + rec = jobs.get(key) + if rec is not None: + rec.staged = True return True @@ -496,6 +505,16 @@ def _mark_notified(session_id: str, job_id: str, *, remote: bool) -> None: _TRACKED_BY_SESSION.pop(session_id, None) +def confirm_bash_job_delivered(session_id: str, job_id: str, *, remote: bool) -> None: + """内核成功跑完 bash_job inject turn 后调用。""" + _mark_notified(session_id, job_id, remote=remote) + + +def abort_bash_job_delivery(session_id: str, job_id: str, *, remote: bool) -> None: + """inject 被 skip/取消/失败时尚未真正投递：清除 staged 以便 poll 重试。""" + _reset_staged(session_id, job_id, remote=remote) + + def _reset_staged(session_id: str, job_id: str, *, remote: bool) -> None: """flush 失败时重置 staged 标记，允许后续 poll 重新尝试投递。""" with _LOCK: diff --git a/src/system/automation/core_gateway.py b/src/system/automation/core_gateway.py index 74e3603..1ecf446 100644 --- a/src/system/automation/core_gateway.py +++ b/src/system/automation/core_gateway.py @@ -1301,6 +1301,11 @@ async def load_skill_for_session( ctx.add_tool_result(call_id, result) injected = True + if injected: + finalize = getattr(agent, "_finalize_turn", None) + if callable(finalize): + await finalize(None) + meta = result.metadata if isinstance(result.metadata, dict) else {} backend = str(meta.get("workspace_backend") or "local") return { diff --git a/src/system/kernel/scheduler.py b/src/system/kernel/scheduler.py index e1690b9..7254443 100644 --- a/src/system/kernel/scheduler.py +++ b/src/system/kernel/scheduler.py @@ -75,6 +75,12 @@ def _kernel_task_exception_detail(exc: BaseException) -> str: # 子任务完成/失败通知：CorePool._inject_to_parent 使用此前端标签（与 P2P 的 agent_msg 区分） _SUBAGENT_LIFECYCLE_FRONTEND_ID = "subagent" +# 会话 cancel 仍应投递的系统 inject（子任务/bash 完成、goal 自启等）。 +# agent_wake 不在此列：skip 时应 abort 以便 poll 重试（见 test_cancel_skip_aborts_agent_wake_delivery）。 +_CANCEL_EXEMPT_SYSTEM_INJECT_FRONTENDS = frozenset( + {"subagent", "bash_job", "goal_start"} +) + # 与 CorePool._inject_to_parent 生成的通知首行一致（信封缺失时仍可从正文解析子 id） _SUBAGENT_LIFECYCLE_LINE_RE = re.compile( r"^\[子任务 (?P[0-9a-fA-F\-]+) (完成|失败)\]", @@ -121,6 +127,12 @@ def _child_sub_session_from_subagent_lifecycle_request( return None +def _is_cancel_exempt_system_inject(request: KernelRequest) -> bool: + """系统级 inject 不受 session cancel 拦截（用户 turn 仍会被 skip）。""" + fi = (request.frontend_id or "").strip() + return fi in _CANCEL_EXEMPT_SYSTEM_INJECT_FRONTENDS + + def _should_suppress_reaped_subagent_parent_notify( request: KernelRequest, core_pool: "CorePool" ) -> bool: @@ -517,6 +529,39 @@ def _finalize_agent_wake_delivery( exc, ) + @staticmethod + def _finalize_bash_job_delivery( + request: KernelRequest, *, confirmed: bool + ) -> None: + md = request.metadata if isinstance(request.metadata, dict) else {} + raw = md.get("_bash_job_notify") + if not isinstance(raw, dict): + return + job_id = str(raw.get("job_id") or "").strip() + if not job_id: + return + remote = bool(raw.get("remote", False)) + sid = (request.session_id or "").strip() + if not sid: + return + try: + from agent_core.tools.bash_job_notify import ( + abort_bash_job_delivery, + confirm_bash_job_delivered, + ) + + if confirmed: + confirm_bash_job_delivered(sid, job_id, remote=remote) + else: + abort_bash_job_delivery(sid, job_id, remote=remote) + except Exception as exc: + logger.debug( + "KernelScheduler: finalize bash_job delivery failed session=%s job=%s: %s", + sid, + job_id, + exc, + ) + async def submit( self, request: KernelRequest, @@ -654,7 +699,10 @@ async def _dispatch_loop(self) -> None: break self._track_dequeue(request.session_id) - if request.session_id in self._cancelled_sessions: + if ( + request.session_id in self._cancelled_sessions + and not _is_cancel_exempt_system_inject(request) + ): logger.info( "KernelScheduler: skipping cancelled queued request " "session_id=%s request_id=%s", @@ -667,6 +715,7 @@ async def _dispatch_loop(self) -> None: ) self._queue.task_done() self._maybe_clear_cancelled(request.session_id) + self._finalize_bash_job_delivery(request, confirmed=False) continue task = asyncio.create_task( @@ -772,7 +821,10 @@ async def _run_and_route(self, request: KernelRequest) -> None: 6. 通过 OutputBus.publish() 广播结果（唯一出口） """ session_id = request.session_id - if session_id in self._cancelled_sessions: + if ( + session_id in self._cancelled_sessions + and not _is_cancel_exempt_system_inject(request) + ): logger.info( "KernelScheduler: skipping cancelled session request session_id=%s request_id=%s", session_id, @@ -786,6 +838,7 @@ async def _run_and_route(self, request: KernelRequest) -> None: # 不用 _session_active_tasks——skip 型 task 自身仍在 set 里且未 done。 self._maybe_clear_cancelled(session_id) self._finalize_agent_wake_delivery(request, confirmed=False) + self._finalize_bash_job_delivery(request, confirmed=False) return if _should_suppress_reaped_subagent_parent_notify(request, self._core_pool): @@ -805,6 +858,7 @@ async def _run_and_route(self, request: KernelRequest) -> None: }, ) self._finalize_agent_wake_delivery(request, confirmed=False) + self._finalize_bash_job_delivery(request, confirmed=False) return self._maybe_wake_agent_inbox_waiter(request) @@ -975,6 +1029,7 @@ async def _do_run() -> None: # - 所有 subscriber 通过 listener 回调获取结果 await self._out_bus.publish(session_id, request.request_id, run_result) self._finalize_agent_wake_delivery(request, confirmed=True) + self._finalize_bash_job_delivery(request, confirmed=True) except asyncio.CancelledError: # 取消时也写 checkpoint，保证 daemon 重启后可恢复 @@ -1006,6 +1061,7 @@ async def _do_run() -> None: asyncio.CancelledError("kernel task cancelled"), ) self._finalize_agent_wake_delivery(request, confirmed=False) + self._finalize_bash_job_delivery(request, confirmed=False) raise except Exception as exc: err_detail = _kernel_task_exception_detail(exc) @@ -1045,6 +1101,7 @@ async def _do_run() -> None: if self._out_bus.has_waiter(request.request_id): await self._out_bus.publish_error(request.request_id, exc) self._finalize_agent_wake_delivery(request, confirmed=False) + self._finalize_bash_job_delivery(request, confirmed=False) else: err_result = AgentRunResult( output_text=f"[后台任务处理出错] {err_detail}", @@ -1054,6 +1111,7 @@ async def _do_run() -> None: session_id, request.request_id, err_result ) self._finalize_agent_wake_delivery(request, confirmed=True) + self._finalize_bash_job_delivery(request, confirmed=False) finally: remaining = self._inflight_sessions.get(session_id, 0) - 1 if remaining > 0: diff --git a/tests/test_bash_job_notify_deliver.py b/tests/test_bash_job_notify_deliver.py index 6db7b1d..bcf8c3c 100644 --- a/tests/test_bash_job_notify_deliver.py +++ b/tests/test_bash_job_notify_deliver.py @@ -82,6 +82,52 @@ async def test_format_notification(): assert "远程(a100)" in remote_text +async def test_bash_job_deferred_mark_until_kernel_confirms(tmp_path): + """inject 入队后不应立即 mark_notified，须等 scheduler 确认投递。""" + from agent_core.tools.bash_job_notify import ( + confirm_bash_job_delivered, + poll_terminal_jobs, + ) + + clear_all_tracking_for_tests() + scheduler = _FakeScheduler() + pool = _FakeCorePool() + set_notify_dependencies(scheduler=scheduler, core_pool=pool) + + ws = str(tmp_path) + register_local_job( + session_id="sid-1", + job_id="job-defer", + command="echo done", + cwd=ws, + log_path=f"{ws}/job.log", + workspace_root=ws, + ) + + notes = await poll_terminal_jobs(max_items=10) + assert len(notes) == 1 + + ok = deliver_via_inject( + session_id="sid-1", + text=format_notification(notes[0]), + note=notes[0], + ) + assert ok is True + assert len(scheduler.injected) == 1 + req = scheduler.injected[0] + assert req.metadata.get("_bash_job_notify") == { + "job_id": "job-defer", + "remote": False, + } + # 尚未确认投递前仍可 poll（staged 会跳过，但注册表仍在） + notes_staged = await poll_terminal_jobs(max_items=10) + assert len(notes_staged) == 0 + + confirm_bash_job_delivered("sid-1", "job-defer", remote=False) + notes_after = await poll_terminal_jobs(max_items=10) + assert len(notes_after) == 0 + + async def test_deliver_inject_turn_when_no_inflight(tmp_path): clear_all_tracking_for_tests() scheduler = _FakeScheduler() diff --git a/tests/test_kernel_runtime.py b/tests/test_kernel_runtime.py index 8b35b2b..d18d040 100644 --- a/tests/test_kernel_runtime.py +++ b/tests/test_kernel_runtime.py @@ -1185,6 +1185,72 @@ async def test_cancel_skip_aborts_agent_wake_delivery() -> None: assert len(poll_due_wakes()) == 1 +@pytest.mark.asyncio +async def test_cancel_does_not_skip_bash_job_system_inject() -> None: + """session cancel 后 bash_job 系统通知仍应投递（不应像用户 turn 一样被 skip）。""" + from agent_core.tools.bash_job_notify import ( + clear_all_tracking_for_tests, + poll_terminal_jobs, + register_local_job, + set_notify_dependencies, + deliver_via_inject, + format_notification, + ) + + clear_all_tracking_for_tests() + scheduler = _minimal_scheduler_for_cancel_tests() + session_id = "cli:root" + scheduler._core_pool.acquire = AsyncMock( # type: ignore[attr-defined] + return_value=SimpleNamespace( + prepare_turn=AsyncMock(return_value=1), + _finalize_turn=AsyncMock(), + _session_logger=None, + ) + ) + + register_local_job( + session_id=session_id, + job_id="job-cancel-exempt", + command="echo done", + cwd="/tmp", + log_path="/tmp/job.log", + workspace_root="/tmp", + ) + notes = await poll_terminal_jobs(max_items=5) + assert len(notes) == 1 + + set_notify_dependencies(scheduler=scheduler, core_pool=None) + deliver_via_inject( + session_id=session_id, + text=format_notification(notes[0]), + note=notes[0], + ) + + scheduler._cancelled_sessions.add(session_id) + req = KernelRequest.create( + text=format_notification(notes[0]), + session_id=session_id, + frontend_id="bash_job", + metadata={ + "_bash_job_notify": {"job_id": "job-cancel-exempt", "remote": False}, + }, + ) + await scheduler._run_and_route(req) + + scheduler._kernel.run.assert_awaited() # type: ignore[attr-defined] + + +@pytest.mark.asyncio +async def test_cancel_still_skips_user_turn_after_cancel() -> None: + """cancel 豁免仅限系统 inject；普通用户 turn 仍应 skip。""" + scheduler = _minimal_scheduler_for_cancel_tests() + session_id = "cli:root" + scheduler._cancelled_sessions.add(session_id) + req = KernelRequest.create(text="user message", session_id=session_id) + await scheduler._run_and_route(req) + scheduler._kernel.run.assert_not_awaited() # type: ignore[attr-defined] + + @pytest.mark.asyncio async def test_cancel_session_tasks_clears_mark_when_lock_waiter_aborted() -> None: """stop 时若第二个请求在等锁，cancel 后 session 应恢复可投递。""" diff --git a/tests/test_load_skill_slash.py b/tests/test_load_skill_slash.py index ef9a0df..05a4766 100644 --- a/tests/test_load_skill_slash.py +++ b/tests/test_load_skill_slash.py @@ -19,6 +19,7 @@ async def test_load_skill_for_session_injects_tool_pair(monkeypatch: pytest.Monk agent.config = MagicMock() agent._context = ctx agent._core_profile = None + agent._finalize_turn = AsyncMock() async def _fake_execute(**kwargs): assert kwargs.get("skill_name") == "demo-skill" @@ -71,3 +72,4 @@ async def __aexit__(self, *args): assert msgs[0]["role"] == "assistant" assert msgs[0]["tool_calls"][0]["function"]["name"] == "load_skill" assert msgs[1]["role"] == "tool" + agent._finalize_turn.assert_awaited_once_with(None)