mirror of
https://github.com/paperclipai/paperclip.git
synced 2026-11-04 11:01:16 +01:00
Merge branch 'codex/sleep-24-voice-inbound-retention' into codex/sleep-25-voice-call-reports
* codex/sleep-24-voice-inbound-retention: fix(server): scope inbound retention wakes and protect maintenance boundary
This commit is contained in:
commit
53aa3bd603
3 files changed
+74
-2
No files matched your search
@@ -119,3 +119,33 @@ Final local consumer checks pass: 52 company-deletion/startup tests, 11 parent
|
||||
voice recovery tests, and final server typecheck. Selected nonoverlapping test
|
||||
coverage totals 218 passing cases. The implementation and local verification are
|
||||
complete; remote PR CI/review and the rest of the original stack remain pending.
|
||||
|
||||
## Review repairs
|
||||
|
||||
The first published head 9c38ddfe53 passed all CI and Greptile 5/5. Paperclip
|
||||
Review found unnecessary endpoint-wide retention refresh on every new call and
|
||||
missing behavioral coverage of removal from the maintenance lane.
|
||||
|
||||
Pre-call admission now allocates its ID before insertion and registers the
|
||||
existing intent for that admission only. Retention lookup uses the primary key;
|
||||
new calls no longer scan the endpoint's retained history. Cleanup receives the
|
||||
same precise hint. Existing endpoint/company configuration changes retain their
|
||||
scope, and startup still reconstructs eligible history. No schema change.
|
||||
|
||||
Two service regressions verify the actual signed pre-call scope and drive the
|
||||
real chat coordinator/voice maintenance path with due, confirmed history. The row
|
||||
survives maintenance and is deleted by the retention worker. Temporarily restoring
|
||||
pruning to voice maintenance makes this latter test fail with the row missing;
|
||||
restoring the implementation passes. All 133 full voice/app wiring tests and
|
||||
server typecheck pass after the repair. Logs are /tmp/sleep24-review-acceptance.log,
|
||||
/tmp/sleep24-review-types.log, /tmp/sleep24-maintenance-mutation.log and
|
||||
/tmp/sleep24-maintenance-restored.log. Typecheck covers production sources; the
|
||||
server tsconfig excludes the test directory, whose behavior is verified by Vitest.
|
||||
Fresh remote checks/review are required on the repaired head.
|
||||
|
||||
Parent #17580 has all checks terminal and Greptile 5/5 at 4f0364d269. Its final
|
||||
Paperclip review added two findings: unresolved cleanup can block deletion, and
|
||||
rejection awaits provider calls. These require explicit disposition/repair at the
|
||||
parent; green checks do not resolve them. An arbitrary retry cap or credential
|
||||
failure is not proof that a provider call ended and must not silently discard the
|
||||
cleanup obligation. Preserve that requirement while assessing the deletion path.
|
||||
@@ -1,3 +1,6 @@
|
||||
import { createChatReconciliationCoordinator } from "../app.js";
|
||||
import { subscribeDatabaseWork } from "@paperclipai/db";
|
||||
import { DELIVERY_QUEUES } from "../services/delivery-work-notifications.js";
|
||||
import * as voiceReplyNotifications from "../services/voice/voice-reply-notifications.js";
|
||||
import { voiceReplySchedule } from "../services/voice/voice-reply-schedule.js";
|
||||
import { registerVoiceReplyWork } from "../services/voice/voice-reply-work.js";
|
||||
@@ -78,6 +81,44 @@ const support = external ? { supported: true } : await getEmbeddedPostgresTestSu
|
||||
return { companyId, endpointId, agentId, connectionId, issueId, conversationId, callerId, sessionId, store, request, reserved, accept, toolInput, publication };
|
||||
}
|
||||
|
||||
it("limits a new pre-call wake to its admission instead of scanning endpoint history", async () => {
|
||||
const f = await inboundFixture(); const hints: string[] = [];
|
||||
const unsubscribe = subscribeDatabaseWork(db, DELIVERY_QUEUES.voiceInboundRetention,
|
||||
(event, scopes) => { if (event === "settled") hints.push(...scopes ?? ["*"]); });
|
||||
try {
|
||||
const id = randomUUID(), incoming = f.sign({ ...f.event, session_id: id, call_id: id });
|
||||
await f.service.inbound.lifecycle(f.endpoint.publicId, incoming.body, incoming.headers);
|
||||
const [saved] = await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.providerSessionId, id));
|
||||
expect(hints).toEqual([`admission:${saved.id}`]);
|
||||
} finally {
|
||||
unsubscribe(); await db.delete(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.companyId, f.companyId));
|
||||
}
|
||||
});
|
||||
|
||||
it("leaves due history intact during chat maintenance and deletes it through retention", async () => {
|
||||
const f = await inboundFixture();
|
||||
const ended = f.sign({ type: "call.status", call_id: f.call.providerSessionId, status: "ended" });
|
||||
await f.service.inbound.lifecycle(f.endpoint.publicId, ended.body, ended.headers);
|
||||
await db.update(chatVoiceInboundCalls).set({ expiresAt: new Date(0), updatedAt: new Date(0) }).where(eq(chatVoiceInboundCalls.id, f.call.id));
|
||||
const onError = vi.fn();
|
||||
const maintenance = createChatReconciliationCoordinator({
|
||||
reconcileVoiceSessions: () => f.service.reconcileMaintenance(),
|
||||
processPendingDeliveries: async () => undefined, projectRunMilestones: async () => 0,
|
||||
flushPublications: async () => undefined, onError,
|
||||
});
|
||||
try {
|
||||
maintenance.reconcile(); await maintenance.drain();
|
||||
expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id))).toHaveLength(1);
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
} finally { maintenance.stop(); await maintenance.drain(); }
|
||||
const coordinator = createDeliveryWorkCoordinator({ owner: db, canRun: () => true, onError });
|
||||
try {
|
||||
await f.service.inbound.registerRetentionWork(coordinator, () => true).ready;
|
||||
expect(await db.select().from(chatVoiceInboundCalls).where(eq(chatVoiceInboundCalls.id, f.call.id))).toHaveLength(0);
|
||||
expect(onError).not.toHaveBeenCalled();
|
||||
} finally { await coordinator.stop(); }
|
||||
});
|
||||
|
||||
it.each(["signed callback", "cleanup receipt"])("schedules retained history from a real %s", async confirmation => {
|
||||
const f = await inboundFixture();
|
||||
vi.useFakeTimers({ toFake: ["Date", "setTimeout", "clearTimeout"], shouldAdvanceTime: true });
|
||||
|
||||
@@ -123,8 +123,9 @@ export function voiceInboundService(db: Db, options: {
|
||||
if (prior) return prior;
|
||||
const recent = await tx.select({state: chatVoiceInboundCalls.state, expiresAt: chatVoiceInboundCalls.expiresAt}).from(chatVoiceInboundCalls).where(and(eq(chatVoiceInboundCalls.companyId, row.companyId), eq(chatVoiceInboundCalls.endpointId, row.id), gte(chatVoiceInboundCalls.createdAt, new Date(Date.now() - 3_600_000)))).limit(60);
|
||||
if (recent.length >= 60 || recent.filter(call => ["guest_intake", "awaiting_approval", "approving"].includes(call.state) && call.expiresAt.getTime() > Date.now()).length >= 10) throw conflict("This phone line has reached its incoming call limit. Try again later.");
|
||||
await notifyVoiceInbound(tx, { kind: "endpoint", id: row.id });
|
||||
const [admission] = await tx.insert(chatVoiceInboundCalls).values({companyId: row.companyId, endpointId: row.id, providerSessionId, providerAccountId: event.organization_id, state: line.guestIntake ? "guest_intake" : "awaiting_approval", approvalCode: String(randomInt(100000, 1000000)), generation: context.generation, credentialFingerprint: context.fingerprint, toolTokenHash: voiceTokenHash(capability), requestFingerprint: fingerprint, callerAuthority: line.guestIntake ? "guest_intake" : null, expiresAt: new Date(Date.now() + (line.guestIntake ? 600_000 : 120_000))}).onConflictDoNothing().returning();
|
||||
const admissionId = randomUUID();
|
||||
await notifyVoiceInbound(tx, { kind: "admission", id: admissionId });
|
||||
const [admission] = await tx.insert(chatVoiceInboundCalls).values({id: admissionId, companyId: row.companyId, endpointId: row.id, providerSessionId, providerAccountId: event.organization_id, state: line.guestIntake ? "guest_intake" : "awaiting_approval", approvalCode: String(randomInt(100000, 1000000)), generation: context.generation, credentialFingerprint: context.fingerprint, toolTokenHash: voiceTokenHash(capability), requestFingerprint: fingerprint, callerAuthority: line.guestIntake ? "guest_intake" : null, expiresAt: new Date(Date.now() + (line.guestIntake ? 600_000 : 120_000))}).onConflictDoNothing().returning();
|
||||
return admission;
|
||||
});
|
||||
if (!existing || existing.requestFingerprint !== fingerprint || existing.toolTokenHash !== voiceTokenHash(capability) || !["awaiting_approval", "guest_intake"].includes(existing.state) || existing.expiresAt.getTime() <= Date.now()) throw forbidden("Incoming call is no longer pending");
|
||||
|
||||
Reference in new issue
Block a user