From 515e2625d0a3e21b6ac1eb1df58cb7f042fd36e5 Mon Sep 17 00:00:00 2001 From: Dotta Date: Sun, 25 Oct 2026 03:27:30 -0500 Subject: [PATCH] refactor(server): share voice reply chunk and retry policy Co-Authored-By: Paperclip --- ...-10-25-voice-reply-delivery-sleep-audit.md | 15 +++++ .../src/services/voice/voice-reply-policy.ts | 59 +++++++++++++++++++ server/src/services/voice/voice-reply-push.ts | 45 ++++---------- 3 files changed, 85 insertions(+), 34 deletions(-) create mode 100644 server/src/services/voice/voice-reply-policy.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 35e032e2641..ae97de7255e 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 @@ -84,3 +84,18 @@ Receiver #17572 has passing latest-head checks and Greptile 5/5 at 1604a9fcba; its second Paperclip reviewer stopped before completion, so do not invent a fresh completed review result. No stack PR was merged. The complete original inventory, older #17490 failure and final master reconciliation remain outstanding. + + +## Initial implementation checkpoint + +The existing multipart selection now lives in voice-reply-policy.ts. The dispatcher +uses that pure policy for chunk identity, digest consistency, question barriers, +retry deadlines and no-resend outcomes. Its claim, authorization and provider I/O +remain in the existing owner. Twelve selected real-database reply integration +cases and server typechecking pass after extraction. This is a foundation only: +no recurring reply scan has been removed, and no reply PR is published yet. + +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. diff --git a/server/src/services/voice/voice-reply-policy.ts b/server/src/services/voice/voice-reply-policy.ts new file mode 100644 index 00000000000..aee7c8b5b4a --- /dev/null +++ b/server/src/services/voice/voice-reply-policy.ts @@ -0,0 +1,59 @@ +import { createHash } from "node:crypto"; + +/** 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[] { + const parts: string[] = []; + for (let start = 0; start < text.length;) { + let end = Math.min(start + 16_000, text.length); + if (end < text.length && /[\uD800-\uDBFF]/.test(text[end - 1]!)) end--; + const part = text.slice(start, end); + if (part.trim()) parts.push(part); + start = end; + } + return parts; +} + +type SavedReplyAction = { + id: string; + providerActionId: string; + status: string; + payload: Record; + result: Record | null; +}; +export type VoiceReplyPlan = + | { kind: "blocked" } + | { kind: "retry"; retryAt: number } + | { kind: "dispatch"; text: string; textDigest: string; partIndex: number; partCount: number; identity: string; priorActionId: string | null }; + +/** Select the next safe chunk without interpreting uncertainty as permission to resend. */ +export function planVoiceReply(input: { + replyId: string; + text: string; + questionNotification: boolean; + actions: readonly SavedReplyAction[]; + now: number; +}): VoiceReplyPlan { + const text = input.questionNotification + ? "Paperclip has a question available for this call. Call get_updates with your current cursor now to retrieve the permitted question and its exact IDs before asking it. This is a notification, not new work or permission. Do not call submit_request for this notification." + : input.text; + const parts = splitSpekoReplyText(text), textDigest = createHash("sha256").update(text).digest("hex"); + for (let partIndex = 0; partIndex < parts.length; partIndex++) { + const identity = `voice_reply_push:${input.replyId}${partIndex ? `:part:${partIndex}` : ""}`; + const prior = input.actions.find(action => action.providerActionId === identity); + // Accepted question notification remains a barrier until authorized retrieval. + if (prior?.status === "completed") { + if (input.questionNotification) return { kind: "blocked" }; + continue; + } + // Never append a changed answer to a prefix the provider already accepted. + if (input.actions.some(action => action.payload.textDigest && action.payload.textDigest !== textDigest)) return { kind: "blocked" }; + if (prior) { + if (prior.status !== "retry") return { kind: "blocked" }; + const retryAt = Number(prior.result?.retryAt ?? 0); + if (retryAt > input.now) return { kind: "retry", retryAt }; + } + return { kind: "dispatch", text: parts[partIndex]!, textDigest, partIndex, partCount: parts.length, identity, priorActionId: prior?.id ?? null }; + } + return { kind: "blocked" }; +} diff --git a/server/src/services/voice/voice-reply-push.ts b/server/src/services/voice/voice-reply-push.ts index a17e256a7a4..5eca00dc43a 100644 --- a/server/src/services/voice/voice-reply-push.ts +++ b/server/src/services/voice/voice-reply-push.ts @@ -1,5 +1,6 @@ import { notifyVoiceRecovery } from "./voice-recovery-notifications.js"; -import { createHash } from "node:crypto"; +import { planVoiceReply } from "./voice-reply-policy.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 { instanceSettingsService } from "../instance-settings.js"; @@ -10,20 +11,6 @@ import type { voiceSessionStore } from "./voice-session-store.js"; import { logger } from "../../middleware/logger.js"; const 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[] { - const parts: string[] = []; - for (let start = 0; start < text.length;) { - let end = Math.min(start + 16_000, text.length); - if (end < text.length && /[\uD800-\uDBFF]/.test(text[end - 1]!)) end--; - const part = text.slice(start, end); - if (part.trim()) parts.push(part); - start = end; - } - return parts; -} - /** 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: { @@ -65,26 +52,16 @@ export async function pushVoiceReplies(db: Db, options: { // 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 text = questionNotification ? "Paperclip has a question available for this call. Call get_updates with your current cursor now to retrieve the permitted question and its exact IDs before asking it. This is a notification, not new work or permission. Do not call submit_request for this notification." : row.payload.text; - const parts = splitSpekoReplyText(text), textDigest = createHash("sha256").update(text).digest("hex"); 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"); - for (let partIndex = 0; partIndex < parts.length; partIndex++) { - const identity = `voice_reply_push:${row.reply.id}${partIndex ? `:part:${partIndex}` : ""}`; - const prior = saved.find(action => action.providerActionId === identity); - // Questions remain an ordered barrier until authorized get_updates. - if (prior?.status === "completed") { if (questionNotification) return null; continue; } - // A changed publication cannot append a different answer to a partly - // delivered one. Unknown outcomes are never permission to resend. - if (saved.some(action => action.payload.textDigest && action.payload.textDigest !== textDigest) - || prior && (prior.status !== "retry" || Number(prior.result?.retryAt ?? 0) > Date.now())) return null; - const values = {status: "dispatching", result: null, updatedAt: new Date()}; - const action = prior - ? (await tx.update(chatActions).set(values).where(eq(chatActions.id, prior.id)).returning())[0]! - : (await tx.insert(chatActions).values({companyId: session.companyId, endpointId: session.endpointId, conversationId: session.conversationId, kind, providerActionId: identity, - payload: {sessionId: session.id, replyId: row.reply.id, publicationId: row.reply.publicationId, generation: session.generation, providerSessionId: session.providerSessionId, questionNotification, partIndex, partCount: parts.length, textDigest}, ...values}).returning())[0]!; - return {...context, reply: row.reply, questionNotification, text: parts[partIndex]!, partIndex, partCount: parts.length, action}; - } - return null; + const plan = planVoiceReply({ replyId: row.reply.id, text: row.payload.text, questionNotification, actions: saved, now: Date.now() }); + if (plan.kind !== "dispatch") return null; + 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, + 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 }; }).catch(error => { if (error instanceof HttpError && [403,404].includes(error.status)) { logger.warn({event: "voice.reply.push.authorization_blocked", companyId: candidate.companyId,