Open History to see the last commit
1import { resolveProjectSettings } from "@t3tools/shared/projectSettings";
2import { isAttachmentOnlyBootstrapPrompt } from "@t3tools/shared/composerContextReferences";
3import {
4 CommandId,
5 type ChatAttachment,
6 type MessageId,
7 type OrchestrationV2AppThread,
8 type OrchestrationV2ConversationMessage,
9 type RunId,
10 type ServerSettingsError,
11 type ThreadId,
12} from "@t3tools/contracts";
13import * as Cause from "effect/Cause";
14import * as Context from "effect/Context";
15import * as Effect from "effect/Effect";
16import * as FileSystem from "effect/FileSystem";
17import * as Layer from "effect/Layer";
18import * as Option from "effect/Option";
19import * as Schedule from "effect/Schedule";
20
21import * as ServerSettings from "../serverSettings.ts";
22import * as TextGeneration from "../textGeneration/TextGeneration.ts";
23import type { OrchestratorV2Error } from "./Orchestrator.ts";
24import * as ProjectStore from "./ProjectStore.ts";
25import * as ThreadManagementService from "./ThreadManagementService.ts";
26
27import { formatThreadTitleContext } from "@t3tools/provider-core/server/threadTitleContext";
28export { formatThreadTitleContext } from "@t3tools/provider-core/server/threadTitleContext";
29
30export class ThreadTitleRegenerationService extends Context.Service<
31 ThreadTitleRegenerationService,
32 {
33 readonly execute: (input: {
34 readonly threadId: ThreadId;
35 readonly requestId: CommandId;
36 readonly kind:
37 | { readonly type: "initial"; readonly messageId: MessageId }
38 | { readonly type: "regenerate" };
39 }) => Effect.Effect<
40 void,
41 | OrchestratorV2Error
42 | ProjectStore.ProjectStoreV2Error
43 | ServerSettingsError
44 | ThreadManagementService.ThreadManagementProviderInstanceDisabledError
45 >;
46 }
47>()("t3/orchestration-v2/ThreadTitleRegenerationService") {}
48
49const make = Effect.gen(function* () {
50 const threads = yield* ThreadManagementService.ThreadManagementService;
51 const projects = yield* ProjectStore.ProjectStoreV2;
52 const serverSettings = yield* ServerSettings.ServerSettingsService;
53 const textGeneration = yield* TextGeneration.TextGeneration;
54 const fileSystem = yield* FileSystem.FileSystem;
55
56 const complete = (input: {
57 readonly threadId: ThreadId;
58 readonly requestId: CommandId;
59 readonly title?: string;
60 }) =>
61 threads
62 .dispatch({
63 type: "thread.title.regeneration.complete",
64 commandId: CommandId.make(`${input.requestId}:title-complete`),
65 threadId: input.threadId,
66 requestId: input.requestId,
67 ...(input.title === undefined ? {} : { title: input.title }),
68 })
69 .pipe(Effect.asVoid);
70
71 const messagesThroughRun = Effect.fn("ThreadTitleRegenerationService.messagesThroughRun")(
72 function* (threadId: ThreadId, runId: RunId) {
73 const source = yield* threads.getThreadRecords(threadId, ["messages", "runs"], {
74 messageRoles: ["user", "assistant"],
75 });
76 const ordinals = new Map(
77 source.runs
78 .filter((run) => run.status !== "rolled_back")
79 .map((run) => [run.id, run.ordinal]),
80 );
81 const forkOrdinal = ordinals.get(runId) ?? -1;
82 const messages = source.messages.filter((message) => {
83 if (message.streaming) return false;
84 if (message.runId === null) return source.thread.historyOrigin === "v1_import";
85 return (ordinals.get(message.runId) ?? Infinity) <= forkOrdinal;
86 });
87 return { thread: source.thread, messages };
88 },
89 );
90
91 // A fork starts with no messages of its own, so its subject lives in the history it inherited.
92 const forkSource = Effect.fn("ThreadTitleRegenerationService.forkSource")(function* (
93 thread: OrchestrationV2AppThread,
94 ) {
95 let title: string | undefined;
96 const history: Array<ReadonlyArray<OrchestrationV2ConversationMessage>> = [];
97 let forkedFrom = thread.forkedFrom;
98 while (forkedFrom?.type === "run") {
99 const source = yield* messagesThroughRun(forkedFrom.threadId, forkedFrom.runId);
100 history.push(source.messages);
101 title ??= source.thread.title;
102 forkedFrom = source.thread.forkedFrom;
103 }
104 const context = formatThreadTitleContext(history.reverse().flat());
105 return title === undefined || context.message.length === 0
106 ? undefined
107 : { title, history: context.message };
108 });
109
110 const execute: ThreadTitleRegenerationService["Service"]["execute"] = Effect.fn(
111 "ThreadTitleRegenerationService.execute",
112 )(function* (input) {
113 const outcome:
114 | { readonly type: "stale" }
115 | { readonly type: "complete"; readonly title?: string } = yield* Effect.gen(function* () {
116 const projection = yield* threads.getThreadRecords(
117 input.threadId,
118 ["messages"],
119 input.kind.type === "initial"
120 ? { messageIds: [input.kind.messageId] }
121 : { messageRoles: ["user", "assistant"] },
122 );
123 if (projection.thread.titleRegeneration?.requestId !== input.requestId) {
124 return { type: "stale" as const };
125 }
126
127 const project = yield* projects.get(projection.thread.projectId);
128 if (Option.isNone(project)) {
129 return { type: "complete" as const };
130 }
131
132 let context: {
133 readonly message: string;
134 readonly attachments: ReadonlyArray<ChatAttachment>;
135 };
136 if (input.kind.type === "initial") {
137 const messageId = input.kind.messageId;
138 const message = projection.messages.find(
139 (candidate) => candidate.id === messageId && !candidate.streaming,
140 );
141 context =
142 message === undefined
143 ? { message: "", attachments: [] }
144 : {
145 message: isAttachmentOnlyBootstrapPrompt(message.text) ? "" : message.text,
146 attachments: message.attachments,
147 };
148 } else {
149 context = formatThreadTitleContext(
150 projection.messages.filter((message) => !message.streaming),
151 );
152 }
153 if (context.message.length === 0 && context.attachments.length === 0) {
154 return { type: "complete" as const };
155 }
156
157 // A deleted source should not cost the fork its title.
158 const inherited = yield* forkSource(projection.thread).pipe(
159 Effect.orElseSucceed(() => undefined),
160 );
161 const settings = resolveProjectSettings(
162 yield* serverSettings.getSettings,
163 projection.thread.projectId,
164 ).settings;
165 // Storage cleanup removes worktrees but keeps the path; CLIs cannot spawn in a missing cwd.
166 const worktreePath = projection.thread.worktreePath;
167 const worktreeExists =
168 worktreePath !== null &&
169 (yield* fileSystem.exists(worktreePath).pipe(Effect.orElseSucceed(() => false)));
170 const result = yield* textGeneration.generateThreadTitle({
171 cwd: worktreeExists ? worktreePath : project.value.workspaceRoot,
172 message: context.message,
173 projectTitle: project.value.title,
174 attachments: context.attachments,
175 ...(input.kind.type === "regenerate" ? { previousTitle: projection.thread.title } : {}),
176 ...(inherited === undefined ? {} : { forkSource: inherited }),
177 instructions: settings.threadTitleInstructions,
178 modelSelection: settings.textGenerationModelSelection,
179 projectId: projection.thread.projectId,
180 });
181 const generatedTitle = result.title.trim();
182 return generatedTitle === "New thread" ||
183 (input.kind.type === "regenerate" && generatedTitle === projection.thread.title.trim())
184 ? { type: "complete" as const }
185 : { type: "complete" as const, title: result.title };
186 }).pipe(
187 Effect.retry({
188 times: input.kind.type === "initial" ? 2 : 0,
189 schedule: Schedule.exponential("2 seconds"),
190 }),
191 Effect.catchCause((cause) =>
192 Cause.hasInterruptsOnly(cause)
193 ? Effect.interrupt
194 : Effect.logWarning("Thread title generation failed", {
195 threadId: input.threadId,
196 requestId: input.requestId,
197 cause,
198 }).pipe(Effect.as({ type: "complete" as const })),
199 ),
200 );
201
202 if (outcome.type === "stale") {
203 return;
204 }
205 yield* complete({
206 ...input,
207 ...(outcome.title === undefined ? {} : { title: outcome.title }),
208 });
209 });
210
211 return ThreadTitleRegenerationService.of({ execute });
212});
213
214export const layer = Layer.effect(ThreadTitleRegenerationService, make);