Spaces:
Paused
Paused
| import { randomUUID } from "node:crypto"; | |
| import { ObjectId } from "mongodb"; | |
| import { authCondition } from "$lib/server/auth"; | |
| import { collections } from "$lib/server/database"; | |
| import { | |
| releaseWorkspaceRetirements, | |
| retiredWorkspaceFilter, | |
| retireAgentWorkspaces, | |
| WorkspaceRetirementConflict, | |
| } from "$lib/server/agent/retireWorkspaces"; | |
| import type { Conversation } from "$lib/types/Conversation"; | |
| import type { Generation } from "$lib/types/Generation"; | |
| import { GENERATION_REGISTRATION_STALE_MS } from "$lib/server/generation/writer"; | |
| export class ConversationDeletionConflict extends Error { | |
| constructor(message = "Conversation changed while it was being deleted") { | |
| super(message); | |
| this.name = "ConversationDeletionConflict"; | |
| } | |
| } | |
| const ACTIVE_GENERATION_DELETE_MESSAGE = | |
| "Stop the active generation before deleting this conversation"; | |
| const CONVERSATION_DELETION_LEASE_MS = 15 * 60 * 1_000; | |
| const STALE_GENERATION_REGISTRATION_MESSAGE = | |
| "Generation registration expired before the run started"; | |
| function isDuplicateKeyError(cause: unknown): boolean { | |
| return ( | |
| typeof cause === "object" && | |
| cause !== null && | |
| "code" in cause && | |
| (cause as { code?: number }).code === 11_000 | |
| ); | |
| } | |
| async function assertNoRunningGeneration(conversationId: Conversation["_id"]): Promise<void> { | |
| const running = await collections.generations.findOne( | |
| { conversationId, status: "running" }, | |
| { projection: { _id: 1 } } | |
| ); | |
| if (running) throw new ConversationDeletionConflict(ACTIVE_GENERATION_DELETE_MESSAGE); | |
| } | |
| async function reconcileStaleGenerationRegistration( | |
| conversation: Conversation, | |
| owner: Record<string, unknown>, | |
| now: Date | |
| ): Promise<Conversation> { | |
| const registration = conversation.generationRegistration; | |
| if (!registration) return conversation; | |
| const staleBefore = new Date(now.getTime() - GENERATION_REGISTRATION_STALE_MS); | |
| if (registration.claimedAt > staleBefore) { | |
| throw new ConversationDeletionConflict(ACTIVE_GENERATION_DELETE_MESSAGE); | |
| } | |
| // Consume the globally unique generation id before releasing a crashed | |
| // registration. If the delayed producer wins this insert race, its running | |
| // row is observed below and deletion remains blocked. If this tombstone wins, | |
| // that producer can never create a running orphan after the conversation is | |
| // gone. | |
| try { | |
| await collections.generations.insertOne({ | |
| _id: new ObjectId(), | |
| generationId: registration.generationId, | |
| conversationId: conversation._id, | |
| messageId: registration.messageId, | |
| ...(conversation.userId | |
| ? { userId: conversation.userId } | |
| : conversation.sessionId | |
| ? { sessionId: conversation.sessionId } | |
| : {}), | |
| status: "error", | |
| seq: 0, | |
| lastHeartbeatAt: now, | |
| startedAt: registration.claimedAt, | |
| endedAt: now, | |
| error: STALE_GENERATION_REGISTRATION_MESSAGE, | |
| ...(conversation.mode ? { mode: conversation.mode } : {}), | |
| ...(conversation.workspaceId ? { workspaceId: conversation.workspaceId } : {}), | |
| ...(conversation.mode === "agent" | |
| ? { agent: { sourceSequence: 0, controlsOpen: false } } | |
| : {}), | |
| createdAt: now, | |
| updatedAt: now, | |
| } satisfies Generation); | |
| } catch (cause) { | |
| if (!isDuplicateKeyError(cause)) throw cause; | |
| } | |
| const generation = await collections.generations.findOne( | |
| { generationId: registration.generationId }, | |
| { projection: { conversationId: 1, status: 1 } } | |
| ); | |
| if (generation?.conversationId.equals(conversation._id) && generation.status === "running") { | |
| throw new ConversationDeletionConflict(ACTIVE_GENERATION_DELETE_MESSAGE); | |
| } | |
| const cleared = await collections.conversations.findOneAndUpdate( | |
| { | |
| _id: conversation._id, | |
| ...owner, | |
| updatedAt: conversation.updatedAt, | |
| ...(conversation.mode === undefined | |
| ? { mode: { $exists: false } } | |
| : { mode: conversation.mode }), | |
| ...(conversation.workspaceId | |
| ? { workspaceId: conversation.workspaceId } | |
| : { workspaceId: { $exists: false } }), | |
| deletion: { $exists: false }, | |
| "generationRegistration.id": registration.id, | |
| }, | |
| { $unset: { generationRegistration: "" } }, | |
| { returnDocument: "after", includeResultMetadata: false } | |
| ); | |
| if (!cleared) throw new ConversationDeletionConflict(); | |
| return cleared; | |
| } | |
| /** | |
| * Delete one conversation without allowing a resumed Job to outlive its | |
| * workspace record. Every linked workspace stays retirement-leased through | |
| * the conversation CAS and its own exact-incarnation delete. | |
| */ | |
| export async function deleteConversationWithAgentWorkspaces( | |
| conversation: Conversation, | |
| locals: App.Locals | |
| ): Promise<void> { | |
| const owner = authCondition(locals); | |
| conversation = await reconcileStaleGenerationRegistration(conversation, owner, new Date()); | |
| // A successful delete may never strand a detached Generation after its | |
| // conversation/workspace authority disappears. Existing runs fail closed; | |
| // callers can stop them and retry once their terminal write has settled. | |
| await assertNoRunningGeneration(conversation._id); | |
| // Registration and deletion contend on this conversation write before any | |
| // linked workspace is retired. A fresh registration therefore cannot leave a | |
| // live conversation pointing at a Job which this failed delete already killed. | |
| const deletionId = randomUUID(); | |
| const deletionClaimedAt = new Date(); | |
| const deletionStaleBefore = new Date( | |
| deletionClaimedAt.getTime() - CONVERSATION_DELETION_LEASE_MS | |
| ); | |
| const claimedConversation = await collections.conversations.findOneAndUpdate( | |
| { | |
| _id: conversation._id, | |
| ...owner, | |
| updatedAt: conversation.updatedAt, | |
| ...(conversation.mode === undefined | |
| ? { mode: { $exists: false } } | |
| : { mode: conversation.mode }), | |
| ...(conversation.workspaceId | |
| ? { workspaceId: conversation.workspaceId } | |
| : { workspaceId: { $exists: false } }), | |
| generationRegistration: { $exists: false }, | |
| $or: [ | |
| { deletion: { $exists: false } }, | |
| { "deletion.claimedAt": { $lt: deletionStaleBefore } }, | |
| ], | |
| }, | |
| { $set: { deletion: { id: deletionId, claimedAt: deletionClaimedAt } } }, | |
| { returnDocument: "after", includeResultMetadata: false } | |
| ); | |
| if (!claimedConversation) { | |
| throw new ConversationDeletionConflict(); | |
| } | |
| const releaseConversationDeletion = () => | |
| collections.conversations.updateOne( | |
| { _id: conversation._id, "deletion.id": deletionId }, | |
| { $unset: { deletion: "" } } | |
| ); | |
| // Close the check -> claim race before any destructive workspace work. A | |
| // Generation whose registration inserted and released just before this claim | |
| // is now visible; once this claim exists, no new production registration can | |
| // pass its `deletion: absent` predicate. | |
| try { | |
| await assertNoRunningGeneration(conversation._id); | |
| } catch (cause) { | |
| await releaseConversationDeletion(); | |
| throw cause; | |
| } | |
| let linkedWorkspaces; | |
| try { | |
| linkedWorkspaces = await collections.workspaces | |
| .find({ | |
| ...owner, | |
| $or: [ | |
| { conversationId: conversation._id }, | |
| ...(conversation.workspaceId ? [{ _id: conversation.workspaceId }] : []), | |
| ], | |
| }) | |
| .toArray(); | |
| } catch (cause) { | |
| await releaseConversationDeletion(); | |
| throw cause; | |
| } | |
| let retired; | |
| try { | |
| retired = await retireAgentWorkspaces(linkedWorkspaces, locals); | |
| } catch (cause) { | |
| await releaseConversationDeletion(); | |
| if (cause instanceof WorkspaceRetirementConflict) { | |
| throw new ConversationDeletionConflict(); | |
| } | |
| throw cause; | |
| } | |
| // Defense in depth for non-production/manual writers which do not participate | |
| // in the conversation registration protocol. | |
| try { | |
| await assertNoRunningGeneration(conversation._id); | |
| } catch (cause) { | |
| await Promise.all([ | |
| releaseWorkspaceRetirements(retired, locals), | |
| releaseConversationDeletion(), | |
| ]); | |
| throw cause; | |
| } | |
| let deleted; | |
| try { | |
| deleted = await collections.conversations.deleteOne({ | |
| _id: conversation._id, | |
| "deletion.id": deletionId, | |
| }); | |
| } catch (cause) { | |
| await Promise.all([ | |
| releaseWorkspaceRetirements(retired, locals), | |
| releaseConversationDeletion(), | |
| ]); | |
| throw cause; | |
| } | |
| if (deleted.deletedCount !== 1) { | |
| await Promise.all([ | |
| releaseWorkspaceRetirements(retired, locals), | |
| releaseConversationDeletion(), | |
| ]); | |
| throw new ConversationDeletionConflict(); | |
| } | |
| for (const workspace of retired) { | |
| const removed = await collections.workspaces.deleteOne( | |
| retiredWorkspaceFilter(workspace, locals) | |
| ); | |
| if (removed.deletedCount !== 1) { | |
| // The owning conversation is already gone, but the lease still blocks | |
| // provisioning and the Job was confirmed stopped. Keep the tombstone for | |
| // an operator-visible retry instead of broadening this delete. | |
| throw new ConversationDeletionConflict(); | |
| } | |
| } | |
| } | |