fix: complete streaming event coverage + generator cleanup

- yield tool_call_start before tool execution (was defined but never yielded)
- yield skill_activated when activate_skill tool succeeds
- try/finally guarantees run_end on early consumer break
- tool_call_end.durationMs made optional (no per-tool timing data available)
- 6 new tests: tool_call_start ordering, skill_activated, fatal/pause run_end,
  run()/stream() equivalence, concurrent tool events

Co-Authored-By: Claude Opus 4.6 (1M context) <noreply@anthropic.com>

Author: barry3406

packages/ai/src/loop/engine.ts

@@ -215,9 +215,14 @@ export async function* runSmartLoopStream(
   /* Phase 1: Initialization                                          */
   /* ================================================================ */
 
+  const runStartTime = Date.now();
+  let finalResultYielded = false;
+
   // 1a. Create engine state from config (or checkpoint for resume)
   const state = createEngineState(config, goal, checkpoint, resumeMessage);
 
+  try {
+
   // 1b. Build tool catalog (inline or deferred)
   const catalog = createToolCatalog(state.tools, config.toolCatalog);
   const allTools = catalog.discoverTool
@@ -399,7 +404,7 @@ export async function* runSmartLoopStream(
             status: "paused",
             checkpoint: buildCheckpoint(state, "paused"),
           };
-          yield { type: "run_end" as const, result: pausedResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
+          finalResultYielded = true; yield { type: "run_end" as const, result: pausedResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
           return pausedResult;
         }
         if (decision.action === "cancel") {
@@ -411,7 +416,7 @@ export async function* runSmartLoopStream(
             status: "canceled",
             checkpoint: buildCheckpoint(state, "canceled"),
           };
-          yield { type: "run_end" as const, result: canceledResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
+          finalResultYielded = true; yield { type: "run_end" as const, result: canceledResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
           return canceledResult;
         }
       } catch {
@@ -485,7 +490,7 @@ export async function* runSmartLoopStream(
             error: `Primary LLM failed: ${primaryMsg}. Fallback LLM also failed: ${fallbackMsg}`,
             checkpoint: buildCheckpoint(state),
           };
-          yield { type: "run_end" as const, result: fatalResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
+          finalResultYielded = true; yield { type: "run_end" as const, result: fatalResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
           return fatalResult;
         }
         // If we reach here, fallback succeeded — skip the context_limit path
@@ -562,7 +567,7 @@ export async function* runSmartLoopStream(
           error: "Context overflow unrecoverable: autocompact and reactive compact both exhausted",
           checkpoint: buildCheckpoint(state),
         };
-        yield { type: "run_end" as const, result: contextFatalResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
+        finalResultYielded = true; yield { type: "run_end" as const, result: contextFatalResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
         return contextFatalResult;
       } else {
         // Non-context error with no fallback configured — use decideContinuation
@@ -583,7 +588,7 @@ export async function* runSmartLoopStream(
             error: errorMsg || contAction.error,
             checkpoint: buildCheckpoint(state),
           };
-          yield { type: "run_end" as const, result: errFatalResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
+          finalResultYielded = true; yield { type: "run_end" as const, result: errFatalResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
           return errFatalResult;
         }
         // Should not reach here for "error" finishReason (decideContinuation returns fatal)
@@ -645,7 +650,7 @@ export async function* runSmartLoopStream(
           status: "completed",
           checkpoint: buildCheckpoint(state, "completed"),
         };
-        yield { type: "run_end" as const, result: budgetResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
+        finalResultYielded = true; yield { type: "run_end" as const, result: budgetResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
         return budgetResult;
       }
     }
@@ -671,15 +676,26 @@ export async function* runSmartLoopStream(
         pendingApproval: blockedApproval,
         checkpoint: cp,
       };
-      yield { type: "run_end" as const, result: approvalResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
+      finalResultYielded = true; yield { type: "run_end" as const, result: approvalResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
       return approvalResult;
     }
 
     // Tool calls were made
     if (toolRecords.length > 0) {
+      // Yield tool_call_start for each tool that was requested
+      for (const req of outcome.toolRequests) {
+        yield { type: "tool_call_start" as const, tool: req.name, args: req.args, iteration: state.iterations, timestamp: Date.now() };
+      }
+
       // Yield tool_call_end events for each tool execution
       for (const record of toolRecords) {
-        yield { type: "tool_call_end" as const, tool: record.tool, args: record.args, result: record.result, status: (record.status ?? "success") as "success" | "error", durationMs: 0, iteration: state.iterations, timestamp: Date.now() };
+        // Detect skill activation
+        if (record.tool === "activate_skill" && record.status !== "error") {
+          const skillResult = record.result as Record<string, unknown>;
+          yield { type: "skill_activated" as const, skill: (skillResult?.skill as string) ?? "unknown", iteration: state.iterations, timestamp: Date.now() };
+        }
+
+        yield { type: "tool_call_end" as const, tool: record.tool, args: record.args, result: record.result, status: (record.status ?? "success") as "success" | "error", iteration: state.iterations, timestamp: Date.now() };
       }
 
       // Append assistant message
@@ -938,7 +954,7 @@ export async function* runSmartLoopStream(
       status: "completed",
       checkpoint: buildCheckpoint(state, "completed"),
     };
-    yield { type: "run_end" as const, result: completedResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
+    finalResultYielded = true; yield { type: "run_end" as const, result: completedResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
     return completedResult;
   }
 
@@ -956,7 +972,7 @@ export async function* runSmartLoopStream(
       status: "completed",
       checkpoint: buildCheckpoint(state, "completed"),
     };
-    yield { type: "run_end" as const, result: forceResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
+    finalResultYielded = true; yield { type: "run_end" as const, result: forceResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
     return forceResult;
   }
 
@@ -970,6 +986,12 @@ export async function* runSmartLoopStream(
     status: "max_iterations",
     checkpoint: buildCheckpoint(state, "max_iterations"),
   };
-  yield { type: "run_end" as const, result: maxIterResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
+  finalResultYielded = true; yield { type: "run_end" as const, result: maxIterResult, durationMs: Date.now() - state.runStartTime, timestamp: Date.now() };
   return maxIterResult;
+
+  } finally {
+    if (!finalResultYielded) {
+      yield { type: "run_end" as const, result: { result: null, iterations: 0, toolCalls: [], taskCalls: [], status: "fatal" as const, error: "Stream terminated early" }, durationMs: Date.now() - runStartTime, timestamp: Date.now() };
+    }
+  }
 }

packages/ai/src/types.ts

@@ -284,7 +284,7 @@ export type AgentEvent =
   | { type: "llm_call_start"; iteration: number; messageCount: number; timestamp: number }
   | { type: "llm_call_end"; iteration: number; content: string; finishReason: string; tokensUsed?: { input: number; output: number } | undefined; durationMs: number; timestamp: number }
   | { type: "tool_call_start"; tool: string; args: unknown; iteration: number; timestamp: number }
-  | { type: "tool_call_end"; tool: string; args: unknown; result: unknown; status: "success" | "error"; durationMs: number; iteration: number; timestamp: number }
+  | { type: "tool_call_end"; tool: string; args: unknown; result: unknown; status: "success" | "error"; durationMs?: number; iteration: number; timestamp: number }
   | { type: "skill_activated"; skill: string; iteration: number; timestamp: number }
   | { type: "compression"; strategy: "snip" | "microcompact" | "autocompact" | "reactive"; tokensBefore: number; tokensAfter: number; timestamp: number }
   | { type: "memory_enrichment"; memoriesInjected: number; iteration: number; timestamp: number }

tests/unit/agent-streaming.test.ts

@@ -213,4 +213,119 @@ describe("Agent Streaming", () => {
     expect(llmEnd).toBeDefined();
     expect((llmEnd as any).content.length).toBeLessThanOrEqual(200);
   });
+
+  it("yields tool_call_start before tool_call_end", async () => {
+    const agent = createSmartAgent({
+      llm: mockLLM([JSON.stringify({ tool: "t", arguments: { x: 1 } }), "Done."]),
+      tools: [{ name: "t", description: "test", async execute(a) { return a.x; } }],
+    });
+    const events: AgentEvent[] = [];
+    const stream = agent.stream("Go");
+    let r = await stream.next();
+    while (!r.done) { events.push(r.value); r = await stream.next(); }
+
+    const startIdx = events.findIndex(e => e.type === "tool_call_start");
+    const endIdx = events.findIndex(e => e.type === "tool_call_end");
+    expect(startIdx).toBeGreaterThanOrEqual(0);
+    expect(endIdx).toBeGreaterThan(startIdx);
+  });
+
+  it("yields skill_activated when activate_skill is called", async () => {
+    const agent = createSmartAgent({
+      llm: mockLLM([JSON.stringify({ tool: "activate_skill", arguments: { skill_name: "debug" } }), "Done."]),
+      tools: [],
+      skills: [{ name: "debug", description: "Debug", trigger: "on bug", prompt: "Step 1: investigate" }],
+    });
+    const events: AgentEvent[] = [];
+    const stream = agent.stream("Debug this");
+    let r = await stream.next();
+    while (!r.done) { events.push(r.value); r = await stream.next(); }
+
+    const skillEvent = events.find(e => e.type === "skill_activated");
+    expect(skillEvent).toBeDefined();
+    expect((skillEvent as any).skill).toBe("debug");
+  });
+
+  it("yields run_end on fatal error", async () => {
+    const failLlm: LLMProvider = {
+      name: "fail",
+      async chat() { throw new Error("Permanent failure"); },
+    };
+    const agent = createSmartAgent({ llm: failLlm, tools: [], maxIterations: 2 });
+    const events: AgentEvent[] = [];
+    const stream = agent.stream("Go");
+    let r = await stream.next();
+    while (!r.done) { events.push(r.value); r = await stream.next(); }
+    const finalResult = r.value;
+
+    expect(finalResult.status).toBe("fatal");
+    const runEnd = events.find(e => e.type === "run_end");
+    expect(runEnd).toBeDefined();
+    expect((runEnd as any).result.status).toBe("fatal");
+  });
+
+  it("yields run_end on pause", async () => {
+    let callCount = 0;
+    const agent = createSmartAgent({
+      llm: { name: "mock", async chat() { callCount++; return { content: "Hi", model: "m" }; } },
+      tools: [],
+      hooks: {
+        async getControlState() {
+          return callCount === 0 ? { action: "pause" as const } : { action: "continue" as const };
+        },
+      },
+    });
+    const events: AgentEvent[] = [];
+    const stream = agent.stream("Go");
+    let r = await stream.next();
+    while (!r.done) { events.push(r.value); r = await stream.next(); }
+
+    expect(r.value.status).toBe("paused");
+    const runEnd = events.find(e => e.type === "run_end");
+    expect(runEnd).toBeDefined();
+  });
+
+  it("run() and stream() produce same final result", async () => {
+    const makeAgent = () => createSmartAgent({
+      llm: mockLLM(["The answer is 42."]),
+      tools: [],
+    });
+
+    const runResult = await makeAgent().run("What is the answer?");
+
+    const streamAgent = makeAgent();
+    const stream = streamAgent.stream("What is the answer?");
+    let r = await stream.next();
+    while (!r.done) { r = await stream.next(); }
+    const streamResult = r.value;
+
+    expect(streamResult.status).toBe(runResult.status);
+    expect(streamResult.result).toBe(runResult.result);
+    expect(streamResult.iterations).toBe(runResult.iterations);
+  });
+
+  it("concurrent tool calls produce multiple tool_call events", async () => {
+    const agent = createSmartAgent({
+      llm: mockLLM([
+        JSON.stringify([
+          { tool: "a", arguments: {} },
+          { tool: "b", arguments: {} },
+        ]),
+        "Done.",
+      ]),
+      tools: [
+        { name: "a", description: "A", isConcurrencySafe: true, async execute() { return 1; } },
+        { name: "b", description: "B", isConcurrencySafe: true, async execute() { return 2; } },
+      ],
+    });
+    const events: AgentEvent[] = [];
+    const stream = agent.stream("Go");
+    let r = await stream.next();
+    while (!r.done) { events.push(r.value); r = await stream.next(); }
+
+    const toolStarts = events.filter(e => e.type === "tool_call_start");
+    const toolEnds = events.filter(e => e.type === "tool_call_end");
+    expect(toolStarts.length).toBeGreaterThanOrEqual(2);
+    expect(toolEnds.length).toBeGreaterThanOrEqual(2);
+  });
 });