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