From 3a1ca0ea8bb2dc5d0c560064b2d4459dd207f654 Mon Sep 17 00:00:00 2001 From: Rex Liu Date: Mon, 26 Jan 2026 22:01:54 -0500 Subject: [PATCH] feat(hooks): Add message:received hook event Implements the message:received hook event that fires when a user message is received, before processing. This enables hooks to: - Track user activity in real-time - Update state files without polling - React to incoming messages Changes: - Add 'message' to InternalHookEventType union - Fire message:received event in dispatchInboundMessage() - Event fires async to avoid blocking message processing - Include context: senderId, channel, messageBody (truncated), timestamp, cfg New bundled hook: - activity-tracker: Updates heartbeat-state.json on every user message - Sets lastUserMessage to current timestamp - Sets backupActive=true to enable hourly backups - Enables session-backup hook to work without polling Tests: - message-received.test.ts: Tests for the new hook event Docs: - Updated hooks.md to document message:received event - Moved from 'Future Events' to 'Message Events' section --- docs/hooks.md | 14 +- src/auto-reply/dispatch.ts | 22 ++++ src/hooks/bundled/activity-tracker/HOOK.md | 59 +++++++++ src/hooks/bundled/activity-tracker/handler.ts | 99 ++++++++++++++ src/hooks/internal-hooks.ts | 2 +- test/hooks/message-received.test.ts | 121 ++++++++++++++++++ 6 files changed, 315 insertions(+), 2 deletions(-) create mode 100644 src/hooks/bundled/activity-tracker/HOOK.md create mode 100644 src/hooks/bundled/activity-tracker/handler.ts create mode 100644 test/hooks/message-received.test.ts diff --git a/docs/hooks.md b/docs/hooks.md index 9870dda88..753b1b8d6 100644 --- a/docs/hooks.md +++ b/docs/hooks.md @@ -241,6 +241,19 @@ These hooks are not event-stream listeners; they let plugins synchronously adjus - **`tool_result_persist`**: transform tool results before they are written to the session transcript. Must be synchronous; return the updated tool result payload or `undefined` to keep it as-is. See [Agent Loop](/concepts/agent-loop). +### Message Events + +Triggered when messages are processed: + +- **`message:received`**: When a user message is received (fires before processing) + +Context includes: +- `senderId`: The sender's ID +- `channel`: The channel the message came from +- `messageBody`: The message content (truncated to 500 chars) +- `timestamp`: When the message was received +- `cfg`: The current config + ### Future Events Planned event types: @@ -249,7 +262,6 @@ Planned event types: - **`session:end`**: When a session ends - **`agent:error`**: When an agent encounters an error - **`message:sent`**: When a message is sent -- **`message:received`**: When a message is received ## Creating Custom Hooks diff --git a/src/auto-reply/dispatch.ts b/src/auto-reply/dispatch.ts index 8b1626c3d..fce235fce 100644 --- a/src/auto-reply/dispatch.ts +++ b/src/auto-reply/dispatch.ts @@ -11,6 +11,7 @@ import { type ReplyDispatcherOptions, type ReplyDispatcherWithTypingOptions, } from "./reply/reply-dispatcher.js"; +import { createInternalHookEvent, triggerInternalHook } from "../hooks/internal-hooks.js"; export type DispatchInboundResult = DispatchFromConfigResult; @@ -22,6 +23,27 @@ export async function dispatchInboundMessage(params: { replyResolver?: typeof import("./reply.js").getReplyFromConfig; }): Promise { const finalized = finalizeInboundContext(params.ctx); + + // Trigger message:received hook for user messages + // This enables hooks to track activity, update state, etc. + const sessionKey = finalized.SessionKey ?? ""; + const senderId = finalized.SenderId ?? finalized.From ?? ""; + const channel = finalized.OriginatingChannel ?? finalized.Channel ?? ""; + const messageBody = finalized.Body ?? finalized.RawBody ?? ""; + + const hookEvent = createInternalHookEvent("message", "received", sessionKey, { + senderId, + channel, + messageBody: typeof messageBody === "string" ? messageBody.slice(0, 500) : "", + timestamp: Date.now(), + cfg: params.cfg, + }); + + // Fire hook asynchronously - don't block message processing + void triggerInternalHook(hookEvent).catch((err) => { + console.error("[message:received hook]", err instanceof Error ? err.message : String(err)); + }); + return await dispatchReplyFromConfig({ ctx: finalized, cfg: params.cfg, diff --git a/src/hooks/bundled/activity-tracker/HOOK.md b/src/hooks/bundled/activity-tracker/HOOK.md new file mode 100644 index 000000000..2f0655a93 --- /dev/null +++ b/src/hooks/bundled/activity-tracker/HOOK.md @@ -0,0 +1,59 @@ +--- +name: activity-tracker +description: "Tracks user activity for memory backup lifecycle management" +homepage: https://docs.clawd.bot/hooks#activity-tracker +metadata: + { + "clawdbot": + { + "emoji": "📍", + "events": ["message:received"], + "requires": { "config": ["workspace.dir"] }, + "install": [{ "id": "bundled", "kind": "bundled", "label": "Bundled with Clawdbot" }], + }, + } +--- + +# Activity Tracker Hook + +Tracks user message activity for memory backup lifecycle management. + +## What It Does + +When a user sends a message (`message:received` event): + +1. Updates `memory/heartbeat-state.json` with current timestamp +2. Sets `backupActive = true` to enable hourly backups +3. Fires asynchronously - doesn't block message processing + +## Why It Matters + +This hook enables the session-backup hook to know when the user is active without: +- Polling session logs repeatedly +- Relying on the agent to remember to update state +- Adding latency to message processing + +## State File + +Updates `memory/heartbeat-state.json`: + +```json +{ + "lastUserMessage": 1234567890, + "backupActive": true, + ... +} +``` + +## Requirements + +- **Config**: `workspace.dir` must be set + +## Disabling + +```bash +clawdbot hooks disable activity-tracker +``` + +If disabled, the session-backup hook will still work but will rely on +polling session logs for activity detection (less efficient). diff --git a/src/hooks/bundled/activity-tracker/handler.ts b/src/hooks/bundled/activity-tracker/handler.ts new file mode 100644 index 000000000..30e4c0454 --- /dev/null +++ b/src/hooks/bundled/activity-tracker/handler.ts @@ -0,0 +1,99 @@ +/** + * Activity tracker hook handler + * + * Updates heartbeat-state.json when user messages are received. + * This enables session-backup to know when user is active without polling. + */ + +import fs from "node:fs/promises"; +import path from "node:path"; +import os from "node:os"; +import type { ClawdbotConfig } from "../../../config/config.js"; +import { resolveAgentWorkspaceDir } from "../../../agents/agent-scope.js"; +import { resolveAgentIdFromSessionKey } from "../../../routing/session-key.js"; +import type { HookHandler } from "../../hooks.js"; + +// ============================================ +// Types +// ============================================ + +interface BackupState { + lastBackup: number | null; + lastDistill: number | null; + lastSessionSummary: number | null; + lastUserMessage: number | null; + backupActive: boolean; +} + +// ============================================ +// State Management +// ============================================ + +async function readState(stateFile: string): Promise { + try { + const content = await fs.readFile(stateFile, "utf-8"); + return JSON.parse(content); + } catch { + return { + lastBackup: null, + lastDistill: null, + lastSessionSummary: null, + lastUserMessage: null, + backupActive: false, + }; + } +} + +async function writeState(stateFile: string, state: BackupState): Promise { + await fs.writeFile(stateFile, JSON.stringify(state, null, 2), "utf-8"); +} + +// ============================================ +// Main Hook Handler +// ============================================ + +const activityTrackerHandler: HookHandler = async (event) => { + // Only trigger on message:received events + if (event.type !== "message" || event.action !== "received") { + return; + } + + const context = event.context || {}; + const cfg = context.cfg as ClawdbotConfig | undefined; + + if (!cfg) { + // Config not available, skip silently + return; + } + + try { + const agentId = resolveAgentIdFromSessionKey(event.sessionKey); + const workspaceDir = resolveAgentWorkspaceDir(cfg, agentId); + const memoryDir = path.join(workspaceDir, "memory"); + const stateFile = path.join(memoryDir, "heartbeat-state.json"); + + // Ensure directory exists + await fs.mkdir(memoryDir, { recursive: true }); + + // Read current state + const state = await readState(stateFile); + + // Update activity + const now = Math.floor(Date.now() / 1000); + state.lastUserMessage = now; + state.backupActive = true; + + // Write updated state + await writeState(stateFile, state); + + console.log(`[activity-tracker] Updated: lastUserMessage=${now}, backupActive=true`); + } catch (err) { + // Log but don't throw - we don't want to affect message processing + console.error( + "[activity-tracker] Failed to update state:", + err instanceof Error ? err.message : String(err) + ); + } +}; + +export default activityTrackerHandler; diff --git a/src/hooks/internal-hooks.ts b/src/hooks/internal-hooks.ts index 6c8cc676f..b239c8dfd 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 { ClawdbotConfig } from "../config/config.js"; -export type InternalHookEventType = "command" | "session" | "agent" | "gateway"; +export type InternalHookEventType = "command" | "session" | "agent" | "gateway" | "message"; export type AgentBootstrapHookContext = { workspaceDir: string; diff --git a/test/hooks/message-received.test.ts b/test/hooks/message-received.test.ts new file mode 100644 index 000000000..1f61db5a5 --- /dev/null +++ b/test/hooks/message-received.test.ts @@ -0,0 +1,121 @@ +/** + * Tests for message:received hook event + */ + +import { describe, it, expect, beforeEach, afterEach, vi } from "vitest"; +import { + registerInternalHook, + unregisterInternalHook, + clearInternalHooks, + createInternalHookEvent, + triggerInternalHook, +} from "../../src/hooks/internal-hooks.js"; + +describe("message:received hook event", () => { + beforeEach(() => { + clearInternalHooks(); + }); + + afterEach(() => { + clearInternalHooks(); + }); + + it("triggers handler on message:received event", async () => { + const handler = vi.fn(); + registerInternalHook("message:received", handler); + + const event = createInternalHookEvent("message", "received", "test-session", { + senderId: "user123", + channel: "telegram", + messageBody: "Hello", + timestamp: Date.now(), + }); + + await triggerInternalHook(event); + + expect(handler).toHaveBeenCalledTimes(1); + expect(handler).toHaveBeenCalledWith(event); + }); + + it("includes message context in event", async () => { + let capturedEvent: any = null; + const handler = vi.fn((event) => { + capturedEvent = event; + }); + registerInternalHook("message:received", handler); + + const event = createInternalHookEvent("message", "received", "agent:main:dm:user123", { + senderId: "user123", + channel: "telegram", + messageBody: "Test message content", + timestamp: 1234567890, + }); + + await triggerInternalHook(event); + + expect(capturedEvent).not.toBeNull(); + expect(capturedEvent.type).toBe("message"); + expect(capturedEvent.action).toBe("received"); + expect(capturedEvent.sessionKey).toBe("agent:main:dm:user123"); + expect(capturedEvent.context.senderId).toBe("user123"); + expect(capturedEvent.context.channel).toBe("telegram"); + expect(capturedEvent.context.messageBody).toBe("Test message content"); + expect(capturedEvent.context.timestamp).toBe(1234567890); + }); + + it("allows multiple handlers for message:received", async () => { + const handler1 = vi.fn(); + const handler2 = vi.fn(); + + registerInternalHook("message:received", handler1); + registerInternalHook("message:received", handler2); + + const event = createInternalHookEvent("message", "received", "test-session", {}); + await triggerInternalHook(event); + + expect(handler1).toHaveBeenCalledTimes(1); + expect(handler2).toHaveBeenCalledTimes(1); + }); + + it("also triggers general message type handler", async () => { + const generalHandler = vi.fn(); + const specificHandler = vi.fn(); + + registerInternalHook("message", generalHandler); + registerInternalHook("message:received", specificHandler); + + const event = createInternalHookEvent("message", "received", "test-session", {}); + await triggerInternalHook(event); + + expect(generalHandler).toHaveBeenCalledTimes(1); + expect(specificHandler).toHaveBeenCalledTimes(1); + }); + + it("does not throw if handler errors", async () => { + const failingHandler = vi.fn(() => { + throw new Error("Handler error"); + }); + registerInternalHook("message:received", failingHandler); + + const event = createInternalHookEvent("message", "received", "test-session", {}); + + // Should not throw + await expect(triggerInternalHook(event)).resolves.not.toThrow(); + }); + + it("continues to other handlers if one fails", async () => { + const failingHandler = vi.fn(() => { + throw new Error("First handler error"); + }); + const successHandler = vi.fn(); + + registerInternalHook("message:received", failingHandler); + registerInternalHook("message:received", successHandler); + + const event = createInternalHookEvent("message", "received", "test-session", {}); + await triggerInternalHook(event); + + expect(failingHandler).toHaveBeenCalled(); + expect(successHandler).toHaveBeenCalled(); + }); +});