From a159fadfa18ada53329058825817cada3db8d701 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 03:31:14 -0500 Subject: [PATCH] refactor(server): centralize ordered voice reply state reads Co-Authored-By: Paperclip --- ...-10-25-voice-reply-delivery-sleep-audit.md | 30 ++++++++++++++++++ .../src/services/voice/voice-reply-policy.ts | 2 ++ server/src/services/voice/voice-reply-push.ts | 31 ++++++------------- .../src/services/voice/voice-reply-state.ts | 30 ++++++++++++++++++ 4 files changed, 72 insertions(+), 21 deletions(-) create mode 100644 server/src/services/voice/voice-reply-state.ts diff --git a/doc/plans/2026-10-25-voice-reply-delivery-sleep-audit.md b/doc/plans/2026-10-25-voice-reply-delivery-sleep-audit.md index ae97de7255e..fc7e0a739f7 100644 --- a/doc/plans/2026-10-25-voice-reply-delivery-sleep-audit.md +++ b/doc/plans/2026-10-25-voice-reply-delivery-sleep-audit.md @@ -99,3 +99,33 @@ Next implement the shared voice reply-state reader and deadline selection, audit all session/reply/action and published-state producers, register the worker, and prove quiet/reactivation before retiring the recurring and direct normal dispatch paths. Preserve the public/manual push API and all adjacent voice maintenance. + +The ordered reply-state reader is now also shared inside the voice owner. It +preserves the caller's session lock, published-only first reply, question barrier, +prior action locks and multipart policy. It exposes a saved dispatch-expiry +boundary without replaying a dispatch. The same 12 delivery cases and server +typechecking pass after this second extraction. + +Implementation direction for the next step: + +- Scoped reconstruction pages identities of active unexpired phone sessions and + outstanding dispatching actions. A changed session may receive one immediate + inspection; this is a startup/committed-write wake, not a recurring deadline. +- Check each session independently: authorize and collect missed replies under + the existing session transaction, then use readVoiceReplyState. No reply or a + permanent barrier disarms it. A retry or in-flight intent keeps its real deadline. + Provider dispatch remains outside the transaction, using the existing claim. +- Keep action-expiry entries independent of session eligibility so terminal calls + do not strand old dispatch intents. Use the existing strict 30-second threshold + and atomic dispatching-to-unknown update. Never resend an unknown action. +- A session check that fails must retain its own backoff and not block successful + checks for other companies. Do not move all authorization/catch-up into one + full-inventory load that fails as a unit. +- Add a session scope to manual push execution so the worker cannot rediscover all + calls per selected item. Retain the existing manual API for explicit callers. +- Route published-state notification in both publications/worker.ts and + publications/resolution.ts through a narrow owner helper. Their endpoint identity + is already available. Avoid adding voice policy to the shared coordinator. +- Notify before reply creation, presentation and action mutation. Wire session and + configuration changes without giving reply delivery its own active-call polling + deadline. Reconstruct on startup/resume; explicit feature changes invalidate it. diff --git a/server/src/services/voice/voice-reply-policy.ts b/server/src/services/voice/voice-reply-policy.ts index aee7c8b5b4a..42d0a0358f5 100644 --- a/server/src/services/voice/voice-reply-policy.ts +++ b/server/src/services/voice/voice-reply-policy.ts @@ -1,5 +1,7 @@ import { createHash } from "node:crypto"; +export const VOICE_REPLY_ACTION_KIND = "speko_voice_reply_push"; + /** Plain voice text stays inside the provider's UTF-16 limit, without splitting * surrogate pairs. Whitespace-only chunks need no audible delivery. */ export function splitSpekoReplyText(text: string): string[] { diff --git a/server/src/services/voice/voice-reply-push.ts b/server/src/services/voice/voice-reply-push.ts index 5eca00dc43a..72e7113af45 100644 --- a/server/src/services/voice/voice-reply-push.ts +++ b/server/src/services/voice/voice-reply-push.ts @@ -1,8 +1,9 @@ import { notifyVoiceRecovery } from "./voice-recovery-notifications.js"; -import { planVoiceReply } from "./voice-reply-policy.js"; +import { VOICE_REPLY_ACTION_KIND } from "./voice-reply-policy.js"; +import { readVoiceReplyState } from "./voice-reply-state.js"; export { splitSpekoReplyText } from "./voice-reply-policy.js"; -import { and, asc, eq, gt, inArray, isNull, lt, sql } from "drizzle-orm"; -import { chatActions, chatEndpoints, chatPublications, chatVoiceReplies, chatVoiceSessions, type Db } from "@paperclipai/db"; +import { and, asc, eq, gt, inArray, isNull, lt } from "drizzle-orm"; +import { chatActions, chatEndpoints, chatVoiceReplies, chatVoiceSessions, type Db } from "@paperclipai/db"; import { instanceSettingsService } from "../instance-settings.js"; import { HttpError } from "../../errors.js"; import { createSpekoProvider, SpekoProviderError } from "./speko-provider.js"; @@ -10,7 +11,7 @@ import type { voiceSessionStore } from "./voice-session-store.js"; import { logger } from "../../middleware/logger.js"; -const kind = "speko_voice_reply_push"; +const kind = VOICE_REPLY_ACTION_KIND; /** Existing action ledger is the push outbox. A committed dispatch intent is * never blindly resent after a crash or an ambiguous provider outcome. */ export async function pushVoiceReplies(db: Db, options: { @@ -40,28 +41,16 @@ export async function pushVoiceReplies(db: Db, options: { const {session} = context; if (!session.providerSessionId) return null; await options.store.collectMissedReplies(tx, session); - const [inFlight] = await tx.select({id: chatActions.id}).from(chatActions).where(and(eq(chatActions.companyId, session.companyId), eq(chatActions.endpointId, session.endpointId), eq(chatActions.kind, kind), eq(chatActions.status, "dispatching"), sql`${chatActions.payload}->>'sessionId' = ${session.id}`)).limit(1); - if (inFlight) return null; - const [row] = await tx.select({reply: chatVoiceReplies, payload: chatPublications.payload}).from(chatVoiceReplies) - .innerJoin(chatPublications, and(eq(chatPublications.id, chatVoiceReplies.publicationId), eq(chatPublications.companyId, session.companyId))) - .where(and(eq(chatVoiceReplies.companyId, session.companyId), eq(chatVoiceReplies.sessionId, session.id), isNull(chatVoiceReplies.deliveredAt), - eq(chatPublications.endpointId, session.endpointId), eq(chatPublications.conversationId, session.conversationId), eq(chatPublications.issueId, session.issueId), eq(chatPublications.state, "published"))) - .orderBy(asc(chatVoiceReplies.cursor)).limit(1); - // Questions need a hint that makes the phone agent retrieve the - // authorized question IDs. Keep this reply pending until get_updates - // presents it; never speak later answers over an unpresented question. - const questionNotification = typeof row?.payload.interactionId === "string"; - if (!row || typeof row.payload.text !== "string" || !row.payload.text.trim()) return null; - const saved = await tx.select().from(chatActions).where(and(eq(chatActions.companyId, session.companyId), eq(chatActions.endpointId, session.endpointId), eq(chatActions.kind, kind), sql`${chatActions.payload}->>'replyId' = ${row.reply.id}`)).for("update"); - const plan = planVoiceReply({ replyId: row.reply.id, text: row.payload.text, questionNotification, actions: saved, now: Date.now() }); - if (plan.kind !== "dispatch") return null; + const state = await readVoiceReplyState(tx, session); + if (state.kind !== "reply" || state.plan.kind !== "dispatch") return null; + const { reply, questionNotification, plan } = state; const values = { status: "dispatching", result: null, updatedAt: new Date() }; const action = plan.priorActionId ? (await tx.update(chatActions).set(values).where(eq(chatActions.id, plan.priorActionId)).returning())[0]! : (await tx.insert(chatActions).values({ companyId: session.companyId, endpointId: session.endpointId, conversationId: session.conversationId, kind, providerActionId: plan.identity, - payload: { sessionId: session.id, replyId: row.reply.id, publicationId: row.reply.publicationId, generation: session.generation, providerSessionId: session.providerSessionId, + payload: { sessionId: session.id, replyId: reply.id, publicationId: reply.publicationId, generation: session.generation, providerSessionId: session.providerSessionId, questionNotification, partIndex: plan.partIndex, partCount: plan.partCount, textDigest: plan.textDigest }, ...values }).returning())[0]!; - return { ...context, reply: row.reply, questionNotification, text: plan.text, partIndex: plan.partIndex, partCount: plan.partCount, action }; + return { ...context, reply, questionNotification, text: plan.text, partIndex: plan.partIndex, partCount: plan.partCount, action }; }).catch(error => { if (error instanceof HttpError && [403,404].includes(error.status)) { logger.warn({event: "voice.reply.push.authorization_blocked", companyId: candidate.companyId, diff --git a/server/src/services/voice/voice-reply-state.ts b/server/src/services/voice/voice-reply-state.ts new file mode 100644 index 00000000000..35c259a54b8 --- /dev/null +++ b/server/src/services/voice/voice-reply-state.ts @@ -0,0 +1,30 @@ +import { and, asc, eq, isNull, sql } from "drizzle-orm"; +import { chatActions, chatPublications, chatVoiceReplies } from "@paperclipai/db"; +import type { VoiceSessionRow, VoiceTransaction } from "./voice-session-store.js"; +import { planVoiceReply, VOICE_REPLY_ACTION_KIND } from "./voice-reply-policy.js"; + +/** Read the ordered delivery barrier under the caller's existing session lock. */ +export async function readVoiceReplyState( + tx: VoiceTransaction, + session: Pick, +) { + const [inFlight] = await tx.select({ updatedAt: chatActions.updatedAt }).from(chatActions) + .where(and(eq(chatActions.companyId, session.companyId), eq(chatActions.endpointId, session.endpointId), + eq(chatActions.kind, VOICE_REPLY_ACTION_KIND), eq(chatActions.status, "dispatching"), + sql`${chatActions.payload}->>'sessionId' = ${session.id}`)) + .orderBy(asc(chatActions.updatedAt)).limit(1); + if (inFlight) return { kind: "in_flight" as const, retryAt: inFlight.updatedAt.getTime() + 30_001 }; + const [row] = await tx.select({ reply: chatVoiceReplies, payload: chatPublications.payload }).from(chatVoiceReplies) + .innerJoin(chatPublications, and(eq(chatPublications.id, chatVoiceReplies.publicationId), eq(chatPublications.companyId, session.companyId))) + .where(and(eq(chatVoiceReplies.companyId, session.companyId), eq(chatVoiceReplies.sessionId, session.id), isNull(chatVoiceReplies.deliveredAt), + eq(chatPublications.endpointId, session.endpointId), eq(chatPublications.conversationId, session.conversationId), + eq(chatPublications.issueId, session.issueId), eq(chatPublications.state, "published"))) + .orderBy(asc(chatVoiceReplies.cursor)).limit(1); + if (!row || typeof row.payload.text !== "string" || !row.payload.text.trim()) return { kind: "blocked" as const }; + const questionNotification = typeof row.payload.interactionId === "string"; + const saved = await tx.select().from(chatActions).where(and(eq(chatActions.companyId, session.companyId), + eq(chatActions.endpointId, session.endpointId), eq(chatActions.kind, VOICE_REPLY_ACTION_KIND), + sql`${chatActions.payload}->>'replyId' = ${row.reply.id}`)).for("update"); + return { kind: "reply" as const, reply: row.reply, questionNotification, + plan: planVoiceReply({ replyId: row.reply.id, text: row.payload.text, questionNotification, actions: saved, now: Date.now() }) }; +}