diff --git a/apps/agent/src/app/bot/bot-handler.ts b/apps/agent/src/app/bot/bot-handler.ts index c1850b8..6f38d21 100644 --- a/apps/agent/src/app/bot/bot-handler.ts +++ b/apps/agent/src/app/bot/bot-handler.ts @@ -12,6 +12,7 @@ import { logger } from '../../infrastructure/logger'; import { agent } from '../agent'; import { resolveTimeZone } from '../agent/runtime-context'; import { AttachmentService } from '../attachments'; +import { trackAgentResponse } from './feedback'; import { normalizeIMessagePost } from './imessage'; import { extractResponseText, formatAskUserQuestion, readAskUserSuspension } from './response'; import { chatState } from './transport'; @@ -75,7 +76,7 @@ export class BotHandler { const suspension = readAskUserSuspension(result); if (suspension) { - await this.#postQuestion(thread, pendingKey, message, suspension); + await this.#postQuestion(thread, pendingKey, message, suspension, result); return; } @@ -83,7 +84,14 @@ export class BotHandler { await chatState.delete(pendingKey); } - await thread.post(normalizeIMessagePost(extractResponseText(result))); + const sent = await thread.post(normalizeIMessagePost(extractResponseText(result))); + await trackAgentResponse(sent.id, { + resourceId, + threadId: thread.id, + traceId: result.traceId, + spanId: result.spanId, + runId: result.runId, + }); logger.info('Inbound message handling completed', { messageId: message.id, @@ -107,6 +115,7 @@ export class BotHandler { pendingKey: string, message: Message, suspension: AskUserSuspension, + result: AgentResult, ) { const pending = { ...suspension, @@ -119,7 +128,14 @@ export class BotHandler { await chatState.set(pendingKey, pending, PENDING_QUESTION_TTL_MS); try { - await thread.post(normalizeIMessagePost(formatAskUserQuestion(suspension))); + const sent = await thread.post(normalizeIMessagePost(formatAskUserQuestion(suspension))); + await trackAgentResponse(sent.id, { + resourceId: message.author.userId, + threadId: thread.id, + traceId: result.traceId, + spanId: result.spanId, + runId: result.runId, + }); } catch (error) { await chatState.delete(pendingKey); throw error; diff --git a/apps/agent/src/app/bot/delivery.ts b/apps/agent/src/app/bot/delivery.ts index 24353ad..24e6fbf 100644 --- a/apps/agent/src/app/bot/delivery.ts +++ b/apps/agent/src/app/bot/delivery.ts @@ -1,10 +1,23 @@ +import type { AgentTraceContext } from './feedback'; import { logger } from '../../infrastructure/logger'; +import { trackAgentResponse } from './feedback'; import { normalizeIMessagePost } from './imessage'; -import { chat, initializeBot } from './transport'; +import { chat } from './transport'; /** Post through the singleton Chat SDK transport boundary. */ -export async function postToThread(threadId: string, text: string) { - await initializeBot(); - await chat.thread(threadId).post(normalizeIMessagePost(text)); +export async function postToThread( + threadId: string, + text: string, + traceContext?: AgentTraceContext, +) { + // Webhooks initialize Chat lazily. Scheduled delivery is outside a webhook, + // so use Chat's native idempotent lifecycle method before posting. + await chat.initialize(); + const sent = await chat.thread(threadId).post(normalizeIMessagePost(text)); + + if (traceContext) { + await trackAgentResponse(sent.id, traceContext); + } + logger.info('Outbound iMessage posted', { threadId }); } diff --git a/apps/agent/src/app/bot/feedback.test.ts b/apps/agent/src/app/bot/feedback.test.ts new file mode 100644 index 0000000..168d4d2 --- /dev/null +++ b/apps/agent/src/app/bot/feedback.test.ts @@ -0,0 +1,243 @@ +import type { FeedbackInput } from '@mastra/core/observability'; +import type { Lock, ReactionEvent } from 'chat'; + +import { describe, expect, it, vi } from 'vitest'; + +import type { FeedbackObservability, FeedbackState } from './feedback'; +import { + createFeedback, + handleReactionFeedback, + responseKey, + trackAgentResponse, +} from './feedback'; + +vi.hoisted(() => { + process.env.DATABASE_URL ??= 'postgres://localhost/agent-test'; +}); + +describe('reaction feedback bridge', () => { + it('persists the agent trace against the posted platform message', async () => { + const state = createState(); + + await trackAgentResponse( + 'imsg-1', + { + resourceId: 'resource-1', + threadId: 'thread-1', + traceId: 'trace-1', + spanId: 'span-1', + runId: 'run-1', + }, + state, + ); + + expect(state.values.get(responseKey('imsg-1'))).toMatchObject({ + resourceId: 'resource-1', + threadId: 'thread-1', + traceId: 'trace-1', + spanId: 'span-1', + runId: 'run-1', + }); + }); + + it('records thumbs feedback and ignores duplicate additions', async () => { + const state = createState(); + const observability = createObservability(); + const dependencies = { state, observability }; + + await trackAgentResponse( + 'imsg-2', + { resourceId: 'resource-2', threadId: 'thread-2', traceId: 'trace-2' }, + state, + ); + + const event = reactionEvent({ + added: true, + messageId: 'imsg-2', + rawEmoji: 'like', + threadId: 'thread-2', + userId: '+48123456789', + }); + + await handleReactionFeedback(event, dependencies); + await handleReactionFeedback(event, dependencies); + + expect(observability.feedback).toHaveLength(1); + expect(observability.feedback[0]).toMatchObject({ + traceId: 'trace-2', + feedback: { + feedbackSource: 'user', + feedbackType: 'thumbs', + value: 1, + metadata: { action: 'added', reaction: 'like' }, + }, + }); + expect((observability.feedback[0] as FeedbackCall).feedback.feedbackUserId).not.toContain( + '+48123456789', + ); + expect(observability.flushCount).toBe(1); + }); + + it('emits a removal tombstone and ignores duplicate removals', async () => { + const state = createState(); + const observability = createObservability(); + const dependencies = { state, observability }; + + await trackAgentResponse( + 'imsg-3', + { resourceId: 'resource-3', threadId: 'thread-3', traceId: 'trace-3' }, + state, + ); + + const added = reactionEvent({ + added: true, + messageId: 'imsg-3', + rawEmoji: 'dislike', + threadId: 'thread-3', + userId: 'user-3', + }); + const removed = reactionEvent({ + added: false, + messageId: 'imsg-3', + rawEmoji: 'dislike', + threadId: 'thread-3', + userId: 'user-3', + }); + + await handleReactionFeedback(added, dependencies); + await handleReactionFeedback(removed, dependencies); + await handleReactionFeedback(removed, dependencies); + + expect(observability.feedback).toHaveLength(2); + expect(observability.feedback[1]).toMatchObject({ + feedback: { + feedbackType: 'thumbs_removed', + value: 'dislike', + metadata: { action: 'removed', reaction: 'dislike' }, + }, + }); + }); + + it('does not emit a removal tombstone when the addition was never observed', async () => { + const state = createState(); + const observability = createObservability(); + + await trackAgentResponse( + 'imsg-4', + { resourceId: 'resource-4', threadId: 'thread-4', traceId: 'trace-4' }, + state, + ); + + await handleReactionFeedback( + reactionEvent({ + added: false, + messageId: 'imsg-4', + rawEmoji: 'love', + threadId: 'thread-4', + userId: 'user-4', + }), + { state, observability }, + ); + + expect(observability.feedback).toHaveLength(0); + }); + + it('maps non-thumb reactions to tapback feedback', () => { + const feedback = createFeedback( + reactionEvent({ + added: true, + messageId: 'imsg-5', + rawEmoji: 'love', + threadId: 'thread-5', + userId: 'user-5', + }), + 'love', + 'source-5', + ); + + expect(feedback).toMatchObject({ + feedbackType: 'tapback', + value: 'love', + sourceId: 'source-5', + }); + }); +}); + +type FeedbackCall = { + traceId?: string; + spanId?: string; + feedback: FeedbackInput; +}; + +function createState() { + const values = new Map(); + const locks = new Map(); + + const state: FeedbackState & { + values: Map; + } = { + values, + async acquireLock(threadId) { + if (locks.has(threadId)) { + return null; + } + + const lock = { expiresAt: Date.now() + 30_000, threadId, token: threadId }; + locks.set(threadId, lock); + return lock; + }, + async releaseLock(lock) { + locks.delete(lock.threadId); + }, + async get(key: string) { + return (values.get(key) as T | undefined) ?? null; + }, + async set(key: string, value: T) { + values.set(key, value); + }, + }; + + return state; +} + +function createObservability() { + const feedback: FeedbackCall[] = []; + + const observability: FeedbackObservability & { + feedback: FeedbackCall[]; + flushCount: number; + } = { + feedback, + flushCount: 0, + async addFeedback(call) { + feedback.push(call); + }, + async flush() { + observability.flushCount += 1; + }, + }; + + return observability; +} + +function reactionEvent({ + added, + messageId, + rawEmoji, + threadId, + userId, +}: { + added: boolean; + messageId: string; + rawEmoji: string; + threadId: string; + userId: string; +}) { + return { + added, + messageId, + rawEmoji, + threadId, + user: { userId }, + } as unknown as ReactionEvent; +} diff --git a/apps/agent/src/app/bot/feedback.ts b/apps/agent/src/app/bot/feedback.ts new file mode 100644 index 0000000..b9b4af8 --- /dev/null +++ b/apps/agent/src/app/bot/feedback.ts @@ -0,0 +1,252 @@ +import type { FeedbackInput } from '@mastra/core/observability'; +import type { Lock, ReactionEvent, StateAdapter } from 'chat'; + +import { createHash } from 'node:crypto'; + +import { logger } from '../../infrastructure/logger'; +import { agentObservability } from '../../mastra/observability'; +import { chatState } from './transport'; + +const RESPONSE_TRACE_TTL_MS = 90 * 24 * 60 * 60 * 1_000; +const REACTION_STATE_TTL_MS = RESPONSE_TRACE_TTL_MS; +const REACTION_LOCK_TTL_MS = 30 * 1_000; + +export type FeedbackState = Pick; +export type FeedbackObservability = Pick; + +export type AgentTraceContext = { + resourceId: string; + threadId: string; + traceId?: string; + spanId?: string; + runId?: string; +}; + +export type ResponseTrace = AgentTraceContext & { + recordedAt: string; +}; + +export type ReactionFeedbackDependencies = { + state: FeedbackState; + observability: FeedbackObservability; +}; + +const defaultDependencies: ReactionFeedbackDependencies = { + state: chatState, + observability: agentObservability, +}; + +/** + * Keep the trace associated with the platform message, not with a local + * process. Reactions can arrive in a later serverless invocation. + */ +export async function trackAgentResponse( + messageId: string, + context: AgentTraceContext, + state: FeedbackState = chatState, +) { + const traceId = context.traceId?.trim(); + + if (!messageId || !traceId) { + return; + } + + const response: ResponseTrace = { + ...context, + traceId, + ...(context.spanId?.trim() ? { spanId: context.spanId.trim() } : {}), + ...(context.runId?.trim() ? { runId: context.runId.trim() } : {}), + recordedAt: new Date().toISOString(), + }; + + try { + await state.set(responseKey(messageId), response, RESPONSE_TRACE_TTL_MS); + } catch (error) { + // The response has already been delivered. A state failure must not make + // the bot post a second failure response to the user. + logger.warn('Agent response feedback mapping could not be persisted', { + messageId, + threadId: context.threadId, + error: describeError(error), + }); + } +} + +/** + * Convert Chat SDK reaction events into Mastra feedback without invoking the + * agent. State transitions are serialized per response/user/reaction so + * duplicate webhook deliveries remain harmless. + */ +export async function handleReactionFeedback( + event: ReactionEvent, + dependencies: ReactionFeedbackDependencies = defaultDependencies, +) { + let lock: Lock | undefined; + try { + const response = await dependencies.state.get(responseKey(event.messageId)); + + if (!response || response.threadId !== event.threadId || !response.traceId) { + return; + } + + const reaction = normalizeReaction(event.rawEmoji); + + if (!reaction || !event.user.userId) { + return; + } + + const reactionStateKey = stateKey(event, reaction); + lock = + (await dependencies.state.acquireLock(reactionStateKey, REACTION_LOCK_TTL_MS)) ?? undefined; + + if (!lock) { + logger.debug('Reaction feedback event skipped because its state is locked', { + messageId: event.messageId, + threadId: event.threadId, + }); + return; + } + + const current = await dependencies.state.get(reactionStateKey); + + // A removal without a previously observed addition is not actionable. + // This can happen when the webhook was enabled after the reaction was + // added, or when an old event is replayed. + if (!event.added && !current?.active) { + return; + } + + if (current?.active === event.added) { + return; + } + + const sourceId = reactionSourceId(event, reaction); + const feedback = createFeedback(event, reaction, sourceId); + + await dependencies.observability.addFeedback({ + traceId: response.traceId, + ...(response.spanId ? { spanId: response.spanId } : {}), + // Supplying correlationContext makes this a direct exporter event. It + // avoids requiring the production Platform-only setup to rehydrate the + // trace from the local Mastra Postgres store. + correlationContext: { + resourceId: response.resourceId, + threadId: response.threadId, + ...(response.runId ? { runId: response.runId } : {}), + source: 'imessage', + serviceName: 'agent', + }, + feedback, + }); + + // Mastra buffers non-tracing signals. Flush explicitly because this + // handler normally runs in a short-lived serverless invocation. + await dependencies.observability.flush(); + + await dependencies.state.set( + reactionStateKey, + { + active: event.added, + sourceId, + updatedAt: new Date().toISOString(), + } satisfies StoredReactionState, + REACTION_STATE_TTL_MS, + ); + + logger.info('Reaction feedback recorded', { + added: event.added, + feedbackType: feedback.feedbackType, + messageId: event.messageId, + threadId: event.threadId, + }); + } catch (error) { + logger.error('Reaction feedback could not be recorded', { + added: event.added, + messageId: event.messageId, + threadId: event.threadId, + error: describeError(error), + }); + } finally { + if (lock) { + await releaseLock(dependencies.state, lock); + } + } +} + +export function createFeedback( + event: Pick, + reaction: string, + sourceId: string, +): FeedbackInput { + const thumbs = reaction === 'like' || reaction === 'dislike'; + const isRemoval = !event.added; + + return { + feedbackSource: 'user', + feedbackType: isRemoval + ? thumbs + ? 'thumbs_removed' + : 'tapback_removed' + : thumbs + ? 'thumbs' + : 'tapback', + value: !isRemoval && thumbs ? (reaction === 'like' ? 1 : -1) : reaction, + feedbackUserId: `imessage-user:${digest(event.user.userId)}`, + sourceId, + metadata: { + action: isRemoval ? 'removed' : 'added', + channel: 'imessage', + messageId: event.messageId, + reaction, + threadId: event.threadId, + }, + }; +} + +export function responseKey(messageId: string) { + return `feedback:response:${encodeURIComponent(messageId)}`; +} + +function stateKey(event: ReactionEvent, reaction: string) { + return `feedback:reaction:${digest( + [event.threadId, event.messageId, event.user.userId, reaction].join('\u0000'), + )}`; +} + +function reactionSourceId(event: ReactionEvent, reaction: string) { + return `imessage-reaction:${digest( + [event.threadId, event.messageId, event.user.userId, reaction].join('\u0000'), + )}`; +} + +function normalizeReaction(rawEmoji: string) { + const reaction = rawEmoji.trim().toLowerCase(); + return reaction || undefined; +} + +function digest(value: string) { + return createHash('sha256').update(value).digest('hex').slice(0, 24); +} + +async function releaseLock(state: FeedbackState, lock: Lock) { + try { + await state.releaseLock(lock); + } catch (error) { + logger.warn('Reaction feedback state lock could not be released', { + threadId: lock.threadId, + error: describeError(error), + }); + } +} + +type StoredReactionState = { + active: boolean; + sourceId: string; + updatedAt: string; +}; + +function describeError(error: unknown) { + return error instanceof Error + ? { name: error.name, message: error.message.slice(0, 500) } + : { name: 'UnknownError', message: String(error).slice(0, 500) }; +} diff --git a/apps/agent/src/app/bot/index.ts b/apps/agent/src/app/bot/index.ts index c08fbf0..38879b9 100644 --- a/apps/agent/src/app/bot/index.ts +++ b/apps/agent/src/app/bot/index.ts @@ -1,5 +1,6 @@ import { BotHandler } from './bot-handler'; -import { chat, chatState, initializeBot } from './transport'; +import { handleReactionFeedback } from './feedback'; +import { chat, chatState } from './transport'; chat.onDirectMessage((thread, message) => BotHandler.respondToMessage({ @@ -26,4 +27,6 @@ chat.onSubscribedMessage((thread, message) => }), ); -export { chat, chatState, initializeBot }; +chat.onReaction(handleReactionFeedback); + +export { chat, chatState }; diff --git a/apps/agent/src/app/bot/response.ts b/apps/agent/src/app/bot/response.ts index 8a232f2..8c5f9a2 100644 --- a/apps/agent/src/app/bot/response.ts +++ b/apps/agent/src/app/bot/response.ts @@ -8,6 +8,8 @@ export type AgentResult = { text?: string; finishReason?: string; runId?: string; + traceId?: string; + spanId?: string; suspendPayload?: unknown; tripwire?: unknown; steps?: ReadonlyArray; diff --git a/apps/agent/src/app/bot/scheduled.ts b/apps/agent/src/app/bot/scheduled.ts index e7be2db..9a28706 100644 --- a/apps/agent/src/app/bot/scheduled.ts +++ b/apps/agent/src/app/bot/scheduled.ts @@ -4,11 +4,11 @@ import { RequestContext, } from '@mastra/core/request-context'; +import type { AgentResult } from './response'; import { logger } from '../../infrastructure/logger'; import { agent } from '../agent'; import { postToThread } from './delivery'; import { extractResponseText } from './response'; -import { initializeBot } from './transport'; export async function runScheduled({ resourceId, @@ -25,8 +25,6 @@ export async function runScheduled({ source: 'one-time-schedule' | 'recurring-schedule'; deliveryId?: string; }) { - await initializeBot(); - logger.info('Scheduled agent turn started', { deliveryId, resourceId, @@ -58,9 +56,16 @@ export async function runScheduled({ requestContext, }, ); - const text = extractResponseText(result as never); + const agentResult = result as AgentResult; + const text = extractResponseText(agentResult); - await postToThread(threadId, text); + await postToThread(threadId, text, { + resourceId, + threadId, + traceId: agentResult.traceId, + spanId: agentResult.spanId, + runId: agentResult.runId, + }); logger.info('Scheduled agent turn completed', { deliveryId, diff --git a/apps/agent/src/app/bot/transport.ts b/apps/agent/src/app/bot/transport.ts index d41bf2f..54be3cd 100644 --- a/apps/agent/src/app/bot/transport.ts +++ b/apps/agent/src/app/bot/transport.ts @@ -1,8 +1,8 @@ import type { Logger as ChatLogger } from 'chat'; import { createPostgresState } from '@chat-adapter/state-pg'; -import { blooio } from '@imessage-sdk/blooio'; import { createIMessageAdapter } from '@imessage-sdk/chat-adapter'; +import { photon } from '@imessage-sdk/photon'; import { Chat } from 'chat'; import { logger } from '../../infrastructure/logger'; @@ -19,7 +19,7 @@ const SAFE_CHAT_LOG_KEYS = new Set([ ]); const imessageAdapter = createIMessageAdapter({ - provider: blooio(), + provider: photon(), }); /** @@ -45,20 +45,6 @@ export const chat = new Chat({ logger: createChatLogger('chat'), }); -let initialization: Promise | undefined; - -/** Initialize the singleton once for out-of-band delivery paths. Webhooks initialize lazily. */ -export function initializeBot() { - initialization ??= chat.initialize().catch((error) => { - // A transient cold-start failure must not poison all later requests in a - // warm function instance. - initialization = undefined; - throw error; - }); - - return initialization; -} - function createChatLogger(component: string): ChatLogger { const child = logger.child({ component }); diff --git a/apps/agent/src/mastra/index.ts b/apps/agent/src/mastra/index.ts index 5069f42..2741039 100644 --- a/apps/agent/src/mastra/index.ts +++ b/apps/agent/src/mastra/index.ts @@ -1,7 +1,6 @@ import { Mastra } from '@mastra/core/mastra'; import { SimpleAuth } from '@mastra/core/server'; import { VercelDeployer } from '@mastra/deployer-vercel'; -import { Observability, SensitiveDataFilter } from '@mastra/observability'; import { PostgresStore } from '@mastra/pg'; import { agent } from '../app/agent/index'; @@ -30,7 +29,7 @@ import { manageKnowledgeTool, readKnowledgeTool } from '../app/tools/knowledge-t import { daySummaryWorkflow } from '../app/workflows/day-summary'; import { databasePool } from '../infrastructure/database'; import { logger } from '../infrastructure/logger'; -import { createAgentObservabilityExporters } from './observability'; +import { agentObservability } from './observability'; configureScheduleDelivery({ runScheduled, postToThread }); @@ -93,17 +92,5 @@ export const mastra = new Mastra({ public: ['/api/agents/agent/channels/imessage/webhook'], }), }, - observability: new Observability({ - configs: { - default: { - serviceName: 'agent', - exporters: createAgentObservabilityExporters(), - spanOutputProcessors: [new SensitiveDataFilter()], - logging: { - enabled: true, - level: 'info', - }, - }, - }, - }), + observability: agentObservability, }); diff --git a/apps/agent/src/mastra/observability.ts b/apps/agent/src/mastra/observability.ts index 5574e55..4da0823 100644 --- a/apps/agent/src/mastra/observability.ts +++ b/apps/agent/src/mastra/observability.ts @@ -1,4 +1,9 @@ -import { MastraPlatformExporter, MastraStorageExporter } from '@mastra/observability'; +import { + MastraPlatformExporter, + MastraStorageExporter, + Observability, + SensitiveDataFilter, +} from '@mastra/observability'; interface ObservabilityEnvironment { NODE_ENV?: string; @@ -16,3 +21,17 @@ export function createAgentObservabilityExporters( return useMastraPlatform ? [new MastraPlatformExporter()] : [new MastraStorageExporter()]; } + +export const agentObservability = new Observability({ + configs: { + default: { + serviceName: 'agent', + exporters: createAgentObservabilityExporters(), + spanOutputProcessors: [new SensitiveDataFilter()], + logging: { + enabled: true, + level: 'info', + }, + }, + }, +});