⚡ fix(sidecar): agent-runtime IO 批次——终态回写 O(N²) 消除/总线读增量化/后台恢复死路 (#411)

apps/sidecar/src/rpc/agent-handlers.ts

@@ -14,6 +14,7 @@ import {
   deleteAgentThread,
   getAgentThreadMeta,
   getAgentThreadMessages,
+  getAgentThreadSDKMessages,
   getRecentAgentThreadMessages,
   listAllAgentThreads,
   listAgentThreads,
@@ -419,6 +420,17 @@ export function createAgentHandlers(
       {
         runStateStore,
         continuationStore,
+        // 崩溃恢复(#411③):后台任务终态通知由 background-process-recovery 落盘
+        // transcript,按 processJobId 取回供 waiting_background checkpoint 转换
+        resolveBackgroundNotification: async (processJobId) => {
+          const notification = getAgentThreadSDKMessages(input.threadId).find(
+            (message) =>
+              message.type === "system" &&
+              message.subtype === "task_notification" &&
+              message.task_id === processJobId,
+          );
+          return notification;
+        },
       },
       async (checkpoint, state) => {
         // interrupted（软中止 checkpoint）：engine 已为被中断工具补 error 占位

apps/sidecar/src/services/agent-runtime/events/thread-event-bus.test.ts

@@ -119,6 +119,26 @@ describe("ThreadEventBus", () => {
     expect(received.map((e) => e.phase)).toEqual(["update", "end"])
   })
 
+  test("增量 read(#411):水位内 afterSeq 走快路,结果与全量一致;早于水位回退全量", async () => {
+    dir = mkdtempSync(join(tmpdir(), "bus-incr-"))
+    const bus = getThreadEventBus(dir)
+    await bus.publish("th1", "r1", skeletonEvent("run", "start"))
+    await bus.publish("th1", "r1", skeletonEvent("message", "end"))
+    // 首次 read 建立水位(全量)
+    expect((await bus.read("th1")).map((e) => e.seq)).toEqual([1, 2])
+    // 水位内增量:只读新字节段
+    await bus.publish("th1", "r1", skeletonEvent("turn", "end"))
+    const inc = await bus.read("th1", 2)
+    expect(inc.map((e) => e.seq)).toEqual([3])
+    // 早于水位的 afterSeq:回退全量仍可答
+    const old = await bus.read("th1", 0)
+    expect(old.map((e) => e.seq)).toEqual([1, 2, 3])
+    // releaseThread 后水位随 state 失效,重建后全量续读不丢
+    releaseThreadEventBus(dir, "th1")
+    await bus.publish("th1", "r2", skeletonEvent("run", "end"))
+    expect((await bus.read("th1")).map((e) => e.seq)).toEqual([1, 2, 3, 4])
+  })
+
   test("hasEvents 与 readFile 截断语义严格一致（F4 分叉判空捷径）", async () => {
     dir = mkdtempSync(join(tmpdir(), "bus-"))
     const bus = new ThreadEventBus(dir)

apps/sidecar/src/services/agent-runtime/events/thread-event-bus.ts

@@ -5,7 +5,7 @@
  * 后续非 update 相位、PERSIST_COALESCE_MS 窗口或 releaseThread 落盘。
  * 此前每个流式 delta 的累计全文都同步写盘,长会话 events 体积近二次增长(~74% 字节浪费)。
  */
-import { appendFileSync, existsSync, mkdirSync, readFileSync, writeFileSync } from "node:fs"
+import { appendFileSync, closeSync, existsSync, fstatSync, mkdirSync, openSync, readFileSync, readSync, writeFileSync } from "node:fs"
 import { join } from "node:path"
 import type { SdkEventEnvelope, SdkLifecycleEvent } from "@lume/shared"
 
@@ -15,6 +15,9 @@ const UPDATE_COALESCE_MS = 16
 /** update 相位持久折叠窗口(ms):崩溃时最多丢该窗口内的流式过渡态(终值由非 update 相位兜底落盘) */
 const PERSIST_COALESCE_MS = 500
 
+/** hasEvents 判空只读文件头部字节数;首条事件通常是小事件(run.started),超出即回退全量判定 */
+const HAS_EVENTS_HEAD_BYTES = 64 * 1024
+
 interface ThreadState {
   /** 下一个待分配的 seq(初始化 = 文件最后一条合法行 seq + 1) */
   nextSeq: number
@@ -25,6 +28,10 @@ interface ThreadState {
   /** 持久折叠缓冲:与 coalesceBuffer 同 key 语义,但负责落盘——折叠后磁盘每个 key 每窗口只有最新累计态 */
   persistBuffer: Map<string, SdkEventEnvelope>
   persistTimer: ReturnType<typeof setTimeout> | null
+  /** 增量读水位:已消费到的文件字节偏移(null=未建立,read 走全量);releaseThread 后随 state 失效 */
+  readOffset: number | null
+  /** readOffset 之前的最大 seq——afterSeq ≥ 该值才可走增量快路(更早的 afterSeq 需要重读全量) */
+  maxSeqSeen: number
 }
 
 export class ThreadEventBus {
@@ -90,17 +97,96 @@ export class ThreadEventBus {
 
   /** 快照/续传:seq > afterSeq 的全部事件(文件 + 未落盘的持久折叠缓冲,按 seq 归并)。 */
   async read(threadId: string, afterSeq?: number): Promise<SdkEventEnvelope[]> {
-    const all = [...this.readFile(threadId), ...this.pendingEnvelopes(threadId)].sort((a, b) => a.seq - b.seq)
+    const st = this.threads.get(threadId)
+    let fileEvents: SdkEventEnvelope[]
+    if (
+      st?.readOffset != null
+      && (afterSeq ?? 0) >= st.maxSeqSeen
+      && existsSync(this.file(threadId))
+    ) {
+      // 增量快路:活跃线程(publish 过)且 afterSeq 不早于已消费水位——只读新字节段。
+      const segment = this.readSegmentFrom(this.file(threadId), st.readOffset)
+      if (segment) {
+        fileEvents = segment.envelopes
+        st.readOffset = segment.nextOffset
+        const last = fileEvents[fileEvents.length - 1]
+        if (last) st.maxSeqSeen = Math.max(st.maxSeqSeen, last.seq)
+      } else {
+        // 文件被替换/截断:回退全量并重立水位
+        fileEvents = this.readFile(threadId)
+        this.resetReadWatermark(st, threadId)
+      }
+    } else {
+      fileEvents = this.readFile(threadId)
+      if (st) this.resetReadWatermark(st, threadId)
+    }
+    const all = [...fileEvents, ...this.pendingEnvelopes(threadId)].sort((a, b) => a.seq - b.seq)
     return afterSeq === undefined ? all : all.filter((e) => e.seq > afterSeq)
   }
 
+  /** 全量读后建立增量水位:offset=文件末尾,maxSeqSeen=最后一条合法行 seq。 */
+  private resetReadWatermark(st: ThreadState, threadId: string): void {
+    const envelopes = this.readFile(threadId)
+    st.maxSeqSeen = envelopes[envelopes.length - 1]?.seq ?? 0
+    let fd: number | undefined
+    try {
+      fd = openSync(this.file(threadId), "r")
+      st.readOffset = fstatSync(fd).size
+    } catch {
+      st.readOffset = null
+    } finally {
+      if (fd !== undefined) closeSync(fd)
+    }
+  }
+
+  /**
+   * 从 start 字节偏移读到 EOF 并解析完整行;返回事件与新偏移(停在最后一个 \n 之后,
+   * 残尾留待下次——append-only 下只会被补全)。文件短于 start(被替换/截断)返回 null。
+   */
+  private readSegmentFrom(file: string, start: number): { envelopes: SdkEventEnvelope[]; nextOffset: number } | null {
+    let fd: number | undefined
+    try {
+      fd = openSync(file, "r")
+      const size = fstatSync(fd).size
+      if (size < start) return null
+      const buffer = Buffer.alloc(size - start)
+      let total = 0
+      while (total < buffer.length) {
+        const n = readSync(fd, buffer, total, buffer.length - total, start + total)
+        if (n <= 0) break
+        total += n
+      }
+      const text = buffer.toString("utf8", 0, total)
+      const lastNewline = text.lastIndexOf("\n")
+      if (lastNewline === -1) return { envelopes: [], nextOffset: start }
+      const complete = text.slice(0, lastNewline + 1)
+      const out: SdkEventEnvelope[] = []
+      for (const line of complete.split("\n")) {
+        if (!line) continue
+        try {
+          out.push(JSON.parse(line) as SdkEventEnvelope)
+        } catch {
+          break
+        }
+      }
+      return { envelopes: out, nextOffset: start + Buffer.byteLength(complete, "utf8") }
+    } catch {
+      return null
+    } finally {
+      if (fd !== undefined) closeSync(fd)
+    }
+  }
+
   /**
    * 判空捷径(F4 分叉用):与 readFile 的截断语义严格一致——逐行找到第一条非空行,
    * JSON.parse 成功即有事件、失败(毒行)即无;全空行/文件缺失为无。不做全量对象分配。
+   * 只读头部 HAS_EVENTS_HEAD_BYTES:首条事件通常是小事件,判空 O(1);头部无定论时回退全量。
    */
   hasEvents(threadId: string): boolean {
     const file = this.file(threadId)
     if (existsSync(file)) {
+      const verdict = this.headHasEventVerdict(file)
+      if (verdict !== "unknown") return verdict === "yes"
       for (const line of readFileSync(file, "utf8").split("\n")) {
         if (!line) continue
         try {
@@ -115,6 +201,39 @@ export class ThreadEventBus {
     return this.pendingEnvelopes(threadId).length > 0
   }
 
+  /** 头部判定:"yes"=首条非空行是合法事件;"no"=毒行;"unknown"=头部无完整非空行。 */
+  private headHasEventVerdict(file: string): "yes" | "no" | "unknown" {
+    let fd: number | undefined
+    try {
+      fd = openSync(file, "r")
+      const size = fstatSync(fd).size
+      const buffer = Buffer.alloc(Math.min(size, HAS_EVENTS_HEAD_BYTES))
+      let total = 0
+      while (total < buffer.length) {
+        const n = readSync(fd, buffer, total, buffer.length - total, total)
+        if (n <= 0) break
+        total += n
+      }
+      const text = buffer.toString("utf8", 0, total)
+      // 末段可能被截断(头部边界切在行中),仅判定以 \n 收尾的完整行
+      const lines = text.endsWith("\n") ? text.split("\n") : text.slice(0, text.lastIndexOf("\n") + 1).split("\n")
+      for (const line of lines) {
+        if (!line) continue
+        try {
+          JSON.parse(line)
+          return "yes"
+        } catch {
+          return "no"
+        }
+      }
+      return "unknown"
+    } catch {
+      return "unknown"
+    } finally {
+      if (fd !== undefined) closeSync(fd)
+    }
+  }
+
   private file(threadId: string): string {
     return join(this.sessionDir, `${threadId}.events.jsonl`)
   }
@@ -132,6 +251,8 @@ export class ThreadEventBus {
         coalesceTimer: null,
         persistBuffer: new Map(),
         persistTimer: null,
+        readOffset: null,
+        maxSeqSeen: 0,
       }
       this.threads.set(threadId, st)
     }

apps/sidecar/src/services/agent-runtime/interruption/resume-service.test.ts

@@ -310,6 +310,95 @@ describe("LumeResumeService", () => {
     expect((await continuationStore.get("run-1"))?.status).toBe("resumed");
   });
 
+  test("waiting_background(#411③):有持久化终态通知时转换为可续跑 checkpoint", async () => {
+    const dir = mkdtempSync(join(tmpdir(), "lume-resume-bg-terminal-"));
+    const runStateStore = createFileBackedLumeRunStateStore(dir);
+    const continuationStore = createFileBackedRunContinuationStore(dir);
+    await runStateStore.create(makeRunState());
+    await continuationStore.upsert({
+      version: 2,
+      runId: "run-1",
+      threadId: "thread-1",
+      status: "waiting_background",
+      checkpoint: {
+        step: "waiting_for_tool_result",
+        toolCallId: "tool-1",
+        toolName: "Bash",
+        toolKind: "execute",
+        processJobId: "job-1"
+      },
+      reason: "后台命令已持久化，恢复时重新附着而不重复执行。",
+      createdAt: "2026-04-29T00:00:00.000Z",
+      updatedAt: "2026-04-29T00:00:00.000Z"
+    });
+
+    let received: RunContinuationState | undefined;
+    const result = await new LumeResumeService(
+      {
+        runStateStore,
+        continuationStore,
+        resolveBackgroundNotification: async (processJobId) => ({
+          type: "system",
+          subtype: "task_notification",
+          task_id: processJobId,
+          tool_use_id: "tool-1",
+          status: "completed",
+          message: "done",
+          session_id: "thread-1"
+        } as any),
+      },
+      async (checkpoint) => {
+        received = checkpoint;
+        return { finalOutput: "resumed" };
+      }
+    ).resumeRun({ runId: "run-1" });
+
+    expect(result.status).toBe("resumed");
+    // 与 live handleAsyncEvent 同形:syntheticToolResult 从终态通知构造,step 推进
+    expect(received?.status).toBe("ready_to_resume");
+    expect(received?.checkpoint.step).toBe("after_tool_result");
+    const synthetic = received?.checkpoint.syntheticToolResult as Record<string, unknown>;
+    expect(synthetic.tool_use_id).toBe("tool-1");
+    expect(synthetic.content).toBe("done");
+    expect(synthetic.is_error).toBeUndefined();
+    expect((await continuationStore.get("run-1"))?.status).toBe("resumed");
+  });
+
+  test("waiting_background(#411③):无终态通知时如实返回等待态且不改动 checkpoint", async () => {
+    const dir = mkdtempSync(join(tmpdir(), "lume-resume-bg-waiting-"));
+    const runStateStore = createFileBackedLumeRunStateStore(dir);
+    const continuationStore = createFileBackedRunContinuationStore(dir);
+    await runStateStore.create(makeRunState());
+    await continuationStore.upsert({
+      version: 2,
+      runId: "run-1",
+      threadId: "thread-1",
+      status: "waiting_background",
+      checkpoint: {
+        step: "waiting_for_tool_result",
+        toolCallId: "tool-1",
+        toolName: "Bash",
+        toolKind: "execute",
+        processJobId: "job-1"
+      },
+      reason: "后台命令已持久化，恢复时重新附着而不重复执行。",
+      createdAt: "2026-04-29T00:00:00.000Z",
+      updatedAt: "2026-04-29T00:00:00.000Z"
+    });
+
+    const result = await new LumeResumeService(
+      {
+        runStateStore,
+        continuationStore,
+        resolveBackgroundNotification: async () => undefined,
+      },
+      async () => ({ finalOutput: "should-not-run" })
+    ).resumeRun({ runId: "run-1" });
+
+    expect(result.status).toBe("waiting_background");
+    expect((await continuationStore.get("run-1"))?.status).toBe("waiting_background");
+  });
+
   test("does not replay a V2 side-effect tool with an unknown result", async () => {
     const dir = mkdtempSync(join(tmpdir(), "lume-resume-v2-unknown-"));
     const runStateStore = createFileBackedLumeRunStateStore(dir);

apps/sidecar/src/services/agent-runtime/interruption/resume-service.ts

@@ -1,3 +1,4 @@
+import type { SDKMessage } from "@lume/shared";
 import type { RunContinuationState } from "../runner/run-continuation";
 import type { LumeRunState } from "../runner/run-state";
 import type { RunContinuationStore } from "../runner/run-continuation-store";
@@ -23,6 +24,12 @@ export class LumeResumeService {
     private readonly stores?: {
       runStateStore: LumeRunStateStore;
       continuationStore: RunContinuationStore;
+      /**
+       * 崩溃恢复(#411③):按 processJobId 取后台任务已持久化的终态通知
+       * (background-process-recovery 服务在重启后落盘 transcript);
+       * undefined = 尚无持久化终态(任务可能仍活着)。
+       */
+      resolveBackgroundNotification?: (processJobId: string) => Promise<SDKMessage | undefined>;
     },
     private readonly continueRunFromCheckpoint?: ContinueRunFromCheckpoint
   ) {}
@@ -88,13 +95,44 @@ export class LumeResumeService {
 
     if (continuation.version === 2 && continuation.status === "waiting_background") {
       if (continuation.checkpoint.syntheticToolResult === undefined) {
-        return {
-          status: "waiting_background",
-          error: continuation.reason ?? "后台任务仍在运行，已重新附着且不会重复执行命令。"
+        // 崩溃后 live 终态监听(run.handleAsyncEvent)已不在:查 recovery 服务落盘的
+        // 持久化终态通知,有则与 live 同形转换(ready_to_resume + syntheticToolResult),
+        // 让恢复真正发生;无则任务可能仍活着(durable 跨重启),如实返回等待态。
+        const notification = continuation.checkpoint.processJobId
+          ? await this.stores.resolveBackgroundNotification?.(continuation.checkpoint.processJobId)
+          : undefined;
+        if (!notification || typeof notification !== "object") {
+          return {
+            status: "waiting_background",
+            error: "后台任务尚未产生持久化终态；若任务仍在运行，终态落盘后再次恢复即可继续。"
+          };
+        }
+        const record = notification as unknown as Record<string, unknown>;
+        const synthetic = {
+          type: "tool_result",
+          tool_use_id:
+            (typeof record.tool_use_id === "string" && record.tool_use_id)
+            || continuation.checkpoint.toolCallId
+            || "",
+          content: record.message ?? record.summary ?? "",
+          ...(record.status === "failed" || record.status === "stopped" || record.status === "interrupted"
+            ? { is_error: true }
+            : {}),
+          ...(record.execution && typeof record.execution === "object"
+            ? { _meta: { execution: record.execution } }
+            : {}),
         };
+        await this.stores.continuationStore.update(input.runId, {
+          status: "ready_to_resume",
+          checkpoint: { ...continuation.checkpoint, step: "after_tool_result", syntheticToolResult: synthetic },
+          reason: "后台命令已进入终态（崩溃恢复）。"
+        });
+        continuation.status = "ready_to_resume";
+        continuation.checkpoint = { ...continuation.checkpoint, step: "after_tool_result", syntheticToolResult: synthetic };
+      } else {
+        await this.stores.continuationStore.update(input.runId, { status: "ready_to_resume" });
+        continuation.status = "ready_to_resume";
       }
-      await this.stores.continuationStore.update(input.runId, { status: "ready_to_resume" });
-      continuation.status = "ready_to_resume";
     }
 
     if (