♻️ fix(sdk): engine 收敛批——惰性 token 估算/stableSerialize 单源/双循环合并/低危积压六项

apps/sidecar/src/services/agent/agent-service.test.ts

@@ -328,7 +328,7 @@ mock.module("../agent-runtime/runtime-core/attempt", () => ({
       } as SDKMessage);
       emit.onSdkMessage({
         type: "result",
-        subtype: "error",
+        subtype: "error_during_execution",
         error: "network failed",
       } as SDKMessage);
       emit.onError("network failed");

apps/sidecar/src/services/agent/agent-submission-store.ts

@@ -3,6 +3,7 @@ import { existsSync, rmSync } from "node:fs";
 import { createRequire } from "node:module";
 import { join } from "node:path";
 import type { AgentSavedFile, AgentSendInput, AgentSubmissionReceipt, AgentThreadMessageDispatchResult } from "@lume/shared";
+import { stableSerialize } from "@lume/shared";
 import { getConfigDir } from "../infra/config-paths";
 import { writeLogRecord } from "../infra/logger";
 
@@ -440,19 +441,7 @@ export function hashAgentSubmission(input: AgentSendInput): string {
     workspaceId: input.workspaceId,
     messageMetadata: input.messageMetadata,
   };
-  return createHash("sha256").update(stableStringify(payload)).digest("hex");
-}
-
-function stableStringify(value: unknown): string {
-  if (Array.isArray(value)) return `[${value.map(stableStringify).join(",")}]`;
-  if (value && typeof value === "object") {
-    return `{${Object.entries(value as Record<string, unknown>)
-      .filter(([, item]) => item !== undefined)
-      .sort(([left], [right]) => left.localeCompare(right))
-      .map(([key, item]) => `${JSON.stringify(key)}:${stableStringify(item)}`)
-      .join(",")}}`;
-  }
-  return JSON.stringify(value) ?? "null";
+  return createHash("sha256").update(stableSerialize(payload)).digest("hex");
 }
 
 function rowToReceipt(row: SubmissionRow): AgentSubmissionReceipt {

packages/natives/index.ts

@@ -424,7 +424,11 @@ export function assertNativeAvailable(): NativeDiagnostics {
 export function countTokens(input: TokenCountInput): TokenCountResult | null {
   const native = loadNative();
   if (!native) return null;
-  return native.countTokens(input);
+  try {
+    return native.countTokens(input);
+  } catch {
+    return null;
+  }
 }
 
 export function countStringTokens(text: string, model?: string): number {
@@ -457,7 +461,7 @@ export async function nativeGrep(
 export function nativeSearch(
   content: string,
   pattern: string,
-  options?: { ignore_case?: boolean; context?: number; max_count?: number },
+  options?: { ignore_case?: boolean; multiline?: boolean; context?: number; max_count?: number },
 ): NativeGrepMatch[] | null {
   const native = loadNative();
   if (!native) return null;

packages/sdk/src/agent.test.ts

@@ -1,8 +1,9 @@
-import { afterEach, describe, expect, test } from "bun:test"
+import { afterEach, describe, expect, spyOn, test } from "bun:test"
 import { mkdirSync, mkdtempSync, rmSync, writeFileSync } from "node:fs"
 import { tmpdir } from "node:os"
 import { join } from "node:path"
 import { createAgent, sessionMessagesFromHistory } from "./agent.js"
+import { QueryEngine } from "./engine.js"
 import { SkillTool } from "./tools/skill-tool.js"
 import type { SDKMessage, ToolDefinition } from "./types.js"
 import type { CreateMessageParams, CreateMessageResponse, LLMProvider } from "./providers/types.js"
@@ -1507,3 +1508,127 @@ describe("Agent session message uuid realignment (#363)", () => {
     expect(rebuilt[2]!.uuid).toBe("u-6")
   })
 })
+
+describe("Agent lazy context usage estimation (#386)", () => {
+  test("run completion does not estimate tokens; getContextUsage computes on demand", async () => {
+    const provider = new StaticProvider()
+    const agent = createAgent({ persistSession: false, tools: [], provider })
+    const spy = spyOn(QueryEngine.prototype, "getContextUsage")
+
+    try {
+      for await (const _event of agent.query("hello")) {
+        // drain query
+      }
+
+      // The per-run finally must not run the full-message token estimation
+      // when no host ever reads usage.
+      expect(spy).toHaveBeenCalledTimes(0)
+
+      const usage = await agent.getContextUsage()
+      expect(spy).toHaveBeenCalledTimes(1)
+      expect(usage.totalTokens).toBeGreaterThan(0)
+      expect(usage.messageBreakdown.assistantMessageTokens).toBeGreaterThan(0)
+
+      // Each on-demand read recomputes against the retained engine.
+      await agent.getContextUsage()
+      expect(spy).toHaveBeenCalledTimes(2)
+    } finally {
+      spy.mockRestore()
+      await agent.close()
+    }
+  })
+
+  test("close releases the retained usage engine; later reads stay safe", async () => {
+    const provider = new StaticProvider()
+    const agent = createAgent({ persistSession: false, tools: [], provider })
+    const spy = spyOn(QueryEngine.prototype, "getContextUsage")
+
+    try {
+      for await (const _event of agent.query("hello")) {
+        // drain query
+      }
+      expect((agent as any).lastUsageEngine).not.toBeNull()
+
+      await agent.close()
+      // The engine (holding the run's full message history) must not stay
+      // reachable through the Agent past close.
+      expect((agent as any).lastUsageEngine).toBeNull()
+
+      // Post-close reads fall back to the safe zero-value shape without
+      // triggering the retained engine's estimation.
+      const usage = await agent.getContextUsage()
+      expect(spy).toHaveBeenCalledTimes(0)
+      expect(usage.totalTokens).toBe(0)
+    } finally {
+      spy.mockRestore()
+    }
+  })
+})
+
+describe("Agent queued async events across runs (#413)", () => {
+  const fakeAsyncEvent = () =>
+    ({ type: "system", subtype: "task_notification", session_id: "s" }) as SDKMessage
+
+  test("abandoning a run mid-stream drops pending async events", async () => {
+    const provider = new StaticProvider()
+    const agent = createAgent({ persistSession: false, tools: [], provider })
+
+    let abandoned = false
+    for await (const _event of agent.query("hello")) {
+      if (abandoned) break
+      // An async event lands in the queue while the consumer is still
+      // iterating, then the consumer abandons the generator.
+      ;(agent as any).queuedSdkEvents.push(fakeAsyncEvent())
+      abandoned = true
+    }
+
+    expect((agent as any).queuedSdkEvents).toHaveLength(0)
+    await agent.close()
+  })
+
+  test("completed runs still deliver queued async events", async () => {
+    const provider = new StaticProvider()
+    const agent = createAgent({ persistSession: false, tools: [], provider })
+
+    const seen: SDKMessage[] = []
+    let injected = false
+    for await (const event of agent.query("hello")) {
+      seen.push(event)
+      if (!injected) {
+        ;(agent as any).queuedSdkEvents.push(fakeAsyncEvent())
+        injected = true
+      }
+    }
+
+    expect(seen.some((event) => event.type === "system" && event.subtype === "task_notification")).toBe(true)
+    expect((agent as any).queuedSdkEvents).toHaveLength(0)
+    await agent.close()
+  })
+
+  test("late background notification after abandoned iteration never enters the queue", async () => {
+    const provider = new StaticProvider()
+    const agent = createAgent({ persistSession: false, tools: [], provider })
+
+    for await (const _event of agent.query("hello")) {
+      break // host abandons mid-stream
+    }
+
+    // Simulate executeSingleTool's post-tool notification path firing after
+    // the host gave up: a background task_notification delivered through the
+    // dead run's captured onAsyncEvent closure.
+    const engine = (agent as any).lastUsageEngine as QueryEngine
+    ;(engine.config.onAsyncEvent as (event: SDKMessage) => void)(fakeAsyncEvent())
+
+    // The closure belongs to a finished generation: dropped, not enqueued...
+    expect((agent as any).queuedSdkEvents).toHaveLength(0)
+
+    // ...and nothing leaks into the next run's stream either.
+    const seen: SDKMessage[] = []
+    for await (const event of agent.query("hello")) {
+      seen.push(event)
+    }
+    expect(seen.some((event) => event.type === "system" && event.subtype === "task_notification")).toBe(false)
+    expect((agent as any).queuedSdkEvents).toHaveLength(0)
+    await agent.close()
+  })
+})

packages/sdk/src/agent.ts

@@ -288,8 +288,14 @@ export class Agent {
   private fileSkillNames = new Set<string>()
   private fileCheckpointState: FileCheckpointState = {}
   private latestUserMessageId: string | undefined
-  private lastContextUsage: ContextUsageResult | null = null
+  private lastUsageEngine: QueryEngine | null = null
   private queuedSdkEvents: SDKMessage[] = []
+  // Generation marker for the async-event queue: advanced when a run's
+  // finally completes. Each run's onAsyncEvent closure captures the value
+  // from before its run, so a late background task_notification firing after
+  // the host abandoned iteration (or between runs) is recognized as stale and
+  // dropped instead of leaking into the next run's event stream.
+  private asyncEventEpoch = 0
   private readonly skillRegistry: SkillRegistry
 
   constructor(options: AgentOptions = {}) {
@@ -910,6 +916,9 @@ export class Agent {
       await this.persistCurrentSession(cwd, opts)
     }
 
+    // Captured before the engine exists: once this run's finally advances the
+    // epoch, closures holding the stale value are recognized as dead runs'.
+    const sdkEventEpoch = this.asyncEventEpoch
     const engine = new QueryEngine({
       cwd,
       model: opts.model || this.modelId,
@@ -966,6 +975,7 @@ export class Agent {
           opts.onAsyncEvent(event)
           return
         }
+        if (sdkEventEpoch !== this.asyncEventEpoch) return
         this.queuedSdkEvents.push(event)
       },
       onLiveEvent: opts.onLiveEvent,
@@ -996,6 +1006,7 @@ export class Agent {
 
     let persistedSessionEvent: SDKMessage | null = null
     let compactionBoundarySeen = false
+    let runCompleted = false
     try {
       for await (const event of engine.submitMessage(modelFacingPrompt)) {
         if (event.type === 'assistant') {
@@ -1030,7 +1041,14 @@ export class Agent {
           yield queued
         }
       }
+      runCompleted = true
     } finally {
+      if (!runCompleted) {
+        // Consumer abandoned the generator mid-run (break / close): pending
+        // async events can no longer be delivered and must not leak into the
+        // next run's event stream.
+        this.queuedSdkEvents.length = 0
+      }
       // Drop any pending debounced write and wait out one already in flight:
       // the awaited persistCurrentSession below writes the same (or fresher)
       // state. Flushing here instead would launch a concurrent fire-and-forget
@@ -1039,8 +1057,15 @@ export class Agent {
       await persistScheduler.cancel()
       opts.abortSignal?.removeEventListener('abort', forwardAbort)
       this.history = engine.getMessages()
-      this.lastContextUsage = engine.getContextUsage()
+      // Keep only the engine reference: the full token estimation runs
+      // lazily when a host actually calls getContextUsage (#386).
+      this.lastUsageEngine = engine
       this.currentEngine = null
+      // Invalidate this run's async-event closures: anything they enqueue from
+      // here on is post-run residue (late background task_notification after
+      // the host stopped iterating) and must not survive into the next run's
+      // drain windows.
+      this.asyncEventEpoch++
       if (compactionBoundarySeen) {
         this.sessionMessages = sessionMessagesFromHistory(this.history, this.sessionMessages)
       }
@@ -1050,11 +1075,14 @@ export class Agent {
       persistedSessionEvent = await this.persistCurrentSession(cwd, opts)
     }
 
+    // Drain before the final yields: once the consumer stops iterating, the
+    // queue must be empty either way — leftover async events belong to this
+    // dead run, not the next one.
+    const tailQueued = this.drainQueuedSdkEvents()
     if (persistedSessionEvent) {
       yield persistedSessionEvent
     }
-
-    for (const queued of this.drainQueuedSdkEvents()) {
+    for (const queued of tailQueued) {
       yield queued
     }
   }
@@ -1322,8 +1350,8 @@ export class Agent {
     if (this.currentEngine) {
       return this.currentEngine.getContextUsage()
     }
-    if (this.lastContextUsage) {
-      return this.lastContextUsage
+    if (this.lastUsageEngine) {
+      return this.lastUsageEngine.getContextUsage()
     }
     const init = await this.getInitializationResult()
     return {
@@ -1456,6 +1484,11 @@ export class Agent {
     this.unregisterFileSkills()
     this.unregisterExplicitSkills()
     this.unregisterPluginSkills()
+    // Release the engine retained for lazy usage estimation (#386): past
+    // close, it would otherwise keep the full message history reachable for
+    // the lifetime of the Agent. getContextUsage falls back to the safe
+    // zero-value shape.
+    this.lastUsageEngine = null
   }
 }
 

packages/sdk/src/engine.test.ts

@@ -1312,10 +1312,7 @@ describe("QueryEngine context controller", () => {
     });
 
     const iterator = engine.submitMessage("run");
-    expect((await iterator.next()).value).toMatchObject({
-      type: "system",
-      subtype: "session_state_changed"
-    });
+    // session_state_changed was retired (#413): init is now the first event.
     expect((await iterator.next()).value).toMatchObject({
       type: "system",
       subtype: "init"