From 558dcbe3bf3679b79abb0a254b92b533f1e2e6c0 Mon Sep 17 00:00:00 2001 From: Justin Maier Date: Tue, 27 Jan 2026 11:32:22 -0700 Subject: [PATCH] feat: Add tool hooks bridge for external UI visibility Adds a bridge that connects agent runtime tool events to the internal hooks system, enabling workspace hooks to react to tool execution lifecycle events (start, update, result). This enables use cases like: - Ephemeral status messages in Discord showing tool progress - Typing indicators during tool execution - Real-time dashboards via WebSocket - Audit logging of tool usage Key changes: - New src/infra/agent-hook-bridge.ts with startAgentHookBridge() - Adds 'tool' to InternalHookEventType - Integrates bridge into gateway startup when hooks are enabled - Comprehensive test coverage for edge cases Design decisions: - Update events are NOT throttled at bridge level - hooks handle their own rate limiting (allows both throttled Discord and real-time WebSocket) - Fire-and-forget pattern so tool execution isn't blocked - Session key parsed into sessionMeta for hook convenience Closes #2904 --- src/gateway/server-startup.ts | 12 + src/hooks/internal-hooks.ts | 2 +- src/infra/agent-hook-bridge.test.ts | 400 ++++++++++++++++++++++++++++ src/infra/agent-hook-bridge.ts | 188 +++++++++++++ 4 files changed, 601 insertions(+), 1 deletion(-) create mode 100644 src/infra/agent-hook-bridge.test.ts create mode 100644 src/infra/agent-hook-bridge.ts diff --git a/src/gateway/server-startup.ts b/src/gateway/server-startup.ts index e6bdbb0dc..5b7f74cb1 100644 --- a/src/gateway/server-startup.ts +++ b/src/gateway/server-startup.ts @@ -15,6 +15,7 @@ import { triggerInternalHook, } from "../hooks/internal-hooks.js"; import { loadInternalHooks } from "../hooks/loader.js"; +import { startAgentHookBridge } from "../infra/agent-hook-bridge.js"; import type { loadMoltbotPlugins } from "../plugins/loader.js"; import { type PluginServicesHandle, startPluginServices } from "../plugins/services.js"; import { startBrowserControlServerIfEnabled } from "./server-browser.js"; @@ -111,6 +112,17 @@ export async function startGatewaySidecars(params: { params.logHooks.error(`failed to load hooks: ${String(err)}`); } + // Start the agent hook bridge to expose tool lifecycle events to hooks. + // This enables workspace hooks to react to tool:start, tool:update, tool:result events. + if (params.cfg.hooks?.internal?.enabled) { + try { + startAgentHookBridge(); + params.logHooks.info("agent hook bridge started"); + } catch (err) { + params.logHooks.error(`agent hook bridge failed to start: ${String(err)}`); + } + } + // Launch configured channels so gateway replies via the surface the message came from. // Tests can opt out via CLAWDBOT_SKIP_CHANNELS (or legacy CLAWDBOT_SKIP_PROVIDERS). const skipChannels = diff --git a/src/hooks/internal-hooks.ts b/src/hooks/internal-hooks.ts index 1b866d444..8c555ce08 100644 --- a/src/hooks/internal-hooks.ts +++ b/src/hooks/internal-hooks.ts @@ -8,7 +8,7 @@ import type { WorkspaceBootstrapFile } from "../agents/workspace.js"; import type { MoltbotConfig } from "../config/config.js"; -export type InternalHookEventType = "command" | "session" | "agent" | "gateway"; +export type InternalHookEventType = "command" | "session" | "agent" | "gateway" | "tool"; export type AgentBootstrapHookContext = { workspaceDir: string; diff --git a/src/infra/agent-hook-bridge.test.ts b/src/infra/agent-hook-bridge.test.ts new file mode 100644 index 000000000..f8fbdb63a --- /dev/null +++ b/src/infra/agent-hook-bridge.test.ts @@ -0,0 +1,400 @@ +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import { + startAgentHookBridge, + stopAgentHookBridge, + type SessionMeta, + type ToolHookEvent, +} from "./agent-hook-bridge.js"; +import { emitAgentEvent } from "./agent-events.js"; +import { + clearInternalHooks, + registerInternalHook, + type InternalHookEvent, +} from "../hooks/internal-hooks.js"; + +describe("agent-hook-bridge", () => { + beforeEach(() => { + clearInternalHooks(); + stopAgentHookBridge(); + }); + + afterEach(() => { + clearInternalHooks(); + stopAgentHookBridge(); + }); + + describe("startAgentHookBridge", () => { + it("should emit tool:start events to hooks", async () => { + const events: InternalHookEvent[] = []; + registerInternalHook("tool:start", async (evt) => { + events.push(evt); + }); + + startAgentHookBridge(); + + emitAgentEvent({ + runId: "run-1", + stream: "tool", + sessionKey: "agent:main:discord:channel:123456", + data: { + phase: "start", + name: "exec", + toolCallId: "tool-1", + args: { command: "npm test" }, + }, + }); + + // Allow async hook to process + await new Promise((r) => setTimeout(r, 10)); + + expect(events).toHaveLength(1); + expect(events[0].type).toBe("tool"); + expect(events[0].action).toBe("start"); + expect(events[0].context).toMatchObject({ + runId: "run-1", + toolName: "exec", + toolCallId: "tool-1", + args: { command: "npm test" }, + }); + }); + + it("should emit tool:update events to hooks", async () => { + const events: InternalHookEvent[] = []; + registerInternalHook("tool:update", async (evt) => { + events.push(evt); + }); + + startAgentHookBridge(); + + emitAgentEvent({ + runId: "run-1", + stream: "tool", + sessionKey: "agent:main:telegram:dm:user123", + data: { + phase: "update", + name: "exec", + toolCallId: "tool-1", + partialResult: "Building...", + }, + }); + + await new Promise((r) => setTimeout(r, 10)); + + expect(events).toHaveLength(1); + expect(events[0].action).toBe("update"); + expect(events[0].context).toMatchObject({ + toolName: "exec", + partialResult: "Building...", + }); + }); + + it("should emit tool:result events to hooks", async () => { + const events: InternalHookEvent[] = []; + registerInternalHook("tool:result", async (evt) => { + events.push(evt); + }); + + startAgentHookBridge(); + + emitAgentEvent({ + runId: "run-1", + stream: "tool", + sessionKey: "agent:main:slack:channel:C123", + data: { + phase: "result", + name: "read", + toolCallId: "tool-2", + result: { content: "file contents" }, + isError: false, + }, + }); + + await new Promise((r) => setTimeout(r, 10)); + + expect(events).toHaveLength(1); + expect(events[0].action).toBe("result"); + expect(events[0].context).toMatchObject({ + toolName: "read", + result: { content: "file contents" }, + isError: false, + }); + }); + + it("should parse sessionMeta correctly", async () => { + const events: InternalHookEvent[] = []; + registerInternalHook("tool", async (evt) => { + events.push(evt); + }); + + startAgentHookBridge(); + + emitAgentEvent({ + runId: "run-1", + stream: "tool", + sessionKey: "agent:main:discord:channel:123456789", + data: { + phase: "start", + name: "browser", + toolCallId: "tool-3", + }, + }); + + await new Promise((r) => setTimeout(r, 10)); + + const sessionMeta = (events[0].context as { sessionMeta: SessionMeta }).sessionMeta; + expect(sessionMeta).toEqual({ + agentId: "main", + platform: "discord", + channelType: "channel", + channelId: "123456789", + raw: "discord:channel:123456789", + }); + }); + + it("should handle channelIds with colons (e.g., Matrix)", async () => { + const events: InternalHookEvent[] = []; + registerInternalHook("tool", async (evt) => { + events.push(evt); + }); + + startAgentHookBridge(); + + emitAgentEvent({ + runId: "run-1", + stream: "tool", + sessionKey: "agent:main:matrix:room:!abc:matrix.org", + data: { + phase: "start", + name: "exec", + toolCallId: "tool-4", + }, + }); + + await new Promise((r) => setTimeout(r, 10)); + + const sessionMeta = (events[0].context as { sessionMeta: SessionMeta }).sessionMeta; + expect(sessionMeta.channelId).toBe("!abc:matrix.org"); + }); + + it("should not emit events for non-tool streams", async () => { + const events: InternalHookEvent[] = []; + registerInternalHook("tool", async (evt) => { + events.push(evt); + }); + + startAgentHookBridge(); + + emitAgentEvent({ + runId: "run-1", + stream: "lifecycle", + sessionKey: "agent:main:discord:channel:123", + data: { phase: "start" }, + }); + + emitAgentEvent({ + runId: "run-1", + stream: "assistant", + sessionKey: "agent:main:discord:channel:123", + data: { text: "Hello" }, + }); + + await new Promise((r) => setTimeout(r, 10)); + + expect(events).toHaveLength(0); + }); + + it("should not emit events for unknown phases", async () => { + const events: InternalHookEvent[] = []; + registerInternalHook("tool", async (evt) => { + events.push(evt); + }); + + startAgentHookBridge(); + + emitAgentEvent({ + runId: "run-1", + stream: "tool", + sessionKey: "agent:main:discord:channel:123", + data: { + phase: "unknown", + name: "exec", + toolCallId: "tool-5", + }, + }); + + await new Promise((r) => setTimeout(r, 10)); + + expect(events).toHaveLength(0); + }); + + it("should prevent double-subscription", () => { + const unsub1 = startAgentHookBridge(); + const unsub2 = startAgentHookBridge(); + + // Should return the same cleanup function + expect(unsub1).toBe(unsub2); + }); + + it("should handle malformed session keys gracefully", async () => { + const events: InternalHookEvent[] = []; + registerInternalHook("tool", async (evt) => { + events.push(evt); + }); + + startAgentHookBridge(); + + // Missing parts + emitAgentEvent({ + runId: "run-1", + stream: "tool", + sessionKey: "agent:main", + data: { + phase: "start", + name: "exec", + toolCallId: "tool-7", + }, + }); + + await new Promise((r) => setTimeout(r, 10)); + + expect(events).toHaveLength(1); + const sessionMeta = (events[0].context as { sessionMeta: SessionMeta }).sessionMeta; + expect(sessionMeta.agentId).toBeNull(); + expect(sessionMeta.platform).toBeNull(); + }); + + it("should handle empty session keys", async () => { + const events: InternalHookEvent[] = []; + registerInternalHook("tool", async (evt) => { + events.push(evt); + }); + + startAgentHookBridge(); + + emitAgentEvent({ + runId: "run-1", + stream: "tool", + sessionKey: "", + data: { + phase: "start", + name: "exec", + toolCallId: "tool-8", + }, + }); + + await new Promise((r) => setTimeout(r, 10)); + + expect(events).toHaveLength(1); + const sessionMeta = (events[0].context as { sessionMeta: SessionMeta }).sessionMeta; + expect(sessionMeta).toEqual({ + agentId: null, + platform: null, + channelType: null, + channelId: null, + raw: null, + }); + }); + + it("should normalize empty channelId to null", async () => { + const events: InternalHookEvent[] = []; + registerInternalHook("tool", async (evt) => { + events.push(evt); + }); + + startAgentHookBridge(); + + // Trailing colon resulting in empty channelId + emitAgentEvent({ + runId: "run-1", + stream: "tool", + sessionKey: "agent:main:discord:channel:", + data: { + phase: "start", + name: "exec", + toolCallId: "tool-9", + }, + }); + + await new Promise((r) => setTimeout(r, 10)); + + expect(events).toHaveLength(1); + const sessionMeta = (events[0].context as { sessionMeta: SessionMeta }).sessionMeta; + expect(sessionMeta.channelId).toBeNull(); + }); + + it("should ignore events with missing toolCallId", async () => { + const events: InternalHookEvent[] = []; + registerInternalHook("tool", async (evt) => { + events.push(evt); + }); + + startAgentHookBridge(); + + emitAgentEvent({ + runId: "run-1", + stream: "tool", + sessionKey: "agent:main:discord:channel:123", + data: { + phase: "start", + name: "exec", + // Missing toolCallId + }, + }); + + await new Promise((r) => setTimeout(r, 10)); + + expect(events).toHaveLength(0); + }); + + it("should ignore events with missing name", async () => { + const events: InternalHookEvent[] = []; + registerInternalHook("tool", async (evt) => { + events.push(evt); + }); + + startAgentHookBridge(); + + emitAgentEvent({ + runId: "run-1", + stream: "tool", + sessionKey: "agent:main:discord:channel:123", + data: { + phase: "start", + toolCallId: "tool-10", + // Missing name + }, + }); + + await new Promise((r) => setTimeout(r, 10)); + + expect(events).toHaveLength(0); + }); + }); + + describe("stopAgentHookBridge", () => { + it("should stop emitting events after stop", async () => { + const events: InternalHookEvent[] = []; + registerInternalHook("tool", async (evt) => { + events.push(evt); + }); + + startAgentHookBridge(); + stopAgentHookBridge(); + + emitAgentEvent({ + runId: "run-1", + stream: "tool", + sessionKey: "agent:main:discord:channel:123", + data: { + phase: "start", + name: "exec", + toolCallId: "tool-6", + }, + }); + + await new Promise((r) => setTimeout(r, 10)); + + expect(events).toHaveLength(0); + }); + }); +}); diff --git a/src/infra/agent-hook-bridge.ts b/src/infra/agent-hook-bridge.ts new file mode 100644 index 000000000..6ecba893b --- /dev/null +++ b/src/infra/agent-hook-bridge.ts @@ -0,0 +1,188 @@ +/** + * Agent Hook Bridge + * + * Bridges agent runtime events (tool lifecycle) to the internal hooks system, + * enabling workspace hooks to react to tool execution. + * + * Events emitted: + * - tool:start - Tool execution begins (includes args) + * - tool:update - Streaming output during execution + * - tool:result - Tool execution completes (includes result, isError) + * + * IMPORTANT: tool:update events can fire at high frequency (token-level streaming). + * Hook scripts are responsible for implementing their own throttling/debouncing + * if targeting rate-limited APIs like Discord. This design allows different hooks + * to handle updates differently (e.g., real-time WebSocket vs throttled Discord). + * + * @see https://github.com/moltbot/moltbot/issues/2904 + */ + +import { onAgentEvent, type AgentEventPayload } from "./agent-events.js"; +import { triggerInternalHook, type InternalHookEvent } from "../hooks/internal-hooks.js"; +import { parseAgentSessionKey } from "../sessions/session-key-utils.js"; + +export type ToolHookPhase = "start" | "update" | "result"; + +export type SessionMeta = { + agentId: string | null; + platform: string | null; + channelType: string | null; + channelId: string | null; + raw: string | null; +}; + +export type ToolHookContext = { + runId: string; + toolName: string; + toolCallId: string; + args?: unknown; + result?: unknown; + partialResult?: unknown; + isError?: boolean; + meta: Record; + sessionMeta: SessionMeta; +}; + +export type ToolHookEvent = Omit & { + type: "tool"; + action: ToolHookPhase; + context: ToolHookContext; +}; + +/** + * Parse session key into structured metadata for hook convenience. + * + * Session keys look like: `agent:main:discord:channel:123456789` + * The `rest` field contains: `discord:channel:123456789` + */ +function parseSessionMeta(sessionKey: string | undefined | null): SessionMeta { + const parsed = parseAgentSessionKey(sessionKey); + if (!parsed) { + return { + agentId: null, + platform: null, + channelType: null, + channelId: null, + raw: null, + }; + } + + const { agentId, rest } = parsed; + const parts = rest.split(":"); + + // Handle IDs that might contain colons (e.g., Matrix IDs) + const rawChannelId = parts.length > 2 ? parts.slice(2).join(":") : null; + // Normalize empty strings to null + const channelId = rawChannelId && rawChannelId.trim() ? rawChannelId : null; + + return { + agentId, + platform: parts[0] ?? null, + channelType: parts[1] ?? null, + channelId, + raw: rest, + }; +} + +/** + * Type guard for tool event data + */ +function isToolEventData( + data: Record | undefined, +): data is { + phase: string; + name: string; + toolCallId: string; + args?: unknown; + result?: unknown; + partialResult?: unknown; + isError?: boolean; + meta?: Record; +} { + if (!data) return false; + return ( + typeof data.phase === "string" && + typeof data.name === "string" && + typeof data.toolCallId === "string" + ); +} + +/** + * Check if a phase is a valid tool hook phase + */ +function isToolHookPhase(phase: string): phase is ToolHookPhase { + return phase === "start" || phase === "update" || phase === "result"; +} + +let unsubscribe: (() => void) | null = null; + +/** + * Start the agent hook bridge. + * + * Subscribes to agent events and translates tool lifecycle events + * to internal hook events. Should be called during gateway startup + * after hooks have been loaded. + * + * @returns Cleanup function to stop the bridge + */ +export function startAgentHookBridge(): () => void { + // Prevent double-subscription + if (unsubscribe) { + return unsubscribe; + } + + unsubscribe = onAgentEvent((evt: AgentEventPayload) => { + // Only care about tool streams + if (evt.stream !== "tool") return; + + // Validate event data structure + if (!isToolEventData(evt.data)) return; + + const { phase, name, toolCallId, args, result, partialResult, isError, meta } = evt.data; + + // Only bridge known phases + if (!isToolHookPhase(phase)) return; + + const hookPayload: ToolHookEvent = { + type: "tool", + action: phase, + sessionKey: evt.sessionKey ?? "", + timestamp: new Date(evt.ts), + // Explicitly undefined to signal this isn't a chat message event + messages: [], + context: { + runId: evt.runId, + toolName: name, + toolCallId, + // Only include relevant data per phase to reduce payload size + args: phase === "start" ? args : undefined, + result: phase === "result" ? result : undefined, + partialResult: phase === "update" ? partialResult : undefined, + isError: phase === "result" ? isError : undefined, + meta: meta ?? {}, + // Parsed session metadata so hooks don't need to parse session keys + sessionMeta: parseSessionMeta(evt.sessionKey), + }, + }; + + // Fire and forget - don't block tool execution on hook processing + void triggerInternalHook(hookPayload as InternalHookEvent); + }); + + return () => { + if (unsubscribe) { + unsubscribe(); + unsubscribe = null; + } + }; +} + +/** + * Stop the agent hook bridge if running. + */ +export function stopAgentHookBridge(): void { + if (unsubscribe) { + unsubscribe(); + unsubscribe = null; + } +}