1import type {
2 OrchestrationV2Command,
3 OrchestrationV2DomainEvent,
4 OrchestrationV2ThreadProjection,
5} from "@t3tools/contracts";
6import type * as DateTime from "effect/DateTime";
7import * as Effect from "effect/Effect";
8
9import type { PendingOrchestrationEffectV2 } from "./EffectOutbox.ts";
10import type * as IdAllocator from "@t3tools/provider-core/server/IdAllocator";
11
12export interface ThreadDeletionPlan {
13 readonly events: ReadonlyArray<OrchestrationV2DomainEvent>;
14 readonly effects: ReadonlyArray<PendingOrchestrationEffectV2>;
15}
16
17/** Plan the same durable cleanup for direct thread deletion and project removal. */
18export const planThreadDeletion = Effect.fn("ThreadDeletion.planThreadDeletion")(function* (input: {
19 readonly command: Extract<OrchestrationV2Command, { readonly type: "thread.delete" }>;
20 readonly projection: Pick<
21 OrchestrationV2ThreadProjection,
22 "thread" | "runs" | "attempts" | "nodes" | "runtimeRequests" | "subagents" | "providerSessions"
23 >;
24 readonly attachmentIds: ReadonlyArray<string>;
25 readonly now: DateTime.Utc;
26 readonly idAllocator: IdAllocator.IdAllocatorV2["Service"];
27}): Effect.fn.Return<ThreadDeletionPlan, IdAllocator.IdAllocatorV2Error> {
28 const { command, projection, now, idAllocator } = input;
29 const events: Array<OrchestrationV2DomainEvent> = [];
30 const effects: Array<PendingOrchestrationEffectV2> = [];
31 let current = projection;
32 const emitEvent = Effect.fn("ThreadDeletion.emitEvent")(function* <
33 Event extends OrchestrationV2DomainEvent,
34 >(event: Omit<Event, "id">) {
35 const id = yield* idAllocator.allocate.event({
36 threadId: event.threadId,
37 commandId: command.commandId,
38 });
39 const withId = { ...event, id } as Event;
40 events.push(withId);
41 if (withId.type === "run.updated") {
42 current = {
43 ...current,
44 runs: current.runs.map((run) => (run.id === withId.payload.id ? withId.payload : run)),
45 };
46 }
47 if (withId.type === "subagent.updated") {
48 current = {
49 ...current,
50 subagents: current.subagents.map((task) =>
51 task.id === withId.payload.id ? withId.payload : task,
52 ),
53 };
54 }
55 });
56
57 yield* emitEvent({
58 type: "thread.deleted",
59 threadId: command.threadId,
60 providerInstanceId: projection.thread.providerInstanceId,
61 occurredAt: now,
62 payload: {
63 ...projection.thread,
64 deletedAt: projection.thread.deletedAt ?? now,
65 titleRegeneration: null,
66 updatedAt: now,
67 },
68 });
69
70 const activeRuns = projection.runs.filter((run) =>
71 ["preparing", "queued", "starting", "running", "waiting"].includes(run.status),
72 );
73 const activeRunById = new Map(activeRuns.map((run) => [run.id, run]));
74 for (const run of activeRuns) {
75 yield* emitEvent({
76 type: "run.updated",
77 threadId: command.threadId,
78 runId: run.id,
79 providerInstanceId: run.providerInstanceId,
80 occurredAt: now,
81 payload: { ...run, status: "cancelled", queuePosition: null, completedAt: now },
82 });
83 }
84 for (const attempt of projection.attempts) {
85 const run = activeRunById.get(attempt.runId);
86 if (run === undefined || (attempt.status !== "pending" && attempt.status !== "running")) {
87 continue;
88 }
89 yield* emitEvent({
90 type: "run-attempt.updated",
91 threadId: command.threadId,
92 runId: attempt.runId,
93 nodeId: attempt.rootNodeId,
94 providerInstanceId: run.providerInstanceId,
95 occurredAt: now,
96 payload: { ...attempt, status: "cancelled", completedAt: now },
97 });
98 }
99 for (const node of projection.nodes) {
100 const run = node.runId === null ? undefined : activeRunById.get(node.runId);
101 if (run === undefined || !["pending", "running", "waiting"].includes(node.status)) {
102 continue;
103 }
104 yield* emitEvent({
105 type: "node.updated",
106 threadId: command.threadId,
107 runId: run.id,
108 nodeId: node.id,
109 providerInstanceId: run.providerInstanceId,
110 occurredAt: now,
111 payload: { ...node, status: "cancelled", completedAt: now },
112 });
113 }
114 for (const request of projection.runtimeRequests.filter(
115 (request) => request.status === "pending",
116 )) {
117 yield* emitEvent({
118 type: "runtime-request.updated",
119 threadId: command.threadId,
120 nodeId: request.nodeId,
121 occurredAt: now,
122 payload: {
123 ...request,
124 status: "cancelled",
125 responseCapability: { type: "not_resumable", reason: "The thread was deleted." },
126 resolvedAt: now,
127 },
128 });
129 }
130
131 const cohortRunIds = new Set([
132 ...current.runs.filter((run) => run.delegatedCompletion !== undefined).map((run) => run.id),
133 ...current.subagents
134 .filter((task) => task.origin === "app_owned" && task.runId !== null)
135 .map((task) => task.runId!),
136 ]);
137 for (const parentRunId of cohortRunIds) {
138 // Use the projected cancellation so disposing a cohort cannot revive its run.
139 const parentRun = current.runs.find((run) => run.id === parentRunId);
140 if (parentRun === undefined) continue;
141 const cohort = parentRun.delegatedCompletion;
142 const tasks = current.subagents.filter(
143 (task) => task.origin === "app_owned" && task.runId === parentRunId,
144 );
145 yield* emitEvent({
146 type: "run.updated",
147 threadId: parentRun.threadId,
148 runId: parentRun.id,
149 ...(parentRun.rootNodeId === null ? {} : { nodeId: parentRun.rootNodeId }),
150 providerInstanceId: parentRun.providerInstanceId,
151 occurredAt: now,
152 payload: {
153 ...parentRun,
154 delegatedCompletion: {
155 disposition: "disposed",
156 nextGeneration: cohort?.nextGeneration ?? 1,
157 delivery: null,
158 },
159 },
160 });
161 for (const task of tasks) {
162 if (
163 task.completionDelivery?.state === "acknowledged" ||
164 task.completionDelivery?.state === "delivered" ||
165 task.completionDelivery?.state === "disposed"
166 ) {
167 continue;
168 }
169 yield* emitEvent({
170 type: "subagent.updated",
171 threadId: parentRun.threadId,
172 ...(task.runId === null ? {} : { runId: task.runId }),
173 nodeId: task.id,
174 driver: task.driver,
175 providerInstanceId: task.providerInstanceId,
176 occurredAt: now,
177 payload: {
178 ...task,
179 completionDelivery: { state: "disposed", observedByRunId: null },
180 updatedAt: now,
181 },
182 });
183 }
184 }
185
186 for (const session of projection.providerSessions) {
187 if (session.status === "stopped" || session.status === "error") continue;
188 yield* emitEvent({
189 type: "provider-session.detached",
190 threadId: command.threadId,
191 driver: session.driver,
192 providerInstanceId: session.providerInstanceId,
193 occurredAt: now,
194 payload: {
195 providerSessionId: session.id,
196 detachedAt: now,
197 reason: "Thread deleted.",
198 },
199 });
200 effects.push({
201 id: `effect:${command.commandId}:provider-session.detach:${session.id}`,
202 commandId: command.commandId,
203 threadId: command.threadId,
204 request: {
205 type: "provider-session.detach",
206 providerSessionId: session.id,
207 detail: "Thread deleted.",
208 revokeMcpCredential: true,
209 },
210 });
211 }
212 effects.push({
213 id: `effect:${command.commandId}:terminal.cleanup`,
214 commandId: command.commandId,
215 threadId: command.threadId,
216 request: { type: "terminal.cleanup" },
217 });
218 effects.push({
219 id: `effect:${command.commandId}:preview.cleanup`,
220 commandId: command.commandId,
221 threadId: command.threadId,
222 request: { type: "preview.cleanup" },
223 });
224 const attachmentIds = Array.from(new Set(input.attachmentIds));
225 if (attachmentIds.length > 0) {
226 effects.push({
227 id: `effect:${command.commandId}:attachment.cleanup`,
228 commandId: command.commandId,
229 threadId: command.threadId,
230 request: { type: "attachment.cleanup", attachmentIds },
231 });
232 }
233 return { events, effects };
234});