|
1 | 1 | import { afterAll, afterEach, describe, expect, test } from "bun:test"; |
2 | 2 | import { randomUUID } from "node:crypto"; |
3 | 3 | import { FIRING_FRAME, frameFiring } from "@openbot/shared/routine-firing"; |
4 | | -import { eq } from "drizzle-orm"; |
| 4 | +import { eq, sql } from "drizzle-orm"; |
| 5 | +import postgres from "postgres"; |
| 6 | +import { |
| 7 | + CHANNEL_ACTIVITY_TOPIC, |
| 8 | + type ChannelActivityEvent, |
| 9 | +} from "../src/channels/events"; |
5 | 10 | import { createAgentProfileStore } from "../src/agents/profile-store"; |
6 | 11 | import type { AgentActor } from "../src/agents/profile-types"; |
7 | 12 | import { createChannelStore } from "../src/channels/routes"; |
@@ -360,11 +365,12 @@ describe("naming a claimed conversation", () => { |
360 | 365 | // Offered before it was named, which is exactly the race two replicas produce. |
361 | 366 | await queue.offer({ kind: CHANNEL_SUMMARY_KIND, key: channel.id }); |
362 | 367 |
|
363 | | - await summariseClaimedChannels( |
| 368 | + const report = await summariseClaimedChannels( |
364 | 369 | options({ title: titler("A second opinion") }), |
365 | 370 | ); |
366 | 371 |
|
367 | 372 | expect((await summaryOf(channel.id))?.summary).toBe("Named already"); |
| 373 | + expect(report.written).toContain(channel.id); |
368 | 374 | }); |
369 | 375 |
|
370 | 376 | test("leaves the conversation unnamed when the model has no answer", async () => { |
@@ -432,6 +438,70 @@ describe("naming a claimed conversation", () => { |
432 | 438 | expect(item?.attempts).toBe(1); |
433 | 439 | }); |
434 | 440 |
|
| 441 | + test.each(["soft", "hard"] as const)( |
| 442 | + "a %s delete while the model is naming a conversation is final", |
| 443 | + async (kind) => { |
| 444 | + const owner = await createUser(); |
| 445 | + const channel = await createUsedChannel(owner); |
| 446 | + await offer(channel.id); |
| 447 | + const started = Promise.withResolvers<void>(); |
| 448 | + const resume = Promise.withResolvers<void>(); |
| 449 | + const work = summariseClaimedChannels( |
| 450 | + options({ |
| 451 | + title: async () => { |
| 452 | + started.resolve(); |
| 453 | + await resume.promise; |
| 454 | + return "A title that arrived after deletion"; |
| 455 | + }, |
| 456 | + }), |
| 457 | + ); |
| 458 | + void work.catch(started.reject); |
| 459 | + const notifications = postgres(databaseUrl, { max: 1 }); |
| 460 | + const barrier = `summary-barrier-${randomUUID()}`; |
| 461 | + const observed = Promise.withResolvers<void>(); |
| 462 | + const summaries: string[] = []; |
| 463 | + try { |
| 464 | + await notifications.listen(CHANNEL_ACTIVITY_TOPIC, (payload) => { |
| 465 | + if (payload === barrier) { |
| 466 | + observed.resolve(); |
| 467 | + return; |
| 468 | + } |
| 469 | + const event = JSON.parse(payload) as ChannelActivityEvent; |
| 470 | + if (event.channelId === channel.id && event.summary !== undefined) |
| 471 | + summaries.push(event.summary); |
| 472 | + }); |
| 473 | + await started.promise; |
| 474 | + if (kind === "soft") |
| 475 | + expect(await store.softDelete(owner, channel.id)).toBe(true); |
| 476 | + else await database.delete(channels).where(eq(channels.id, channel.id)); |
| 477 | + resume.resolve(); |
| 478 | + const report = await work; |
| 479 | + // NOTIFY is commit-ordered. This marker follows the completed worker transaction, so |
| 480 | + // observing it also observes any summary announcement that worker could have committed. |
| 481 | + await database.execute( |
| 482 | + sql`select pg_notify(${CHANNEL_ACTIVITY_TOPIC}, ${barrier})`, |
| 483 | + ); |
| 484 | + await observed.promise; |
| 485 | + expect(summaries).toEqual([]); |
| 486 | + const row = await summaryOf(channel.id); |
| 487 | + if (kind === "soft") { |
| 488 | + expect(row?.summary).toBeNull(); |
| 489 | + expect(row?.summaryAt).toBeNull(); |
| 490 | + } else expect(row).toBeUndefined(); |
| 491 | + expect(report.written).not.toContain(channel.id); |
| 492 | + const [item] = await database |
| 493 | + .select({ finishedAt: workItems.finishedAt }) |
| 494 | + .from(workItems) |
| 495 | + .where(eq(workItems.key, channel.id)); |
| 496 | + expect(item?.finishedAt).not.toBeNull(); |
| 497 | + } finally { |
| 498 | + resume.resolve(); |
| 499 | + await Promise.allSettled([work]); |
| 500 | + await notifications.end({ timeout: 1 }); |
| 501 | + } |
| 502 | + }, |
| 503 | + ); |
| 504 | + |
435 | 505 | test("a soft-deleted conversation is dropped the same way", async () => { |
436 | 506 | const owner = await createUser(); |
437 | 507 | const channel = await createUsedChannel(owner); |
|
0 commit comments