Merge 10f7436a68e6ebb283f69a68c5c041fa98679647 into 598f1826d8b2bc969aace2c6459824737667218c

src/agents/pi-embedded-runner/run/attempt.ts

@@ -2554,6 +2554,7 @@ export async function runEmbeddedAttempt(
       const {
         assistantTexts,
         toolMetas,
+        getTotalToolCallCount,
         unsubscribe,
         waitForCompactionRetry,
         isCompactionInFlight,
@@ -3071,6 +3072,7 @@ export async function runEmbeddedAttempt(
         // This is fire-and-forget, so we don't await
         // Run even on compaction timeout so plugins can log/cleanup
         if (hookRunner?.hasHooks("agent_end")) {
+          const usage = getUsageTotals?.();
           hookRunner
             .runAgentEnd(
               {
@@ -3078,6 +3080,16 @@ export async function runEmbeddedAttempt(
                 success: !aborted && !promptError,
                 error: promptError ? describeUnknownError(promptError) : undefined,
                 durationMs: Date.now() - promptStartedAt,
+                tokenUsage: usage
+                  ? {
+                      input: usage.input,
+                      output: usage.output,
+                      cacheRead: usage.cacheRead,
+                      cacheWrite: usage.cacheWrite,
+                      total: usage.total,
+                    }
+                  : undefined,
+                toolCallCount: getTotalToolCallCount(),
               },
               {
                 agentId: hookAgentId,

src/agents/pi-embedded-subscribe.handlers.messages.ts

@@ -3,6 +3,7 @@ import { resolveSendableOutboundReplyParts } from "openclaw/plugin-sdk/reply-pay
 import { parseReplyDirectives } from "../auto-reply/reply/reply-directives.js";
 import { SILENT_REPLY_TOKEN } from "../auto-reply/tokens.js";
 import { emitAgentEvent } from "../infra/agent-events.js";
+import { getGlobalHookRunner } from "../plugins/hook-runner-global.js";
 import { createInlineCodeState } from "../markdown/code-spans.js";
 import {
   isMessagingToolDuplicateNormalized,
@@ -122,7 +123,21 @@ export function handleMessageUpdate(
 
   if (evtType === "thinking_start" || evtType === "thinking_delta" || evtType === "thinking_end") {
     if (evtType === "thinking_start" || evtType === "thinking_delta") {
-      ctx.state.reasoningStreamOpen = true;
+      if (!ctx.state.reasoningStreamOpen) {
+        ctx.state.reasoningStreamOpen = true;
+        ctx.state.thinkingStartedAt = Date.now();
+        // Emit thinking_start hook on first thinking event
+        const hookRunner = ctx.hookRunner ?? getGlobalHookRunner();
+        if (hookRunner?.hasHooks("thinking_start")) {
+          void hookRunner.runThinkingStart(
+            { runId: ctx.params.runId },
+            {
+              agentId: ctx.params.agentId,
+              sessionKey: ctx.params.sessionKey,
+            },
+          );
+        }
+      }
     }
     const thinkingDelta = typeof assistantRecord?.delta === "string" ? assistantRecord.delta : "";
     const thinkingContent =
@@ -142,10 +157,40 @@ export function handleMessageUpdate(
       ctx.emitReasoningStream(partialThinking || thinkingContent || thinkingDelta);
     }
     if (evtType === "thinking_end") {
+      // Ensure plugins always get a matching thinking_start before thinking_end
       if (!ctx.state.reasoningStreamOpen) {
         ctx.state.reasoningStreamOpen = true;
+        ctx.state.thinkingStartedAt = Date.now();
+        const hookRunner = ctx.hookRunner ?? getGlobalHookRunner();
+        if (hookRunner?.hasHooks("thinking_start")) {
+          void hookRunner.runThinkingStart(
+            { runId: ctx.params.runId },
+            {
+              agentId: ctx.params.agentId,
+              sessionKey: ctx.params.sessionKey,
+            },
+          );
+        }
       }
       emitReasoningEnd(ctx);
+      // Emit thinking_end hook
+      const hookRunner = ctx.hookRunner ?? getGlobalHookRunner();
+      if (hookRunner?.hasHooks("thinking_end")) {
+        const fullThinking = extractAssistantThinking(msg);
+        void hookRunner.runThinkingEnd(
+          {
+            runId: ctx.params.runId,
+            text: fullThinking || thinkingContent || undefined,
+            durationMs: ctx.state.thinkingStartedAt
+              ? Date.now() - ctx.state.thinkingStartedAt
+              : undefined,
+          },
+          {
+            agentId: ctx.params.agentId,
+            sessionKey: ctx.params.sessionKey,
+          },
+        );
+      }
     }
     return;
   }
@@ -207,16 +252,51 @@ export function handleMessageUpdate(
       inlineCode: createInlineCodeState(),
     })
     .trim();
+
+  // Track <think> tag transitions independently of visible text.
+  // stripBlockTags updates partialBlockState as a side effect, so we must
+  // call it even when `next` is empty (pure-thinking chunks).
+  const wasThinking = ctx.state.partialBlockState.thinking;
+  const visibleDelta = chunk ? ctx.stripBlockTags(chunk, ctx.state.partialBlockState) : "";
+  if (!wasThinking && ctx.state.partialBlockState.thinking) {
+    ctx.state.reasoningStreamOpen = true;
+    ctx.state.thinkingStartedAt = Date.now();
+    // Fire thinking_start hook for <think> tag flows
+    const hookRunner = ctx.hookRunner ?? getGlobalHookRunner();
+    if (hookRunner?.hasHooks("thinking_start")) {
+      void hookRunner.runThinkingStart(
+        { runId: ctx.params.runId },
+        {
+          agentId: ctx.params.agentId,
+          sessionKey: ctx.params.sessionKey,
+        },
+      );
+    }
+  }
+  // Detect when thinking block ends (</think> tag processed)
+  if (wasThinking && !ctx.state.partialBlockState.thinking) {
+    emitReasoningEnd(ctx);
+    // Fire thinking_end hook for <think> tag flows
+    const hookRunner = ctx.hookRunner ?? getGlobalHookRunner();
+    if (hookRunner?.hasHooks("thinking_end")) {
+      const fullThinking = extractThinkingFromTaggedText(ctx.state.deltaBuffer);
+      void hookRunner.runThinkingEnd(
+        {
+          runId: ctx.params.runId,
+          text: fullThinking || undefined,
+          durationMs: ctx.state.thinkingStartedAt
+            ? Date.now() - ctx.state.thinkingStartedAt
+            : undefined,
+        },
+        {
+          agentId: ctx.params.agentId,
+          sessionKey: ctx.params.sessionKey,
+        },
+      );
+    }
+  }
+
   if (next) {
-    const wasThinking = ctx.state.partialBlockState.thinking;
-    const visibleDelta = chunk ? ctx.stripBlockTags(chunk, ctx.state.partialBlockState) : "";
-    if (!wasThinking && ctx.state.partialBlockState.thinking) {
-      ctx.state.reasoningStreamOpen = true;
-    }
-    // Detect when thinking block ends (</think> tag processed)
-    if (wasThinking && !ctx.state.partialBlockState.thinking) {
-      emitReasoningEnd(ctx);
-    }
     const parsedDelta = visibleDelta ? ctx.consumePartialReplyDirectives(visibleDelta) : null;
     const parsedFull = parseReplyDirectives(stripTrailingDirective(next));
     const cleanedText = parsedFull.text;

src/agents/pi-embedded-subscribe.handlers.tools.ts

@@ -441,6 +441,7 @@ export async function handleToolExecutionEnd(
   const callSummary = ctx.state.toolMetaById.get(toolCallId);
   const meta = callSummary?.meta;
   ctx.state.toolMetas.push({ toolName, meta });
+  ctx.state.totalToolCallCount++;
   ctx.state.toolMetaById.delete(toolCallId);
   ctx.state.toolSummaryById.delete(toolCallId);
   if (isToolError) {

src/agents/pi-embedded-subscribe.handlers.types.ts

@@ -33,6 +33,7 @@ export type ToolCallSummary = {
 export type EmbeddedPiSubscribeState = {
   assistantTexts: string[];
   toolMetas: Array<{ toolName?: string; meta?: string }>;
+  totalToolCallCount: number;
   toolMetaById: Map<string, ToolCallSummary>;
   toolSummaryById: Set<string>;
   lastToolError?: ToolErrorSummary;
@@ -53,6 +54,7 @@ export type EmbeddedPiSubscribeState = {
   lastStreamedReasoning?: string;
   lastBlockReplyText?: string;
   reasoningStreamOpen: boolean;
+  thinkingStartedAt?: number;
   assistantMessageIndex: number;
   lastAssistantTextMessageIndex: number;
   lastAssistantTextNormalized?: string;

src/agents/pi-embedded-subscribe.ts

@@ -38,6 +38,7 @@ export function subscribeEmbeddedPiSession(params: SubscribeEmbeddedPiSessionPar
   const state: EmbeddedPiSubscribeState = {
     assistantTexts: [],
     toolMetas: [],
+    totalToolCallCount: 0,
     toolMetaById: new Map(),
     toolSummaryById: new Set(),
     lastToolError: undefined,
@@ -679,6 +680,7 @@ export function subscribeEmbeddedPiSession(params: SubscribeEmbeddedPiSessionPar
   return {
     assistantTexts,
     toolMetas,
+    getTotalToolCallCount: () => state.totalToolCallCount,
     unsubscribe,
     isCompacting: () => state.compactionInFlight || state.pendingCompactionRetry > 0,
     isCompactionInFlight: () => state.compactionInFlight,

src/infra/agent-events.ts

@@ -1,6 +1,12 @@
 import type { VerboseLevel } from "../auto-reply/thinking.js";
 
-export type AgentEventStream = "lifecycle" | "tool" | "assistant" | "error" | (string & {});
+export type AgentEventStream =
+  | "lifecycle"
+  | "tool"
+  | "thinking"
+  | "assistant"
+  | "error"
+  | (string & {});
 
 export type AgentEventPayload = {
   runId: string;

src/plugins/hooks.ts

@@ -30,6 +30,8 @@ import type {
   PluginHookGatewayContext,
   PluginHookGatewayStartEvent,
   PluginHookGatewayStopEvent,
+  PluginHookThinkingStartEvent,
+  PluginHookThinkingEndEvent,
   PluginHookMessageContext,
   PluginHookMessageReceivedEvent,
   PluginHookMessageSendingEvent,
@@ -100,6 +102,8 @@ export type {
   PluginHookGatewayContext,
   PluginHookGatewayStartEvent,
   PluginHookGatewayStopEvent,
+  PluginHookThinkingStartEvent,
+  PluginHookThinkingEndEvent,
 };
 
 export type HookRunnerLogger = {
@@ -900,6 +904,30 @@ export function createHookRunner(registry: PluginRegistry, options: HookRunnerOp
     return runVoidHook("gateway_stop", event, ctx);
   }
 
+  /**
+   * Run thinking_start hook.
+   * Fires when the agent begins a thinking/reasoning phase.
+   * Runs in parallel (fire-and-forget).
+   */
+  async function runThinkingStart(
+    event: PluginHookThinkingStartEvent,
+    ctx: PluginHookAgentContext,
+  ): Promise<void> {
+    return runVoidHook("thinking_start", event, ctx);
+  }
+
+  /**
+   * Run thinking_end hook.
+   * Fires when the agent completes a thinking/reasoning phase.
+   * Runs in parallel (fire-and-forget).
+   */
+  async function runThinkingEnd(
+    event: PluginHookThinkingEndEvent,
+    ctx: PluginHookAgentContext,
+  ): Promise<void> {
+    return runVoidHook("thinking_end", event, ctx);
+  }
+
   // =========================================================================
   // Utility
   // =========================================================================
@@ -952,6 +980,9 @@ export function createHookRunner(registry: PluginRegistry, options: HookRunnerOp
     // Gateway hooks
     runGatewayStart,
     runGatewayStop,
+    // Thinking hooks
+    runThinkingStart,
+    runThinkingEnd,
     // Utility
     hasHooks,
     getHookCount,

src/plugins/registry.ts

@@ -2,6 +2,7 @@ import path from "node:path";
 import type { AnyAgentTool } from "../agents/tools/common.js";
 import type { ChannelPlugin } from "../channels/plugins/types.js";
 import { registerContextEngineForOwner } from "../context-engine/registry.js";
+import { onAgentEvent, type AgentEventPayload } from "../infra/agent-events.js";
 import type {
   GatewayRequestHandler,
   GatewayRequestHandlers,
@@ -980,6 +981,19 @@ export function createPluginRegistry(registryParams: PluginRegistryParams) {
         }
       },
       resolvePath: (input: string) => resolveUserPath(input),
+      onAgentEvent: registrationMode === "full"
+        ? (listener: (evt: AgentEventPayload) => void, filter?: { sessionKey?: string }) => {
+            if (!filter?.sessionKey) {
+              return onAgentEvent(listener);
+            }
+            const sk = filter.sessionKey;
+            return onAgentEvent((evt) => {
+              if (evt.sessionKey === sk) {
+                listener(evt);
+              }
+            });
+          }
+        : () => () => {},
       on: (hookName, handler, opts) =>
         registrationMode === "full"
           ? registerTypedHook(record, hookName, handler, opts, params.hookPolicy)

src/plugins/types.ts

@@ -2,6 +2,7 @@ import type { IncomingMessage, ServerResponse } from "node:http";
 import type { AgentMessage } from "@mariozechner/pi-agent-core";
 import type { StreamFn } from "@mariozechner/pi-agent-core";
 import type { Api, Model } from "@mariozechner/pi-ai";
+import type { AgentEventPayload } from "../infra/agent-events.js";
 import type { ModelRegistry } from "@mariozechner/pi-coding-agent";
 import type { Command } from "commander";
 import type {
@@ -1342,6 +1343,22 @@ export type OpenClawPluginApi = {
     factory: import("../context-engine/registry.js").ContextEngineFactory,
   ) => void;
   resolvePath: (input: string) => string;
+  /**
+   * Subscribe to the real-time agent event stream.
+   * Receives all agent events (thinking deltas, tool start/result, assistant text,
+   * lifecycle phases) as they happen. Returns an unsubscribe function.
+   *
+   * For high-frequency events (thinking deltas, assistant text), prefer this over
+   * hooks. For lifecycle moments (tool complete, agent end), prefer `on()` hooks.
+   *
+   * @param listener - Callback invoked for each event.
+   * @param filter - Optional filter. When `sessionKey` is provided, only events
+   *   matching that session are delivered. Omit for the global firehose.
+   */
+  onAgentEvent: (
+    listener: (evt: AgentEventPayload) => void,
+    filter?: { sessionKey?: string },
+  ) => () => void;
   /** Register a lifecycle hook handler */
   on: <K extends PluginHookName>(
     hookName: K,
@@ -1391,6 +1408,8 @@ export type PluginHookName =
   | "subagent_delivery_target"
   | "subagent_spawned"
   | "subagent_ended"
+  | "thinking_start"
+  | "thinking_end"
   | "gateway_start"
   | "gateway_stop";
 
@@ -1418,6 +1437,8 @@ export const PLUGIN_HOOK_NAMES = [
   "subagent_delivery_target",
   "subagent_spawned",
   "subagent_ended",
+  "thinking_start",
+  "thinking_end",
   "gateway_start",
   "gateway_stop",
 ] as const satisfies readonly PluginHookName[];
@@ -1573,6 +1594,16 @@ export type PluginHookAgentEndEvent = {
   success: boolean;
   error?: string;
   durationMs?: number;
+  /** Token usage totals for this agent run. */
+  tokenUsage?: {
+    input: number;
+    output: number;
+    cacheRead: number;
+    cacheWrite: number;
+    total: number;
+  };
+  /** Number of tool calls made during this agent run. */
+  toolCallCount?: number;
 };
 
 // Compaction hooks
@@ -1865,6 +1896,22 @@ export type PluginHookGatewayStopEvent = {
   reason?: string;
 };
 
+// thinking_start hook
+export type PluginHookThinkingStartEvent = {
+  /** Stable run identifier for this agent invocation. */
+  runId?: string;
+};
+
+// thinking_end hook
+export type PluginHookThinkingEndEvent = {
+  /** Stable run identifier for this agent invocation. */
+  runId?: string;
+  /** Full thinking/reasoning text. */
+  text?: string;
+  /** Duration of the thinking phase in milliseconds. */
+  durationMs?: number;
+};
+
 // Hook handler types mapped by hook name
 export type PluginHookHandlerMap = {
   before_model_resolve: (
@@ -1967,6 +2014,14 @@ export type PluginHookHandlerMap = {
     event: PluginHookGatewayStopEvent,
     ctx: PluginHookGatewayContext,
   ) => Promise<void> | void;
+  thinking_start: (
+    event: PluginHookThinkingStartEvent,
+    ctx: PluginHookAgentContext,
+  ) => Promise<void> | void;
+  thinking_end: (
+    event: PluginHookThinkingEndEvent,
+    ctx: PluginHookAgentContext,
+  ) => Promise<void> | void;
 };
 
 export type PluginHookRegistration<K extends PluginHookName = PluginHookName> = {