diff --git a/src/features/workspaces/ai/chat/chat-endpoint.ts b/src/features/workspaces/ai/chat/chat-endpoint.ts index 762d213c..9fc57ba2 100644 --- a/src/features/workspaces/ai/chat/chat-endpoint.ts +++ b/src/features/workspaces/ai/chat/chat-endpoint.ts @@ -54,6 +54,7 @@ import { checkWorkspaceAiMessageAccess, trackWorkspaceAiMessageUsage, } from "#/integrations/autumn/workspace-ai-usage"; +import { recordOperationalFailure } from "#/integrations/observability/operational-events"; const MAX_AGENT_STEPS = 16; // Summaries need capable long-context reading but not the session's model. @@ -388,7 +389,18 @@ export async function handleAiChatTurn(input: { // Stream already closed — persistence below still lands. } await setThreadTitle({ threadId, title: normalized }); - })().catch((error) => console.error("[ai-chat] title delivery failed:", error)), + })().catch((error) => + recordOperationalFailure({ + distinctId: userId, + error, + event: "pg_ai_chat_title", + fields: { + thread_id: threadId, + workspace_id: threadContext.workspaceId, + }, + schedule: (task) => ctx.waitUntil(task), + }), + ), ); } diff --git a/src/features/workspaces/recordings/recording-transcription-workflow.test.ts b/src/features/workspaces/recordings/recording-transcription-workflow.test.ts new file mode 100644 index 00000000..31650510 --- /dev/null +++ b/src/features/workspaces/recordings/recording-transcription-workflow.test.ts @@ -0,0 +1,162 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import type { WorkflowEvent, WorkflowStep } from "cloudflare:workers"; + +const persistence = vi.hoisted(() => ({ + failWorkspaceRecording: vi.fn(), + publishWorkspaceRecordingTranscript: vi.fn(), + readWorkspaceRecordingForTranscription: vi.fn(), +})); +const observability = vi.hoisted(() => ({ + recordWorkspaceRecordingTranscriptionOutcome: vi.fn(), +})); +const timeline = vi.hoisted(() => ({ + buildWorkspaceRecordingTranscript: vi.fn(() => ({ cues: [] })), +})); + +vi.mock("#/features/workspaces/recordings/workspace-recording-persistence", () => persistence); +vi.mock("#/features/workspaces/recordings/workspace-recording-observability", () => observability); +vi.mock("#/features/workspaces/recordings/workspace-recording-timeline", () => timeline); +vi.mock("cloudflare:workers", () => ({ + WorkflowEntrypoint: class { + ctx: unknown; + env: unknown; + constructor(ctx: unknown, env: unknown) { + this.ctx = ctx; + this.env = env; + } + }, +})); + +import { + RecordingTranscriptionWorkflow, + type RecordingTranscriptionWorkflowParams, +} from "#/features/workspaces/recordings/recording-transcription-workflow"; + +const attempt = 2; +const itemId = "item_1"; +const eventTimestamp = new Date("2026-01-01T00:00:00.000Z"); + +interface RecordingStub { + durationMs: number; + mimeType: string; + objectKey: string | null; + status: string; + transcriptionAttempt: number; + workspaceId: string; +} + +function createRecording(overrides: Partial = {}): RecordingStub { + return { + durationMs: 5000, + mimeType: "audio/webm", + objectKey: "recordings/item_1.webm", + status: "processing", + transcriptionAttempt: attempt, + workspaceId: "workspace_1", + ...overrides, + }; +} + +function createStep() { + const names: string[] = []; + const step = { + async do(name: string, first: unknown, second?: unknown) { + names.push(name); + const callback = typeof first === "function" ? first : second; + return await (callback as () => Promise)(); + }, + }; + return { names, step }; +} + +function createWorkflow() { + const waitUntil = vi.fn(); + const env = { AI: { run: vi.fn() }, WORKSPACE_FILES: { get: vi.fn() } }; + const workflow = new RecordingTranscriptionWorkflow({ waitUntil } as never, env as never); + return { env, waitUntil, workflow }; +} + +function createEvent(): Readonly> { + return { + instanceId: "wf_1", + payload: { attempt, itemId }, + timestamp: eventTimestamp, + workflowName: "recording-transcription", + }; +} + +describe("RecordingTranscriptionWorkflow telemetry", () => { + beforeEach(() => { + vi.useFakeTimers(); + vi.setSystemTime(new Date("2026-01-01T00:01:40.000Z")); + persistence.failWorkspaceRecording.mockReset().mockResolvedValue(undefined); + persistence.publishWorkspaceRecordingTranscript.mockReset(); + persistence.readWorkspaceRecordingForTranscription.mockReset(); + observability.recordWorkspaceRecordingTranscriptionOutcome.mockReset(); + }); + + afterEach(() => { + vi.useRealTimers(); + }); + + it("records a failure with the workspace and the elapsed-since-event duration", async () => { + persistence.readWorkspaceRecordingForTranscription.mockResolvedValue( + createRecording({ objectKey: null }), + ); + const { step } = createStep(); + const { workflow } = createWorkflow(); + + await workflow.run(createEvent(), step as unknown as WorkflowStep); + + expect( + observability.recordWorkspaceRecordingTranscriptionOutcome, + ).toHaveBeenCalledExactlyOnceWith({ + attempt, + durationMs: 100_000, + error: expect.any(Error), + instanceId: "wf_1", + itemId, + schedule: expect.any(Function), + workspaceId: "workspace_1", + }); + }); + + it("records the failure outcome even when marking the row failed rejects", async () => { + persistence.readWorkspaceRecordingForTranscription.mockResolvedValue( + createRecording({ objectKey: null }), + ); + persistence.failWorkspaceRecording.mockRejectedValue(new Error("database unavailable")); + const { step } = createStep(); + const { workflow } = createWorkflow(); + + await expect(workflow.run(createEvent(), step as unknown as WorkflowStep)).rejects.toThrow( + "database unavailable", + ); + expect(observability.recordWorkspaceRecordingTranscriptionOutcome).toHaveBeenCalledOnce(); + expect( + observability.recordWorkspaceRecordingTranscriptionOutcome.mock.calls[0]?.[0], + ).toMatchObject({ attempt, itemId, workspaceId: "workspace_1" }); + }); + + it("records a success with the workspace and the elapsed-since-event duration", async () => { + persistence.readWorkspaceRecordingForTranscription.mockResolvedValue(createRecording()); + persistence.publishWorkspaceRecordingTranscript.mockResolvedValue("applied"); + const { env, workflow } = createWorkflow(); + env.WORKSPACE_FILES.get.mockResolvedValue({ body: "audio", etag: "etag_1" }); + env.AI.run.mockResolvedValue({ text: "hello" }); + const { step } = createStep(); + + await workflow.run(createEvent(), step as unknown as WorkflowStep); + + expect( + observability.recordWorkspaceRecordingTranscriptionOutcome, + ).toHaveBeenCalledExactlyOnceWith({ + attempt, + durationMs: 100_000, + instanceId: "wf_1", + itemId, + schedule: expect.any(Function), + workspaceId: "workspace_1", + }); + }); +}); diff --git a/src/features/workspaces/recordings/recording-transcription-workflow.ts b/src/features/workspaces/recordings/recording-transcription-workflow.ts index 8e20d4ef..b9f9d1dc 100644 --- a/src/features/workspaces/recordings/recording-transcription-workflow.ts +++ b/src/features/workspaces/recordings/recording-transcription-workflow.ts @@ -1,4 +1,5 @@ import { WorkflowEntrypoint, type WorkflowEvent, type WorkflowStep } from "cloudflare:workers"; +import { recordWorkspaceRecordingTranscriptionOutcome } from "#/features/workspaces/recordings/workspace-recording-observability"; import { failWorkspaceRecording, publishWorkspaceRecordingTranscript, @@ -22,11 +23,19 @@ export class RecordingTranscriptionWorkflow extends WorkflowEntrypoint< step: WorkflowStep, ) { const { itemId, attempt } = event.payload; + // Latency is measured from when the workflow event was created, not from + // this invocation: a resumed run replays completed steps, so an + // invocation-local start would report only the current attempt's work. + const startedAt = event.timestamp.getTime(); + // Retained once the recording is read so a later transcription or publish + // failure still names the workspace; null only when the read itself fails. + let workspaceId: string | null = null; try { const recording = await step.do("read recording", () => readWorkspaceRecordingForTranscription(itemId), ); if (recording.status !== "processing" || recording.transcriptionAttempt !== attempt) return; + workspaceId = recording.workspaceId; const transcript = await step.do( "transcribe recording", { @@ -46,10 +55,39 @@ export class RecordingTranscriptionWorkflow extends WorkflowEntrypoint< return buildWorkspaceRecordingTranscript(result, recording.durationMs); }, ); - await step.do("publish transcript", () => + const published = await step.do("publish transcript", () => publishWorkspaceRecordingTranscript(this.env, { itemId, attempt, transcript }), ); + if (published === "applied") { + await step.do("record transcription success", async () => { + recordWorkspaceRecordingTranscriptionOutcome({ + attempt, + durationMs: Date.now() - startedAt, + instanceId: event.instanceId, + itemId, + schedule: (task) => this.ctx.waitUntil(task), + workspaceId, + }); + return { recorded: true }; + }); + } } catch (error) { + // Record the failure before marking the row, so a persistence outage + // that rejects the mark cannot suppress the only structured signal for + // the transcription failure. The mark still runs and, if it rejects, + // its error propagates to the workflow retry. + await step.do("record transcription failure", async () => { + recordWorkspaceRecordingTranscriptionOutcome({ + attempt, + durationMs: Date.now() - startedAt, + error, + instanceId: event.instanceId, + itemId, + schedule: (task) => this.ctx.waitUntil(task), + workspaceId, + }); + return { recorded: true }; + }); await step.do("mark transcription failed", () => failWorkspaceRecording( this.env, diff --git a/src/features/workspaces/recordings/workspace-recording-observability.test.ts b/src/features/workspaces/recordings/workspace-recording-observability.test.ts new file mode 100644 index 00000000..7cd3cf13 --- /dev/null +++ b/src/features/workspaces/recordings/workspace-recording-observability.test.ts @@ -0,0 +1,67 @@ +import { beforeEach, describe, expect, it, vi } from "vitest"; + +vi.mock("#/integrations/observability/operational-events", () => ({ + recordOperationalOutcome: vi.fn(), +})); + +import { recordOperationalOutcome } from "#/integrations/observability/operational-events"; +import { recordWorkspaceRecordingTranscriptionOutcome } from "#/features/workspaces/recordings/workspace-recording-observability"; + +const mocked = vi.mocked(recordOperationalOutcome); + +describe("recordWorkspaceRecordingTranscriptionOutcome", () => { + beforeEach(() => { + mocked.mockClear(); + }); + + it("records a successful run with the item, attempt, workflow and latency context", () => { + recordWorkspaceRecordingTranscriptionOutcome({ + attempt: 2, + durationMs: 1234, + instanceId: "wf_1", + itemId: "item_1", + schedule: () => {}, + workspaceId: "workspace_1", + }); + + expect(mocked).toHaveBeenCalledExactlyOnceWith({ + error: undefined, + event: "recording_transcription", + fields: { + attempt: 2, + duration_ms: 1234, + item_id: "item_1", + workflow_id: "wf_1", + workspace_id: "workspace_1", + }, + schedule: expect.any(Function), + }); + }); + + it("records a failed run with the error object so the failure is traceable", () => { + const error = new Error("transcription failed"); + + recordWorkspaceRecordingTranscriptionOutcome({ + attempt: 1, + durationMs: 50, + error, + instanceId: "wf_2", + itemId: "item_2", + schedule: () => {}, + workspaceId: null, + }); + + expect(mocked).toHaveBeenCalledExactlyOnceWith({ + error, + event: "recording_transcription", + fields: { + attempt: 1, + duration_ms: 50, + item_id: "item_2", + workflow_id: "wf_2", + workspace_id: null, + }, + schedule: expect.any(Function), + }); + }); +}); diff --git a/src/features/workspaces/recordings/workspace-recording-observability.ts b/src/features/workspaces/recordings/workspace-recording-observability.ts new file mode 100644 index 00000000..0b157562 --- /dev/null +++ b/src/features/workspaces/recordings/workspace-recording-observability.ts @@ -0,0 +1,34 @@ +import { recordOperationalOutcome } from "#/integrations/observability/operational-events"; +import type { PostHogTelemetryScheduler } from "#/integrations/posthog/scheduler"; + +/** + * Records the settled outcome of one transcription run. The workflow marks a + * failed recording in the database but settles the run without surfacing it, + * so this is the only signal an operator has: a success carries the run's + * end-to-end latency since the workflow event was created, a failure carries + * the error, and both carry the item, attempt, workspace, and workflow id so a + * stuck or repeatedly failing recording is traceable from the event back to its + * row. + */ +export function recordWorkspaceRecordingTranscriptionOutcome(input: { + attempt: number; + durationMs: number; + error?: unknown; + instanceId: string; + itemId: string; + schedule: PostHogTelemetryScheduler; + workspaceId: string | null; +}) { + recordOperationalOutcome({ + error: input.error, + event: "recording_transcription", + fields: { + attempt: input.attempt, + duration_ms: input.durationMs, + item_id: input.itemId, + workflow_id: input.instanceId, + workspace_id: input.workspaceId, + }, + schedule: input.schedule, + }); +}