From 26bab6402f6065c313508b0e66d11946348de8c3 Mon Sep 17 00:00:00 2001 From: Urjit Chakraborty <135136842+urjitc@users.noreply.github.com> Date: Mon, 28 Sep 2026 22:33:40 +0000 Subject: [PATCH 1/2] feat(observability): record recording transcription outcomes and title-save failures Co-authored-by: polylane[bot] <277585245+polylane[bot]@users.noreply.github.com> --- .../workspaces/ai/chat/chat-endpoint.ts | 14 +++- .../recording-transcription-workflow.ts | 29 +++++++- .../workspace-recording-observability.test.ts | 67 +++++++++++++++++++ .../workspace-recording-observability.ts | 33 +++++++++ 4 files changed, 141 insertions(+), 2 deletions(-) create mode 100644 src/features/workspaces/recordings/workspace-recording-observability.test.ts create mode 100644 src/features/workspaces/recordings/workspace-recording-observability.ts 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.ts b/src/features/workspaces/recordings/recording-transcription-workflow.ts index 8e20d4ef..f2dd3795 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,6 +23,7 @@ export class RecordingTranscriptionWorkflow extends WorkflowEntrypoint< step: WorkflowStep, ) { const { itemId, attempt } = event.payload; + const startedAt = Date.now(); try { const recording = await step.do("read recording", () => readWorkspaceRecordingForTranscription(itemId), @@ -46,9 +48,22 @@ 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: recording.workspaceId, + }); + return { recorded: true }; + }); + } } catch (error) { await step.do("mark transcription failed", () => failWorkspaceRecording( @@ -58,6 +73,18 @@ export class RecordingTranscriptionWorkflow extends WorkflowEntrypoint< error instanceof Error ? error.message : String(error), ), ); + 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: null, + }); + return { recorded: true }; + }); } } } 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..03d3a895 --- /dev/null +++ b/src/features/workspaces/recordings/workspace-recording-observability.ts @@ -0,0 +1,33 @@ +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 + * latency for the attempt, a failure carries the error, and both carry the item, + * attempt, 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, + }); +} From 599082a8a41a9055bcdcddc8d2df70cfd13cdcb0 Mon Sep 17 00:00:00 2001 From: Urjit Chakraborty <135136842+urjitc@users.noreply.github.com> Date: Mon, 28 Sep 2026 22:45:41 +0000 Subject: [PATCH 2/2] fix(recordings): retain workspace and elapsed time in outcomes Co-authored-by: polylane[bot] <277585245+polylane[bot]@users.noreply.github.com> --- .../recording-transcription-workflow.test.ts | 162 ++++++++++++++++++ .../recording-transcription-workflow.ts | 33 ++-- .../workspace-recording-observability.ts | 7 +- 3 files changed, 188 insertions(+), 14 deletions(-) create mode 100644 src/features/workspaces/recordings/recording-transcription-workflow.test.ts 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 f2dd3795..b9f9d1dc 100644 --- a/src/features/workspaces/recordings/recording-transcription-workflow.ts +++ b/src/features/workspaces/recordings/recording-transcription-workflow.ts @@ -23,12 +23,19 @@ export class RecordingTranscriptionWorkflow extends WorkflowEntrypoint< step: WorkflowStep, ) { const { itemId, attempt } = event.payload; - const startedAt = Date.now(); + // 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", { @@ -59,20 +66,16 @@ export class RecordingTranscriptionWorkflow extends WorkflowEntrypoint< instanceId: event.instanceId, itemId, schedule: (task) => this.ctx.waitUntil(task), - workspaceId: recording.workspaceId, + workspaceId, }); return { recorded: true }; }); } } catch (error) { - await step.do("mark transcription failed", () => - failWorkspaceRecording( - this.env, - itemId, - attempt, - error instanceof Error ? error.message : String(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, @@ -81,10 +84,18 @@ export class RecordingTranscriptionWorkflow extends WorkflowEntrypoint< instanceId: event.instanceId, itemId, schedule: (task) => this.ctx.waitUntil(task), - workspaceId: null, + workspaceId, }); return { recorded: true }; }); + await step.do("mark transcription failed", () => + failWorkspaceRecording( + this.env, + itemId, + attempt, + error instanceof Error ? error.message : String(error), + ), + ); } } } diff --git a/src/features/workspaces/recordings/workspace-recording-observability.ts b/src/features/workspaces/recordings/workspace-recording-observability.ts index 03d3a895..0b157562 100644 --- a/src/features/workspaces/recordings/workspace-recording-observability.ts +++ b/src/features/workspaces/recordings/workspace-recording-observability.ts @@ -5,9 +5,10 @@ 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 - * latency for the attempt, a failure carries the error, and both carry the item, - * attempt, and workflow id so a stuck or repeatedly failing recording is - * traceable from the event back to its row. + * 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;