fix(scheduler): deliver system injects through session cancel; defer bash job ack

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:

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 {

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<sub_id>[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:

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()

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 应恢复可投递。"""

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)