fix(api): stop active runs before deleting threads

packages/api/src/features/agent/service.test.ts

@@ -1194,6 +1194,7 @@ describe("agentService.threads.delete", () => {
 
 		await agentService.threads.delete({ id: "thread-own", userId: "user-own" });
 
+		expect(dbMock.transaction).toHaveBeenCalled();
 		expect(dbMock.delete).toHaveBeenCalledBefore(storageServiceMock.delete as never);
 		expect(updateSet).toHaveBeenCalledBefore(storageServiceMock.delete as never);
 		expect(storageServiceMock.delete).toHaveBeenCalledWith("uploads/user-own/agent/thread-own");
@@ -1201,6 +1202,46 @@ describe("agentService.threads.delete", () => {
 		expect(updateSet).toHaveBeenCalledWith(expect.objectContaining({ status: "deleted" }));
 	});
 
+	it("clears active-run state before deleting an active thread", async () => {
+		const ownedThread = buildArchivedThread({
+			id: "thread-own",
+			userId: "user-own",
+			status: "active",
+			activeRunId: "run-1",
+			activeStreamId: "stream-1",
+			archivedAt: null,
+		});
+
+		dbMock.select.mockImplementation(() => {
+			const limit = vi.fn(async () => [ownedThread]);
+			const where = vi.fn(() => ({ limit }));
+			const from = vi.fn(() => ({ where }));
+			return { from };
+		});
+
+		const deleteWhere = vi.fn(async () => undefined);
+		dbMock.delete.mockReturnValue({ where: deleteWhere });
+
+		const updateWhere = vi.fn(async () => undefined);
+		const updateSet = vi.fn(() => ({ where: updateWhere }));
+		dbMock.update.mockReturnValue({ set: updateSet });
+		clearActiveAgentRunIfCurrentMock.mockResolvedValue(undefined);
+		storageServiceMock.delete.mockResolvedValue(undefined);
+
+		const { agentService } = await import("./service");
+
+		await agentService.threads.delete({ id: "thread-own", userId: "user-own" });
+
+		expect(clearActiveAgentRunIfCurrentMock).toHaveBeenCalledWith({
+			threadId: "thread-own",
+			userId: "user-own",
+			runId: "run-1",
+			streamId: "stream-1",
+		});
+		expect(clearActiveAgentRunIfCurrentMock).toHaveBeenCalledBefore(updateSet as never);
+		expect(updateSet).toHaveBeenCalledWith(expect.objectContaining({ status: "deleted" }));
+	});
+
 	it("still completes when storage cleanup fails after the thread is soft-deleted", async () => {
 		const ownedThread = buildArchivedThread({
 			id: "thread-own",
@@ -1232,6 +1273,32 @@ describe("agentService.threads.delete", () => {
 		await expect(agentService.threads.delete({ id: "thread-own", userId: "user-own" })).resolves.toBeUndefined();
 		expect(updateSet).toHaveBeenCalledWith(expect.objectContaining({ status: "deleted" }));
 	});
+
+	it("does not delete storage when transactional soft-delete fails", async () => {
+		const ownedThread = buildArchivedThread({
+			id: "thread-own",
+			userId: "user-own",
+			status: "active",
+			activeRunId: null,
+			activeStreamId: null,
+			archivedAt: null,
+		});
+
+		dbMock.select.mockImplementation(() => {
+			const limit = vi.fn(async () => [ownedThread]);
+			const where = vi.fn(() => ({ limit }));
+			const from = vi.fn(() => ({ where }));
+			return { from };
+		});
+		dbMock.transaction.mockRejectedValue(new Error("transaction failed"));
+
+		const { agentService } = await import("./service");
+
+		await expect(agentService.threads.delete({ id: "thread-own", userId: "user-own" })).rejects.toThrow(
+			"transaction failed",
+		);
+		expect(storageServiceMock.delete).not.toHaveBeenCalled();
+	});
 });
 
 describe("agentService.actions.revert", () => {

packages/api/src/features/agent/service.ts

@@ -944,13 +944,32 @@ export const agentService = {
 		delete: async (input: { id: string; userId: string }) => {
 			assertAgentEnvironment();
 
-			await getThread({ id: input.id, userId: input.userId });
+			const thread = await getThread({ id: input.id, userId: input.userId });
+			const activeRunId = thread.activeRunId;
+			const activeStreamId = thread.activeStreamId;
 
-			await db.delete(schema.agentAttachment).where(eq(schema.agentAttachment.threadId, input.id));
-			await db
-				.update(schema.agentThread)
-				.set({ status: "deleted", deletedAt: new Date() })
-				.where(and(eq(schema.agentThread.id, input.id), eq(schema.agentThread.userId, input.userId)));
+			if (activeRunId) {
+				activeRunControllers.get(activeRunId)?.abort("USER_DELETED");
+				activeRunControllers.delete(activeRunId);
+				try {
+					await clearActiveAgentRunIfCurrent({
+						threadId: input.id,
+						userId: input.userId,
+						runId: activeRunId,
+						streamId: activeStreamId,
+					});
+				} catch (error) {
+					console.error("[agent] Failed to clear active run during delete", error);
+				}
+			}
+
+			await db.transaction(async (tx) => {
+				await tx.delete(schema.agentAttachment).where(eq(schema.agentAttachment.threadId, input.id));
+				await tx
+					.update(schema.agentThread)
+					.set({ status: "deleted", deletedAt: new Date() })
+					.where(and(eq(schema.agentThread.id, input.id), eq(schema.agentThread.userId, input.userId)));
+			});
 
 			try {
 				await getStorageService().delete(`uploads/${input.userId}/agent/${input.id}`);