refactor(server): centralize ordered voice reply state reads

Co-Authored-By: Paperclip <noreply@paperclip.ing>

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.

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[] {

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,

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<VoiceSessionRow, "id" | "companyId" | "endpointId" | "conversationId" | "issueId">,
+) {
+  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() }) };
+}