From 9a471697d8021a23b31d5c03b25b55dbeae51caf Mon Sep 17 00:00:00 2001 From: Mason Hall Date: Mon, 31 Aug 2026 18:45:30 -0400 Subject: [PATCH] Add evlog observability --- agent/channels/linq.ts | 124 +++++++++-- agent/hooks/evlog.ts | 11 + agent/instrumentation.ts | 7 + package.json | 1 + pnpm-lock.yaml | 74 +++++++ .../channels/linq-message-delivery.test.ts | 197 +++++++++++++++++- tests/agent/hooks/evlog.test.ts | 157 ++++++++++++++ 7 files changed, 556 insertions(+), 15 deletions(-) create mode 100644 agent/hooks/evlog.ts create mode 100644 agent/instrumentation.ts create mode 100644 tests/agent/hooks/evlog.test.ts diff --git a/agent/channels/linq.ts b/agent/channels/linq.ts index 47e6b88a..c289a83c 100644 --- a/agent/channels/linq.ts +++ b/agent/channels/linq.ts @@ -6,6 +6,8 @@ import { type LinqChannelConfig, type LinqChannelCredentials, } from "eve/channels/linq"; +import { parseError } from "evlog"; +import { useLogger as getEvlog } from "evlog/eve"; import { z } from "zod"; import { getAuth } from "@/auth"; import { normalizeAuthPhoneNumber } from "@/auth/phone-number"; @@ -118,21 +120,115 @@ export const linqChannelConfig = { .map((line) => line.trim()) .find(Boolean) ?? null) : null; - if (context.thread) { - const messageId = context.thread.toJSON().currentMessage?.id; - if ( - messageId && - context.state.acknowledgedLinqMessageId !== messageId - ) { - try { - await context.bot - .getAdapter("linq") - .addReaction(context.thread.id, messageId, "thumbs_up"); - context.state.acknowledgedLinqMessageId = messageId; - } catch { - // SMS/RCS and some carrier paths do not support iMessage tapbacks. - } + let log: ReturnType | undefined; + try { + log = getEvlog(session); + } catch (error) { + console.warn("[linq] evlog unavailable", { + error: parseError(error), + sessionId: session.session.id, + turnId: event.turnId, + }); + } + if (!context.thread) { + const reaction = { outcome: "missing-thread" }; + if (log) { + log.warn("Linq reaction skipped", { + channel: { linq: { reactions: [reaction] } }, + }); + } else { + console.warn("[linq] reaction skipped", { + ...reaction, + sessionId: session.session.id, + turnId: event.turnId, + }); + } + return; + } + + const messageId = context.thread.toJSON().currentMessage?.id; + if (!messageId) { + const reaction = { + outcome: "missing-message-id", + threadId: context.thread.id, + }; + if (log) { + log.warn("Linq reaction skipped", { + channel: { linq: { reactions: [reaction] } }, + }); + } else { + console.warn("[linq] reaction skipped", { + ...reaction, + sessionId: session.session.id, + turnId: event.turnId, + }); + } + return; + } + + if (context.state.acknowledgedLinqMessageId === messageId) { + const reaction = { + messageId, + outcome: "already-acknowledged", + threadId: context.thread.id, + }; + if (log) { + log.info("Linq reaction skipped", { + channel: { linq: { reactions: [reaction] } }, + }); + } else { + console.info("[linq] reaction skipped", { + ...reaction, + sessionId: session.session.id, + turnId: event.turnId, + }); + } + return; + } + + try { + await context.bot + .getAdapter("linq") + .addReaction(context.thread.id, messageId, "thumbs_up"); + context.state.acknowledgedLinqMessageId = messageId; + const reaction = { + emoji: "thumbs_up", + messageId, + outcome: "accepted", + threadId: context.thread.id, + }; + if (log) { + log.set({ + channel: { linq: { reactions: [reaction] } }, + }); + } else { + console.info("[linq] reaction accepted", { + ...reaction, + sessionId: session.session.id, + turnId: event.turnId, + }); } + } catch (error) { + const failure = parseError(error); + const reaction = { + emoji: "thumbs_up", + error: failure, + messageId, + outcome: "failed", + threadId: context.thread.id, + }; + log?.warn("Linq reaction failed", { + channel: { + linq: { + reactions: [reaction], + }, + }, + }); + console.warn("[linq] reaction failed", { + ...reaction, + sessionId: session.session.id, + turnId: event.turnId, + }); } return; } diff --git a/agent/hooks/evlog.ts b/agent/hooks/evlog.ts new file mode 100644 index 00000000..7ed40a50 --- /dev/null +++ b/agent/hooks/evlog.ts @@ -0,0 +1,11 @@ +import { defineEvlogHook } from "evlog/eve"; + +export default defineEvlogHook({ + init: { + env: { service: "open-instinct" }, + redact: false, + }, + message: "full", + redact: false, + sessionEvent: true, +}); diff --git a/agent/instrumentation.ts b/agent/instrumentation.ts new file mode 100644 index 00000000..e2b6bb88 --- /dev/null +++ b/agent/instrumentation.ts @@ -0,0 +1,7 @@ +import { defineEvlogInstrumentation } from "evlog/eve"; + +export default defineEvlogInstrumentation({ + recordInputs: true, + recordOutputs: true, + traceChannelRequests: true, +}); diff --git a/package.json b/package.json index ae305dc6..e9318031 100644 --- a/package.json +++ b/package.json @@ -28,6 +28,7 @@ "credit-card-type": "^10.3.0", "drizzle-orm": "^0.45.2", "eve": "^0.46.1", + "evlog": "2.27.1", "lucide-react": "1.34.0", "motion": "13.1.1", "nanoid": "6.0.1", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 8164917c..683f6144 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -86,6 +86,9 @@ importers: eve: specifier: ^0.46.1 version: 0.46.1(@electric-sql/pglite@0.5.8)(@opentelemetry/api@1.9.1)(@vercel/blob@2.8.0)(@vercel/functions@3.9.5(ws@8.21.3))(ai@7.0.83(zod@4.4.3))(chokidar@4.0.0)(dotenv@17.4.2)(drizzle-orm@0.45.2(@electric-sql/pglite@0.5.8)(@neondatabase/serverless@1.1.0)(@opentelemetry/api@1.9.1)(@types/pg@8.23.1)(kysely@0.29.5)(pg@8.23.0))(giget@3.3.1)(jiti@2.7.0)(lru-cache@11.5.2)(vite@8.2.2(@types/node@24.13.3)(esbuild@0.27.0)(jiti@2.7.0)(tsx@4.21.0)(yaml@2.9.0)) + evlog: + specifier: 2.27.1 + version: 2.27.1(ai@7.0.83(zod@4.4.3))(eve@0.46.1(@electric-sql/pglite@0.5.8)(@opentelemetry/api@1.9.1)(@vercel/blob@2.8.0)(@vercel/functions@3.9.5(ws@8.21.3))(ai@7.0.83(zod@4.4.3))(chokidar@4.0.0)(dotenv@17.4.2)(drizzle-orm@0.45.2(@electric-sql/pglite@0.5.8)(@neondatabase/serverless@1.1.0)(@opentelemetry/api@1.9.1)(@types/pg@8.23.1)(kysely@0.29.5)(pg@8.23.0))(giget@3.3.1)(jiti@2.7.0)(lru-cache@11.5.2)(vite@8.2.2(@types/node@24.13.3)(esbuild@0.27.0)(jiti@2.7.0)(tsx@4.21.0)(yaml@2.9.0)))(express@5.2.1)(hono@4.13.5)(next@16.3.3(@babel/core@7.29.7)(@opentelemetry/api@1.9.1)(@types/node@24.13.3)(react-dom@19.2.8(react@19.2.8))(react@19.2.8))(ofetch@2.0.0-alpha.3)(react@19.2.8)(vite@8.2.2(@types/node@24.13.3)(esbuild@0.27.0)(jiti@2.7.0)(tsx@4.21.0)(yaml@2.9.0)) lucide-react: specifier: 1.34.0 version: 1.34.0(react@19.2.8) @@ -4345,6 +4348,66 @@ packages: resolution: {integrity: sha512-CRT1WTyuQoD771GW56XEZFQ/ZoSfWid1alKGDYMmkt2yl8UXrVR4pspqWNEcqKvVIzg6PAltWjxcSSPrboA4iA==} engines: {node: '>=18.0.0'} + evlog@2.27.1: + resolution: {integrity: sha512-Wye/LVkhvzD2sAP/nekaw9c3/d+F2SMQe3m57hgmfLXPkX37rRtDv4ncZf1QLNfo8VBvKUCgZOChJubFRMgXTw==} + engines: {node: '>=18.0.0'} + peerDependencies: + '@nestjs/common': '>=11.1.28' + '@nuxt/kit': ^4.4.2 + '@orpc/server': '>=1.14.8' + '@tanstack/start-client-core': ^1.170.14 + ai: '>=6.0.168 <8.0.0' + elysia: '>=1.4.29' + eve: '>=0.30.0' + express: '>=5.2.1' + fastify: '>=5.10.0' + h3: ^1.15.11 + hono: '>=4.12.30' + next: '>=16.2.10' + nitro: ^3.0.260311-beta + nitropack: ^2.13.4 + ofetch: ^1.5.1 + react: '>=19.2.7' + react-router: '>=7.18.1' + vite: ^7.0.0 || ^8.0.0 + peerDependenciesMeta: + '@nestjs/common': + optional: true + '@nuxt/kit': + optional: true + '@orpc/server': + optional: true + '@tanstack/start-client-core': + optional: true + ai: + optional: true + elysia: + optional: true + eve: + optional: true + express: + optional: true + fastify: + optional: true + h3: + optional: true + hono: + optional: true + next: + optional: true + nitro: + optional: true + nitropack: + optional: true + ofetch: + optional: true + react: + optional: true + react-router: + optional: true + vite: + optional: true + execa@3.2.0: resolution: {integrity: sha512-kJJfVbI/lZE1PZYDI5VPxp8zXPO9rtxOkhpZ0jMKha56AI9y2gGVC6bkukStQf0ka5Rh15BA5m7cCCH4jmHqkw==} engines: {node: ^8.12.0 || >=9.7.0} @@ -10546,6 +10609,17 @@ snapshots: dependencies: eventsource-parser: 3.1.1 + evlog@2.27.1(ai@7.0.83(zod@4.4.3))(eve@0.46.1(@electric-sql/pglite@0.5.8)(@opentelemetry/api@1.9.1)(@vercel/blob@2.8.0)(@vercel/functions@3.9.5(ws@8.21.3))(ai@7.0.83(zod@4.4.3))(chokidar@4.0.0)(dotenv@17.4.2)(drizzle-orm@0.45.2(@electric-sql/pglite@0.5.8)(@neondatabase/serverless@1.1.0)(@opentelemetry/api@1.9.1)(@types/pg@8.23.1)(kysely@0.29.5)(pg@8.23.0))(giget@3.3.1)(jiti@2.7.0)(lru-cache@11.5.2)(vite@8.2.2(@types/node@24.13.3)(esbuild@0.27.0)(jiti@2.7.0)(tsx@4.21.0)(yaml@2.9.0)))(express@5.2.1)(hono@4.13.5)(next@16.3.3(@babel/core@7.29.7)(@opentelemetry/api@1.9.1)(@types/node@24.13.3)(react-dom@19.2.8(react@19.2.8))(react@19.2.8))(ofetch@2.0.0-alpha.3)(react@19.2.8)(vite@8.2.2(@types/node@24.13.3)(esbuild@0.27.0)(jiti@2.7.0)(tsx@4.21.0)(yaml@2.9.0)): + optionalDependencies: + ai: 7.0.83(zod@4.4.3) + eve: 0.46.1(@electric-sql/pglite@0.5.8)(@opentelemetry/api@1.9.1)(@vercel/blob@2.8.0)(@vercel/functions@3.9.5(ws@8.21.3))(ai@7.0.83(zod@4.4.3))(chokidar@4.0.0)(dotenv@17.4.2)(drizzle-orm@0.45.2(@electric-sql/pglite@0.5.8)(@neondatabase/serverless@1.1.0)(@opentelemetry/api@1.9.1)(@types/pg@8.23.1)(kysely@0.29.5)(pg@8.23.0))(giget@3.3.1)(jiti@2.7.0)(lru-cache@11.5.2)(vite@8.2.2(@types/node@24.13.3)(esbuild@0.27.0)(jiti@2.7.0)(tsx@4.21.0)(yaml@2.9.0)) + express: 5.2.1 + hono: 4.13.5 + next: 16.3.3(@babel/core@7.29.7)(@opentelemetry/api@1.9.1)(@types/node@24.13.3)(react-dom@19.2.8(react@19.2.8))(react@19.2.8) + ofetch: 2.0.0-alpha.3 + react: 19.2.8 + vite: 8.2.2(@types/node@24.13.3)(esbuild@0.27.0)(jiti@2.7.0)(tsx@4.21.0)(yaml@2.9.0) + execa@3.2.0: dependencies: cross-spawn: 7.0.6 diff --git a/tests/agent/channels/linq-message-delivery.test.ts b/tests/agent/channels/linq-message-delivery.test.ts index 1be3ac93..a39b9ed3 100644 --- a/tests/agent/channels/linq-message-delivery.test.ts +++ b/tests/agent/channels/linq-message-delivery.test.ts @@ -1,6 +1,7 @@ /* oxlint-disable typescript/no-unsafe-type-assertion -- The handler fixture supplies only the Chat SDK fields exercised here. */ import type { HookContext } from "eve/hooks"; -import { describe, expect, it, vi } from "vitest"; +import type { AuditableLogger } from "evlog"; +import { beforeEach, describe, expect, it, vi } from "vitest"; import type * as Blob from "@vercel/blob"; import type { AccessScope } from "@/lib/access-scope"; import workerCancellationHook from "@/agent/hooks/worker-cancellation-delivery"; @@ -26,6 +27,15 @@ const linqChannelCapture = vi.hoisted(() => ({ ) => Promise >(), })); +const evlogCapture = vi.hoisted(() => ({ + info: vi.fn(), + set: vi.fn(), + useLogger: vi.fn<() => Pick>(), + warn: vi.fn(), +})); +vi.mock("evlog/eve", () => ({ + useLogger: evlogCapture.useLogger, +})); vi.mock("@/db/services/browser-images", () => ({ async readReadyBrowserImageArtifact( scope: AccessScope, @@ -88,6 +98,14 @@ interface LinqTestState { } describe("Linq message delivery", () => { + beforeEach(() => { + evlogCapture.info.mockClear(); + evlogCapture.set.mockClear(); + evlogCapture.useLogger.mockReset(); + evlogCapture.useLogger.mockReturnValue(evlogCapture); + evlogCapture.warn.mockClear(); + }); + it("posts final responses as native iMessage Markdown", async () => { const message = [ "Still blocked. No order was submitted.", @@ -238,9 +256,186 @@ describe("Linq message delivery", () => { "message-1", "thumbs_up" ); + expect(evlogCapture.set).toHaveBeenCalledExactlyOnceWith({ + channel: { + linq: { + reactions: [ + { + emoji: "thumbs_up", + messageId: "message-1", + outcome: "accepted", + threadId: "linq:dm:chat-1", + }, + ], + }, + }, + }); expect(state.pendingToolCallMessage).toBe("Checking the checkout"); }); + it("records Linq reaction failures without failing the turn", async () => { + const { addReaction, context, post } = handlerContext(); + const consoleWarn = vi.spyOn(console, "warn").mockReturnValue(undefined); + const error = Object.assign(new Error("Reaction denied"), { + code: "LINQ_REACTION_DENIED", + status: 403, + traceId: "trace-1", + }); + addReaction.mockRejectedValueOnce(error); + + await deliverCompletedMessage( + completedEvent({ finishReason: "tool-calls", message: "Checking" }), + context, + sessionContext() + ); + + expect(post).not.toHaveBeenCalled(); + expect(evlogCapture.warn).toHaveBeenCalledExactlyOnceWith( + "Linq reaction failed", + { + channel: { + linq: { + reactions: [ + { + emoji: "thumbs_up", + error: { + code: "LINQ_REACTION_DENIED", + message: "Reaction denied", + raw: error, + status: 403, + }, + messageId: "message-1", + outcome: "failed", + threadId: "linq:dm:chat-1", + }, + ], + }, + }, + } + ); + expect(consoleWarn).toHaveBeenCalledExactlyOnceWith( + "[linq] reaction failed", + { + emoji: "thumbs_up", + error: { + code: "LINQ_REACTION_DENIED", + message: "Reaction denied", + raw: error, + status: 403, + }, + messageId: "message-1", + outcome: "failed", + sessionId: "session-1", + threadId: "linq:dm:chat-1", + turnId: "turn-1", + } + ); + consoleWarn.mockRestore(); + }); + + it("falls back for accepted and skipped reactions when evlog is unavailable", async () => { + const { addReaction, context } = handlerContext(); + const error = new Error("No logger for this resumed turn"); + const consoleWarn = vi.spyOn(console, "warn").mockReturnValue(undefined); + const consoleInfo = vi.spyOn(console, "info").mockReturnValue(undefined); + evlogCapture.useLogger.mockImplementation(() => { + throw error; + }); + + await deliverCompletedMessage( + completedEvent({ finishReason: "tool-calls", message: "Checking" }), + context, + sessionContext() + ); + await deliverCompletedMessage( + completedEvent({ finishReason: "tool-calls", message: "Still checking" }), + context, + sessionContext() + ); + + expect(addReaction).toHaveBeenCalledExactlyOnceWith( + "linq:dm:chat-1", + "message-1", + "thumbs_up" + ); + expect(consoleWarn).toHaveBeenCalledTimes(2); + for (const call of consoleWarn.mock.calls) { + expect(call).toEqual([ + "[linq] evlog unavailable", + { + error: { + message: "No logger for this resumed turn", + raw: error, + status: 500, + }, + sessionId: "session-1", + turnId: "turn-1", + }, + ]); + } + expect(consoleInfo).toHaveBeenNthCalledWith(1, "[linq] reaction accepted", { + emoji: "thumbs_up", + messageId: "message-1", + outcome: "accepted", + sessionId: "session-1", + threadId: "linq:dm:chat-1", + turnId: "turn-1", + }); + expect(consoleInfo).toHaveBeenNthCalledWith(2, "[linq] reaction skipped", { + messageId: "message-1", + outcome: "already-acknowledged", + sessionId: "session-1", + threadId: "linq:dm:chat-1", + turnId: "turn-1", + }); + consoleWarn.mockRestore(); + consoleInfo.mockRestore(); + }); + + it("appends repeated reaction outcomes within one turn", async () => { + const { context } = handlerContext(); + const session = sessionContext(); + + await deliverCompletedMessage( + completedEvent({ finishReason: "tool-calls", message: "Checking" }), + context, + session + ); + await deliverCompletedMessage( + completedEvent({ finishReason: "tool-calls", message: "Still checking" }), + context, + session + ); + + expect(evlogCapture.set).toHaveBeenCalledWith({ + channel: { + linq: { + reactions: [ + { + emoji: "thumbs_up", + messageId: "message-1", + outcome: "accepted", + threadId: "linq:dm:chat-1", + }, + ], + }, + }, + }); + expect(evlogCapture.info).toHaveBeenCalledWith("Linq reaction skipped", { + channel: { + linq: { + reactions: [ + { + messageId: "message-1", + outcome: "already-acknowledged", + threadId: "linq:dm:chat-1", + }, + ], + }, + }, + }); + }); + it("does not post an empty final response", async () => { const { context, post } = handlerContext(); diff --git a/tests/agent/hooks/evlog.test.ts b/tests/agent/hooks/evlog.test.ts new file mode 100644 index 00000000..6c60bf1c --- /dev/null +++ b/tests/agent/hooks/evlog.test.ts @@ -0,0 +1,157 @@ +import type { HookContext } from "eve/hooks"; +import type { DrainContext, WideEvent } from "evlog"; +import { initLogger } from "evlog"; +import { resetEvlogEveForTests, useLogger } from "evlog/eve"; +import { beforeAll, describe, expect, it } from "vitest"; +import evlogHook from "@/agent/hooks/evlog"; + +const capturedEvents: WideEvent[] = []; + +beforeAll(() => { + resetEvlogEveForTests(); + initLogger({ + drain({ event }: DrainContext) { + capturedEvents.push(structuredClone(event)); + }, + env: { service: "open-instinct-test" }, + redact: false, + silent: true, + }); +}); + +describe("evlog hook", () => { + it("emits full messages, appends turn observations, and isolates later turns", async () => { + const first = hookContext("turn-1", 0); + const second = hookContext("turn-2", 1); + + await emit("turn.started", turnStarted("turn-1", 0), first); + await emit( + "message.received", + messageReceived( + "turn-1", + "email mason@example.com card 4111111111111111" + ), + first + ); + const log = useLogger(first); + log.set({ + channel: { + linq: { reactions: [{ outcome: "accepted" }] }, + }, + }); + log.set({ + channel: { + linq: { reactions: [{ outcome: "already-acknowledged" }] }, + }, + }); + await emit( + "message.completed", + messageCompleted("turn-1", "full response mason@example.com"), + first + ); + await emit("turn.completed", turnCompleted("turn-1"), first); + + await emit("turn.started", turnStarted("turn-2", 1), second); + await emit( + "message.received", + messageReceived("turn-2", "next turn"), + second + ); + await emit( + "message.completed", + messageCompleted("turn-2", "next response"), + second + ); + await emit("turn.completed", turnCompleted("turn-2"), second); + + expect(capturedEvents).toHaveLength(2); + expect(capturedEvents[0]).toMatchObject({ + channel: { + kind: "linq", + linq: { + reactions: [ + { outcome: "accepted" }, + { outcome: "already-acknowledged" }, + ], + }, + }, + message: { + received: "email mason@example.com card 4111111111111111", + response: "full response mason@example.com", + }, + }); + expect(capturedEvents[1]).toMatchObject({ + channel: { kind: "linq" }, + message: { received: "next turn", response: "next response" }, + }); + expect(capturedEvents[1]).not.toHaveProperty("channel.linq"); + }); +}); + +type EvlogEvents = NonNullable; + +async function emit( + name: Name, + event: Parameters>[0], + context: HookContext +) { + const handler = evlogHook.events?.[name]; + if (!handler) throw new Error(`Evlog handler ${name} is not configured.`); + await handler(event, context); +} + +function hookContext(turnId: string, sequence: number) { + return { + agent: { name: "root" }, + channel: { kind: "linq" }, + async getSandbox() { + throw new Error("Sandbox access is outside this focused test."); + }, + getSkill() { + throw new Error("Skill access is outside this focused test."); + }, + session: { + auth: { current: null, initiator: null }, + id: "session-1", + turn: { id: turnId, sequence }, + }, + } satisfies HookContext; +} + +function turnStarted(turnId: string, sequence: number) { + return { + data: { sequence, turnId }, + meta: { at: "2026-08-31T22:00:00.000Z", id: `start-${turnId}` }, + type: "turn.started", + } satisfies Parameters>[0]; +} + +function messageReceived(turnId: string, message: string) { + return { + data: { message, sequence: 1, turnId }, + meta: { at: "2026-08-31T22:00:01.000Z", id: `received-${turnId}` }, + type: "message.received", + } satisfies Parameters>[0]; +} + +function messageCompleted(turnId: string, message: string) { + return { + data: { + finishReason: "stop", + message, + sequence: 2, + stepIndex: 0, + turnId, + }, + meta: { at: "2026-08-31T22:00:02.000Z", id: `message-${turnId}` }, + type: "message.completed", + } satisfies Parameters>[0]; +} + +function turnCompleted(turnId: string) { + return { + data: { sequence: 3, turnId }, + meta: { at: "2026-08-31T22:00:03.000Z", id: `complete-${turnId}` }, + type: "turn.completed", + } satisfies Parameters>[0]; +}