From 03a7c47fdd582d1f2fb69da613dbdec2a721b4aa Mon Sep 17 00:00:00 2001 From: Yusuf007R Date: Sat, 1 Aug 2026 02:18:32 -0500 Subject: [PATCH 1/4] fix(orchestration-v2): restore generated thread titles Queue automatic and manual title generation through the durable V2 effect outbox, use the configured text-generation model, and cover web, desktop, and mobile send paths. --- .../src/state/use-thread-outbox-drain.ts | 6 +- .../src/orchestration-v2/EffectOutbox.ts | 8 + .../src/orchestration-v2/EffectWorker.test.ts | 41 ++++ .../src/orchestration-v2/EffectWorker.ts | 20 ++ .../src/orchestration-v2/Orchestrator.ts | 74 ++++++- .../ThreadLaunchService.test.ts | 186 ++++++++++++++--- .../orchestration-v2/ThreadLaunchService.ts | 31 +-- .../ThreadTitleRegenerationService.ts | 195 +++++++++++------- .../src/orchestration-v2/runtimeLayer.ts | 65 +++--- .../testkit/OrchestratorScenario.ts | 1 + .../testkit/ProviderReplayHarness.ts | 7 +- apps/server/src/ws.ts | 3 + apps/web/src/components/ChatView.tsx | 13 -- .../src/operations/commands.test.ts | 5 + .../client-runtime/src/operations/commands.ts | 6 +- packages/contracts/src/orchestrationV2.ts | 10 + 16 files changed, 481 insertions(+), 190 deletions(-) diff --git a/apps/mobile/src/state/use-thread-outbox-drain.ts b/apps/mobile/src/state/use-thread-outbox-drain.ts index 4936731101a..706abdc1330 100644 --- a/apps/mobile/src/state/use-thread-outbox-drain.ts +++ b/apps/mobile/src/state/use-thread-outbox-drain.ts @@ -17,7 +17,10 @@ import { AsyncResult, Atom } from "effect/unstable/reactivity"; import { useCallback, useEffect, useRef, useState } from "react"; import { scopedThreadKey } from "../lib/scopedEntities"; -import { buildProjectThreadStartTurnInput } from "../lib/projectThreadStartTurn"; +import { + buildProjectThreadStartTurnInput, + deriveThreadTitleFromPrompt, +} from "../lib/projectThreadStartTurn"; import { toUploadChatImageAttachments } from "../lib/composerImages"; import { randomHex } from "../lib/uuid"; import { appAtomRegistry } from "./atom-registry"; @@ -231,6 +234,7 @@ export function useThreadOutboxDrain(): void { attachments: toUploadChatImageAttachments(queuedMessage.attachments), }, modelSelection: settings.modelSelection, + titleSeed: deriveThreadTitleFromPrompt(queuedMessage.text), runtimeMode: settings.runtimeMode, interactionMode: settings.interactionMode, createdAt: queuedMessage.createdAt, diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index 748cfdb2904..dc6768c89d8 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -90,6 +90,13 @@ export const OrchestrationEffectRequestV2 = Schema.Union([ type: Schema.Literal("attachment.cleanup"), attachmentIds: Schema.Array(Schema.String), }), + Schema.Struct({ + type: Schema.Literal("thread-title.generate"), + kind: Schema.Union([ + Schema.Struct({ type: Schema.Literal("initial"), messageId: MessageId }), + Schema.Struct({ type: Schema.Literal("regenerate") }), + ]), + }), ]); export type OrchestrationEffectRequestV2 = typeof OrchestrationEffectRequestV2.Type; @@ -99,6 +106,7 @@ export const REPLAY_SAFE_EFFECT_TYPES_AFTER_PROCESS_LOSS = [ "checkpoint.capture", "terminal.cleanup", "attachment.cleanup", + "thread-title.generate", ] as const satisfies ReadonlyArray; export const PROCESS_BOUND_EFFECT_TYPES = [ diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index 6fab8a95769..b39105f36f8 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -1,6 +1,7 @@ import { assert, it } from "@effect/vitest"; import { CommandId, + MessageId, ProviderSessionId, ProviderThreadId, ProviderTurnId, @@ -36,6 +37,7 @@ import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts"; import { ProviderTurnControlServiceV2 } from "./ProviderTurnControlService.ts"; import { ProviderTurnStartError, ProviderTurnStartServiceV2 } from "./ProviderTurnStartService.ts"; import { RuntimeRequestServiceV2 } from "./RuntimeRequestService.ts"; +import { ThreadTitleRegenerationService } from "./ThreadTitleRegenerationService.ts"; const threadId = ThreadId.make("thread:effect-worker-restart"); const oldSessionId = ProviderSessionId.make("provider-session:effect-worker-restart:old"); @@ -142,6 +144,12 @@ function makeExecutorLayer(input: { RuntimeRequestServiceV2, RuntimeRequestServiceV2.of({ respond: () => Effect.void }), ), + Layer.succeed( + ThreadTitleRegenerationService, + ThreadTitleRegenerationService.of({ + execute: ({ requestId, kind }) => record(`title:${kind.type}:${requestId}`), + }), + ), ); return executorLayer.pipe(Layer.provide(dependencies)); } @@ -714,6 +722,39 @@ it.effect("detaches a handed-off session only after the old turn terminalizes", }), ); +it.effect("executes durable thread title generation effects", () => + Effect.gen(function* () { + const now = DateTime.formatIso(yield* DateTime.now); + const events = yield* Ref.make>([]); + const commandId = CommandId.make("command:title-generation"); + const effect: OrchestrationEffectV2 = { + id: "effect:title-generation", + commandId, + threadId, + request: { + type: "thread-title.generate", + kind: { type: "initial", messageId: MessageId.make("message:title-generation") }, + }, + status: "running", + attemptCount: 1, + availableAt: now, + leaseOwner: "test-worker", + leaseExpiresAt: now, + createdAt: now, + updatedAt: now, + completedAt: null, + lastError: null, + }; + + yield* Effect.gen(function* () { + const executor = yield* OrchestrationEffectExecutorV2; + yield* executor.execute(effect); + }).pipe(Effect.provide(makeExecutorLayer({ events }))); + + assert.deepEqual(yield* Ref.get(events), [`title:initial:${commandId}`]); + }), +); + it.effect("safely retries after replacement cleanup succeeds and start fails", () => Effect.gen(function* () { const now = yield* DateTime.now; diff --git a/apps/server/src/orchestration-v2/EffectWorker.ts b/apps/server/src/orchestration-v2/EffectWorker.ts index b6d4c7c4ae2..1139d2fea3b 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.ts @@ -27,6 +27,7 @@ import { ProviderSessionManagerV2 } from "./ProviderSessionManager.ts"; import { ProviderTurnControlServiceV2 } from "./ProviderTurnControlService.ts"; import { ProviderTurnStartServiceV2 } from "./ProviderTurnStartService.ts"; import { RuntimeRequestServiceV2 } from "./RuntimeRequestService.ts"; +import { ThreadTitleRegenerationService } from "./ThreadTitleRegenerationService.ts"; export class OrchestrationEffectExecutionError extends Schema.TaggedErrorClass()( "OrchestrationEffectExecutionError", @@ -81,6 +82,7 @@ export const executorLayer: Layer.Layer< | ProviderTurnControlServiceV2 | ProviderTurnStartServiceV2 | RuntimeRequestServiceV2 + | ThreadTitleRegenerationService > = Layer.effect( OrchestrationEffectExecutorV2, Effect.gen(function* () { @@ -91,6 +93,7 @@ export const executorLayer: Layer.Layer< const providerTurnControl = yield* ProviderTurnControlServiceV2; const providerTurnStart = yield* ProviderTurnStartServiceV2; const runtimeRequests = yield* RuntimeRequestServiceV2; + const threadTitleRegeneration = yield* ThreadTitleRegenerationService; return OrchestrationEffectExecutorV2.of({ execute: (effect) => { switch (effect.request.type) { @@ -290,6 +293,23 @@ export const executorLayer: Layer.Layer< }), ), ); + case "thread-title.generate": + return threadTitleRegeneration + .execute({ + threadId: effect.threadId, + requestId: effect.commandId, + kind: effect.request.kind, + }) + .pipe( + Effect.mapError( + (cause) => + new OrchestrationEffectExecutionError({ + effectId: effect.id, + effectType: effect.request.type, + cause, + }), + ), + ); } }, }); diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 55eb2ecc8b3..80fb9e87fce 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -1,6 +1,7 @@ import { type ChatAttachment, CommandId, + type MessageId, type ModelSelection, OrchestrationV2Command, type OrchestrationV2AppThread, @@ -196,6 +197,7 @@ function commandThreadId(command: OrchestrationV2Command): ThreadId { case "thread.visit": case "thread.mark-unread": case "thread.metadata.update": + case "thread.title.regeneration.complete": case "thread.runtime-mode.set": case "thread.interaction-mode.set": case "thread.model-selection.set": @@ -223,6 +225,21 @@ function commandThreadId(command: OrchestrationV2Command): ThreadId { } } +function pendingThreadTitleGenerationEffect( + commandId: CommandId, + threadId: ThreadId, + kind: + | { readonly type: "initial"; readonly messageId: MessageId } + | { readonly type: "regenerate" }, +): PendingOrchestrationEffectV2 { + return { + id: `effect:${commandId}:thread-title.generate`, + commandId, + threadId, + request: { type: "thread-title.generate", kind }, + }; +} + function nextTurnItemOrdinal(projection: OrchestrationV2ThreadProjection): number { return Math.max(0, ...projection.turnItems.map((item) => item.ordinal)) + 1; } @@ -943,6 +960,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio | "thread.visit" | "thread.mark-unread" | "thread.metadata.update" + | "thread.title.regeneration.complete" | "thread.runtime-mode.set" | "thread.interaction-mode.set" | "thread.model-selection.set" @@ -1100,11 +1118,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio const updatedThread: OrchestrationV2AppThread = (() => { switch (command.type) { case "thread.archive": - return { ...thread, archivedAt: now, updatedAt: now }; + return { ...thread, archivedAt: now, titleRegeneration: null, updatedAt: now }; case "thread.unarchive": return { ...thread, archivedAt: null, updatedAt: now }; case "thread.delete": - return { ...thread, deletedAt: thread.deletedAt ?? now, updatedAt: now }; + return { + ...thread, + deletedAt: thread.deletedAt ?? now, + titleRegeneration: null, + updatedAt: now, + }; case "thread.settle": { const alreadySettled = thread.settledOverride === "settled" && thread.settledAt !== null; return { @@ -1173,6 +1196,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio : {}), updatedAt: now, }; + case "thread.title.regeneration.complete": + return thread.titleRegeneration?.requestId === command.requestId + ? { + ...thread, + ...(command.title === undefined ? {} : { title: command.title }), + titleRegeneration: null, + updatedAt: now, + } + : thread; case "thread.runtime-mode.set": return { ...thread, runtimeMode: command.runtimeMode, updatedAt: now }; case "thread.interaction-mode.set": @@ -1208,6 +1240,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "thread.mark-unread": return "thread.marked-unread" as const; case "thread.metadata.update": + case "thread.title.regeneration.complete": return "thread.metadata-updated" as const; case "thread.runtime-mode.set": return "thread.runtime-mode-updated" as const; @@ -1230,6 +1263,15 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio payload: updatedThread, }); + if (command.type === "thread.metadata.update" && command.regenerateTitle === true) { + yield* Ref.update(effects, (existing) => [ + ...existing, + pendingThreadTitleGenerationEffect(command.commandId, command.threadId, { + type: "regenerate", + }), + ]); + } + if (command.type === "thread.archive" || command.type === "thread.delete") { const emitEvent = emit(events, command); const activeRunIds = new Set( @@ -2343,6 +2385,33 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio }); projection = yield* getProjectionWithPendingEvents(command.threadId, events); } + if (command.titleSeed !== undefined && projection.messages.length === 0) { + const now = yield* DateTime.now; + const thread: OrchestrationV2AppThread = { + ...projection.thread, + title: command.titleSeed, + titleRegeneration: { requestId: command.commandId, startedAt: now }, + updatedAt: now, + }; + yield* emit( + events, + command, + )({ + type: "thread.metadata-updated", + threadId: command.threadId, + providerInstanceId: thread.providerInstanceId, + occurredAt: now, + payload: thread, + }); + yield* Ref.update(effects, (existing) => [ + ...existing, + pendingThreadTitleGenerationEffect(command.commandId, command.threadId, { + type: "initial", + messageId: command.messageId, + }), + ]); + projection = yield* getProjectionWithPendingEvents(command.threadId, events); + } const modelSelection = command.modelSelection ?? projection.thread.modelSelection; const dispatchMode = command.dispatchMode; const sourcePlanProjection = @@ -5702,6 +5771,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio case "thread.visit": case "thread.mark-unread": case "thread.metadata.update": + case "thread.title.regeneration.complete": case "thread.runtime-mode.set": case "thread.interaction-mode.set": case "thread.model-selection.set": diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index 21e35189838..ef9c057b2c4 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -10,6 +10,7 @@ import { ThreadId, } from "@t3tools/contracts"; import * as Deferred from "effect/Deferred"; +import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -17,9 +18,11 @@ import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; import * as SqlClient from "effect/unstable/sql/SqlClient"; +import * as TestClock from "effect/testing/TestClock"; import * as GitWorkflow from "../git/GitWorkflowService.ts"; import { SqlitePersistenceMemory } from "../persistence/Layers/Sqlite.ts"; +import { ProjectionProjectRepository } from "../persistence/Services/ProjectionProjects.ts"; import * as ProjectService from "../project/ProjectService.ts"; import * as ProjectSetupScriptRunner from "../project/ProjectSetupScriptRunner.ts"; import { makeProviderRegistryLayer } from "../provider/testUtils/providerRegistryMock.ts"; @@ -33,6 +36,7 @@ import type { ProviderAdapterV2Shape } from "./ProviderAdapter.ts"; import * as ProviderAdapterRegistry from "./ProviderAdapterRegistry.ts"; import * as ThreadLaunch from "./ThreadLaunchService.ts"; import * as ThreadManagement from "./ThreadManagementService.ts"; +import * as ThreadTitleRegeneration from "./ThreadTitleRegenerationService.ts"; import { makeOrchestratorV2ReplayLayerWithRegistry } from "./testkit/ProviderReplayHarness.ts"; const projectId = ProjectId.make("project:launch-test"); @@ -94,6 +98,9 @@ function makeHarness(options: HarnessOptions = {}) { const generateBranchName = vi.fn( options.generateBranchName ?? (() => Effect.succeed({ branch: "generated-branch" })), ); + const generateThreadTitle = vi.fn( + options.generateTitle ?? (() => Effect.succeed({ title: "Generated title" })), + ); const externalServices = Layer.mergeAll( Layer.succeed(ProjectService.ProjectService, { create: () => Effect.die("unused"), @@ -115,8 +122,7 @@ function makeHarness(options: HarnessOptions = {}) { runForThread: runSetup, }), Layer.mock(TextGeneration.TextGeneration)({ - generateThreadTitle: - options.generateTitle ?? (() => Effect.succeed({ title: "Generated title" })), + generateThreadTitle, generateBranchName, }), ServerSettings.layerTest(options.serverSettings), @@ -125,10 +131,31 @@ function makeHarness(options: HarnessOptions = {}) { const launch = ThreadLaunch.layer.pipe( Layer.provide(Layer.mergeAll(externalServices, threadManagement, receipts, IdAllocator.layer)), ); + const projectedProjects = Layer.mock(ProjectionProjectRepository)({ + getById: ({ projectId: requestedProjectId }) => + Effect.succeed( + requestedProjectId === projectId + ? Option.some({ + projectId, + title: project.title, + workspaceRoot: project.workspaceRoot, + defaultModelSelection: project.defaultModelSelection, + scripts: project.scripts, + createdAt: project.createdAt, + updatedAt: project.updatedAt, + deletedAt: project.deletedAt, + }) + : Option.none(), + ), + }); + const titleRegeneration = ThreadTitleRegeneration.layer.pipe( + Layer.provide(Layer.mergeAll(threadManagement, projectedProjects, externalServices)), + ); return { - layer: Layer.mergeAll(launch, threadManagement, outbox, database), + layer: Layer.mergeAll(launch, threadManagement, titleRegeneration, outbox, database), createWorktree, generateBranchName, + generateThreadTitle, runSetup, }; } @@ -412,45 +439,140 @@ it.effect( }), ); -it.effect("does not put optional title generation on the provisioning critical path", () => +it.effect("arms durable title generation after accepting the first message", () => Effect.gen(function* () { - const titleStarted = yield* Deferred.make(); - const allowTitle = yield* Deferred.make(); - const setupEntered = yield* Deferred.make(); const harness = makeHarness({ - generateTitle: () => - Deferred.succeed(titleStarted, undefined).pipe( - Effect.andThen(Deferred.await(allowTitle)), - Effect.as({ title: "Generated later" }), - ), - runSetup: () => - Deferred.succeed(setupEntered, undefined).pipe(Effect.as({ status: "no-script" as const })), + generateTitle: (input) => + Effect.succeed({ + title: input.previousTitle === undefined ? "Generated title" : "Regenerated title", + }), }); yield* Effect.gen(function* () { const launches = yield* ThreadLaunch.ThreadLaunchService; const threads = yield* ThreadManagement.ThreadManagementService; - const input = launchInput({ - command: "command:launch:title-independent", - thread: "thread:launch:title-independent", - message: "Generate my title slowly", - }); + const outbox = yield* EffectOutbox.EffectOutboxV2; + const titleRegeneration = yield* ThreadTitleRegeneration.ThreadTitleRegenerationService; + const input = { + ...launchInput({ + command: "command:launch:title-generation", + thread: "thread:launch:title-generation", + message: "Generate my title", + }), + title: "Generate my title", + generateTitle: true, + }; const launched = yield* launches.launch(input); - yield* Deferred.await(titleStarted); - yield* Deferred.await(setupEntered); - yield* waitUntil(() => - threads - .getThreadProjection(launched.threadId) - .pipe(Effect.map((projection) => projection.runs[0]?.status === "starting")), + const generationCommandId = CommandId.make("command:launch:title-generation:initial-message"); + + const projection = yield* threads.getThreadProjection(launched.threadId); + assert.equal(projection.thread.title, "Generate my title"); + assert.equal(projection.thread.titleRegeneration?.requestId, generationCommandId); + assert.deepEqual( + (yield* outbox.listByCommandId(generationCommandId)).map((effect) => effect.request), + [ + { + type: "thread-title.generate", + kind: { type: "initial", messageId: MessageId.make("Generate my title:id") }, + }, + ], + ); + yield* titleRegeneration.execute({ + threadId: launched.threadId, + requestId: generationCommandId, + kind: { type: "initial", messageId: MessageId.make("Generate my title:id") }, + }); + const generated = yield* threads.getThreadProjection(launched.threadId); + assert.equal(generated.thread.title, "Generated title"); + assert.deepEqual( + harness.generateThreadTitle.mock.calls[0]?.[0].modelSelection, + DEFAULT_SERVER_SETTINGS.textGenerationModelSelection, ); + + const manualRequestId = CommandId.make("command:title-generation:manual"); + yield* threads.dispatch({ + type: "thread.metadata.update", + commandId: manualRequestId, + threadId: launched.threadId, + regenerateTitle: true, + }); + assert.deepEqual( + (yield* outbox.listByCommandId(manualRequestId)).map((effect) => effect.request), + [{ type: "thread-title.generate", kind: { type: "regenerate" } }], + ); + yield* titleRegeneration.execute({ + threadId: launched.threadId, + requestId: manualRequestId, + kind: { type: "regenerate" }, + }); + const regenerated = yield* threads.getThreadProjection(launched.threadId); + assert.equal(regenerated.thread.title, "Regenerated title"); + assert.equal(harness.generateThreadTitle.mock.calls[1]?.[0].previousTitle, "Generated title"); + + yield* threads.dispatch({ + type: "thread.metadata.update", + commandId: CommandId.make("command:title-generation:user-rename"), + threadId: launched.threadId, + title: "Keep my title", + }); + const renamed = yield* threads.getThreadProjection(launched.threadId); + yield* TestClock.adjust(Duration.seconds(1)); + yield* threads.dispatch({ + type: "thread.title.regeneration.complete", + commandId: CommandId.make("command:title-generation:stale-completion"), + threadId: launched.threadId, + requestId: generationCommandId, + title: "Stale generated title", + }); + const afterStaleCompletion = yield* threads.getThreadProjection(launched.threadId); + assert.equal(afterStaleCompletion.thread.title, "Keep my title"); assert.equal( - (yield* threads.getThreadProjection(launched.threadId)).thread.title, - "New thread", + DateTime.toEpochMillis(afterStaleCompletion.thread.updatedAt), + DateTime.toEpochMillis(renamed.thread.updatedAt), ); - yield* Deferred.succeed(allowTitle, undefined); - yield* waitUntil(() => - threads - .getThreadProjection(launched.threadId) - .pipe(Effect.map((projection) => projection.thread.title === "Generated later")), + }).pipe(Effect.provide(harness.layer)); + }), +); + +it.effect("generates an initial title for an attachment-only message", () => + Effect.gen(function* () { + const harness = makeHarness(); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const titleRegeneration = yield* ThreadTitleRegeneration.ThreadTitleRegenerationService; + const messageId = MessageId.make("message:image-only"); + const input = { + ...launchInput({ + command: "command:launch:image-only", + thread: "thread:launch:image-only", + }), + title: "Image: screenshot.png", + generateTitle: true, + initialMessage: { + messageId, + text: "", + attachments: [ + { + type: "image" as const, + id: "attachment-image-only", + name: "screenshot.png", + mimeType: "image/png", + sizeBytes: 128, + }, + ], + }, + }; + + const launched = yield* launches.launch(input); + yield* titleRegeneration.execute({ + threadId: launched.threadId, + requestId: CommandId.make("command:launch:image-only:initial-message"), + kind: { type: "initial", messageId }, + }); + + assert.equal(harness.generateThreadTitle.mock.calls[0]?.[0].message, ""); + assert.equal( + harness.generateThreadTitle.mock.calls[0]?.[0].attachments?.[0]?.name, + "screenshot.png", ); }).pipe(Effect.provide(harness.layer)); }), diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index ae67feec50a..600fe0b1df0 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -59,6 +59,7 @@ export interface ThreadLaunchInput { readonly reuseExistingThread?: boolean; readonly projectId: ProjectId; readonly title: string; + readonly generateTitle?: boolean; readonly modelSelection: ModelSelection; readonly runtimeMode: RuntimeMode; readonly interactionMode: ProviderInteractionMode; @@ -198,34 +199,6 @@ export const make = Effect.gen(function* () { ), ); - if (input.title === "New thread" && input.initialMessage !== undefined) { - yield* textGeneration - .generateThreadTitle({ - cwd: project.workspaceRoot, - message: input.initialMessage.text, - attachments: input.initialMessage.attachments, - modelSelection: input.modelSelection, - }) - .pipe( - Effect.flatMap((result) => - threads.dispatch({ - type: "thread.metadata.update", - commandId: CommandId.make(`${input.commandId}:title`), - threadId, - title: result.title, - }), - ), - Effect.catchCause((cause) => - Effect.logWarning("Thread title generation failed", { - commandId: input.commandId, - threadId, - cause, - }), - ), - Effect.forkIn(preparationScope), - ); - } - const initialMessage = input.initialMessage; let branch = input.workspaceStrategy.type === "worktree" && @@ -452,6 +425,7 @@ export const make = Effect.gen(function* () { type: "thread.metadata.update", commandId: input.commandId, threadId: candidateThreadId, + title: input.title, }) : threads.dispatch({ type: "thread.create", @@ -502,6 +476,7 @@ export const make = Effect.gen(function* () { messageId, text: input.initialMessage.text, attachments: input.initialMessage.attachments, + ...(input.generateTitle === true ? { titleSeed: input.title } : {}), modelSelection: input.modelSelection, dispatchMode: { type: "defer_start" }, createdBy: input.createdBy, diff --git a/apps/server/src/orchestration-v2/ThreadTitleRegenerationService.ts b/apps/server/src/orchestration-v2/ThreadTitleRegenerationService.ts index 98476f7bc51..9abd3aa63da 100644 --- a/apps/server/src/orchestration-v2/ThreadTitleRegenerationService.ts +++ b/apps/server/src/orchestration-v2/ThreadTitleRegenerationService.ts @@ -1,11 +1,21 @@ -import { type ChatAttachment, CommandId, type ThreadId } from "@t3tools/contracts"; +import { + type ChatAttachment, + CommandId, + type MessageId, + type ServerSettingsError, + type ThreadId, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; -import * as Stream from "effect/Stream"; -import * as ProjectService from "../project/ProjectService.ts"; +import type { ProjectionRepositoryError } from "../persistence/Errors.ts"; +import { ProjectionProjectRepository } from "../persistence/Services/ProjectionProjects.ts"; +import * as ServerSettings from "../serverSettings.ts"; import * as TextGeneration from "../textGeneration/TextGeneration.ts"; +import type { OrchestratorV2Error } from "./Orchestrator.ts"; import { ThreadManagementService } from "./ThreadManagementService.ts"; const MAX_REGENERATION_ATTACHMENTS = 4; @@ -69,92 +79,117 @@ export function formatThreadTitleContext( }; } -/** - * Reacts to thread.metadata-updated events that arm the titleRegeneration - * marker: generates a title from the thread's conversation and lands it via a - * follow-up metadata update (which clears the marker), or clears the marker - * explicitly when generation fails. Event-driven so every dispatch surface - * (ws, MCP, mobile) gets the same behavior. - */ -export const workerLive = Layer.effectDiscard( +export class ThreadTitleRegenerationService extends Context.Service< + ThreadTitleRegenerationService, + { + readonly execute: (input: { + readonly threadId: ThreadId; + readonly requestId: CommandId; + readonly kind: + | { readonly type: "initial"; readonly messageId: MessageId } + | { readonly type: "regenerate" }; + }) => Effect.Effect< + void, + OrchestratorV2Error | ProjectionRepositoryError | ServerSettingsError + >; + } +>()("t3/orchestration-v2/ThreadTitleRegenerationService") {} + +export const layer = Layer.effect( + ThreadTitleRegenerationService, Effect.gen(function* () { const threads = yield* ThreadManagementService; - const projects = yield* ProjectService.ProjectService; + const projects = yield* ProjectionProjectRepository; + const serverSettings = yield* ServerSettings.ServerSettingsService; const textGeneration = yield* TextGeneration.TextGeneration; - const handledRequestIds = new Set(); - const regenerate = Effect.fn("ThreadTitleRegenerationService.regenerate")(function* ( - threadId: ThreadId, - requestId: CommandId, - ) { - const projection = yield* threads.getThreadProjection(threadId); - if (projection.thread.titleRegeneration?.requestId !== requestId) { - return; - } - const project = yield* projects.getById(projection.thread.projectId); - if (Option.isNone(project)) { - return; - } - const context = formatThreadTitleContext( - projection.messages.filter((message) => !message.streaming), - ); - if (context.message.length === 0) { - yield* threads.dispatch({ - type: "thread.metadata.update", - commandId: CommandId.make(`${requestId}:title-clear`), - threadId, - regenerateTitle: false, + const complete = (input: { + readonly threadId: ThreadId; + readonly requestId: CommandId; + readonly title?: string; + }) => + threads + .dispatch({ + type: "thread.title.regeneration.complete", + commandId: CommandId.make(`${input.requestId}:title-complete`), + threadId: input.threadId, + requestId: input.requestId, + ...(input.title === undefined ? {} : { title: input.title }), + }) + .pipe(Effect.asVoid); + + const execute: ThreadTitleRegenerationService["Service"]["execute"] = Effect.fn( + "ThreadTitleRegenerationService.execute", + )(function* (input) { + const outcome: + | { readonly type: "stale" } + | { readonly type: "complete"; readonly title?: string } = yield* Effect.gen(function* () { + const projection = yield* threads.getThreadProjection(input.threadId); + if (projection.thread.titleRegeneration?.requestId !== input.requestId) { + return { type: "stale" as const }; + } + + const project = yield* projects.getById({ projectId: projection.thread.projectId }); + if (Option.isNone(project)) { + return { type: "complete" as const }; + } + + let context: { + readonly message: string; + readonly attachments: ReadonlyArray; + }; + if (input.kind.type === "initial") { + const messageId = input.kind.messageId; + const message = projection.messages.find( + (candidate) => candidate.id === messageId && !candidate.streaming, + ); + context = + message === undefined + ? { message: "", attachments: [] } + : { message: message.text, attachments: message.attachments }; + } else { + context = formatThreadTitleContext( + projection.messages.filter((message) => !message.streaming), + ); + } + if (context.message.length === 0 && context.attachments.length === 0) { + return { type: "complete" as const }; + } + + const settings = yield* serverSettings.getSettings; + const result = yield* textGeneration.generateThreadTitle({ + cwd: projection.thread.worktreePath ?? project.value.workspaceRoot, + message: context.message, + attachments: context.attachments, + ...(input.kind.type === "regenerate" ? { previousTitle: projection.thread.title } : {}), + modelSelection: settings.textGenerationModelSelection, }); + const generatedTitle = result.title.trim(); + return generatedTitle === "New thread" || + (input.kind.type === "regenerate" && generatedTitle === projection.thread.title.trim()) + ? { type: "complete" as const } + : { type: "complete" as const, title: result.title }; + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.logWarning("Thread title generation failed", { + threadId: input.threadId, + requestId: input.requestId, + cause, + }).pipe(Effect.as({ type: "complete" as const })), + ), + ); + + if (outcome.type === "stale") { return; } - const result = yield* textGeneration.generateThreadTitle({ - cwd: projection.thread.worktreePath ?? Option.getOrThrow(project).workspaceRoot, - message: context.message, - attachments: context.attachments, - modelSelection: projection.thread.modelSelection, - }); - yield* threads.dispatch({ - type: "thread.metadata.update", - commandId: CommandId.make(`${requestId}:title`), - threadId, - title: result.title, + yield* complete({ + ...input, + ...(outcome.title === undefined ? {} : { title: outcome.title }), }); }); - yield* threads.streamDomainEvents.pipe( - Stream.runForEach((event) => { - if (event.type !== "thread.metadata-updated") { - return Effect.void; - } - const marker = event.payload.titleRegeneration; - if (marker === null || marker === undefined || handledRequestIds.has(marker.requestId)) { - return Effect.void; - } - handledRequestIds.add(marker.requestId); - return regenerate(event.threadId, marker.requestId).pipe( - Effect.catchCause((cause) => - Effect.logWarning("Thread title regeneration failed", { - threadId: event.threadId, - requestId: marker.requestId, - cause, - }).pipe( - Effect.andThen( - threads - .dispatch({ - type: "thread.metadata.update", - commandId: CommandId.make(`${marker.requestId}:title-clear`), - threadId: event.threadId, - regenerateTitle: false, - }) - .pipe(Effect.ignore), - ), - ), - ), - Effect.forkDetach, - Effect.asVoid, - ); - }), - Effect.forkScoped, - ); + return ThreadTitleRegenerationService.of({ execute }); }), ); diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 38aa57b570b..6f4df38e1d9 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -4,6 +4,7 @@ import { OrchestrationLayerLive, } from "../orchestration/runtimeLayer.ts"; import { ProjectionProjectRepositoryLive } from "../persistence/Layers/ProjectionProjects.ts"; +import * as TextGeneration from "../textGeneration/TextGeneration.ts"; import { layer as projectServiceLayer } from "../project/ProjectService.ts"; import { layer as projectSetupScriptRunnerLayer } from "../project/ProjectSetupScriptRunner.ts"; import { layer as checkpointCaptureServiceLayer } from "./CheckpointCaptureService.ts"; @@ -27,7 +28,7 @@ import { layer as projectionMaintenanceLayer } from "./ProjectionMaintenance.ts" import { layerFromProviderInstanceRegistry as providerAdapterRegistryLayerFromProviderInstances } from "./ProviderAdapterRegistry.ts"; import { layer as providerContinuationRequestsLayer } from "./ProviderContinuationRequests.ts"; import { workerLive as providerContinuationWorkerLive } from "./ProviderContinuationService.ts"; -import { workerLive as threadTitleRegenerationWorkerLive } from "./ThreadTitleRegenerationService.ts"; +import { layer as threadTitleRegenerationServiceLayer } from "./ThreadTitleRegenerationService.ts"; import { layer as providerEventIngestorLayer } from "./ProviderEventIngestor.ts"; import { layer as providerSessionManagerLayer } from "./ProviderSessionManager.ts"; import { layer as providerRuntimeRecoveryLayer } from "./ProviderRuntimeRecoveryService.ts"; @@ -157,33 +158,6 @@ const runFinalizationServiceProvided = runFinalizationServiceLayer.pipe( Layer.provide(Layer.merge(checkpointCaptureServiceProvided, projectionStoreLayer)), ); -const effectExecutorProvided = effectExecutorLayer.pipe( - Layer.provide( - Layer.mergeAll( - runFinalizationServiceProvided, - checkpointRollbackServiceProvided, - providerSessionManagerProvided, - providerTurnControlServiceProvided, - providerTurnStartServiceProvided, - runtimeRequestServiceProvided, - ), - ), -); -const effectWorkerProvided = effectWorkerLayer.pipe( - Layer.provide(Layer.merge(storesLayer, effectExecutorProvided)), -); -const providerRuntimeRecoveryProvided = providerRuntimeRecoveryLayer.pipe( - Layer.provide( - Layer.mergeAll( - effectWorkerProvided, - storesLayer, - eventSinkProvided, - idAllocatorLayer, - projectionStoreLayer, - ), - ), -); - const orchestratorProvided = orchestratorLayer.pipe( Layer.provide( Layer.mergeAll( @@ -191,7 +165,6 @@ const orchestratorProvided = orchestratorLayer.pipe( commandPolicyLayer, storesLayer, eventSinkProvided, - effectWorkerProvided, commandReceiptStoreProvided, contextHandoffServiceProvided, idAllocatorLayer, @@ -237,8 +210,37 @@ const providerContinuationWorkerProvided = providerContinuationWorkerLive.pipe( Layer.mergeAll(providerContinuationRequestsLayer, threadManagementProvided, idAllocatorLayer), ), ); -const threadTitleRegenerationWorkerProvided = threadTitleRegenerationWorkerLive.pipe( - Layer.provide(Layer.merge(threadManagementProvided, ProjectServiceLayerLive)), +const threadTitleRegenerationProvided = threadTitleRegenerationServiceLayer.pipe( + Layer.provide( + Layer.mergeAll(threadManagementProvided, ProjectionProjectRepositoryLive, TextGeneration.layer), + ), +); +const effectExecutorProvided = effectExecutorLayer.pipe( + Layer.provide( + Layer.mergeAll( + runFinalizationServiceProvided, + checkpointRollbackServiceProvided, + providerSessionManagerProvided, + providerTurnControlServiceProvided, + providerTurnStartServiceProvided, + runtimeRequestServiceProvided, + threadTitleRegenerationProvided, + ), + ), +); +const effectWorkerProvided = effectWorkerLayer.pipe( + Layer.provide(Layer.merge(storesLayer, effectExecutorProvided)), +); +const providerRuntimeRecoveryProvided = providerRuntimeRecoveryLayer.pipe( + Layer.provide( + Layer.mergeAll( + effectWorkerProvided, + storesLayer, + eventSinkProvided, + idAllocatorLayer, + projectionStoreLayer, + ), + ), ); export const OrchestrationV2LayerLive = Layer.mergeAll( @@ -254,7 +256,6 @@ export const OrchestrationV2LayerLive = Layer.mergeAll( export const OrchestrationV2ProductionLayerLive = Layer.mergeAll( OrchestrationLayerLive, OrchestrationV2LayerLive, - threadTitleRegenerationWorkerProvided, ProjectServiceLayerLive, threadLaunchProvided, threadLifecycleProvided, diff --git a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts index 8b1cc076678..5c574d86da0 100644 --- a/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts +++ b/apps/server/src/orchestration-v2/testkit/OrchestratorScenario.ts @@ -108,6 +108,7 @@ function commandThreadIds(command: OrchestrationV2Command): ReadonlyArray( const runFinalizationServiceProvided = runFinalizationServiceLayer.pipe( Layer.provide(Layer.merge(checkpointCaptureServiceProvided, storesLayer)), ); + const threadTitleRegenerationTestLayer = Layer.succeed( + ThreadTitleRegenerationService, + ThreadTitleRegenerationService.of({ execute: () => Effect.void }), + ); const effectExecutorProvided = effectExecutorLayer.pipe( Layer.provide( Layer.mergeAll( @@ -343,6 +348,7 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( providerTurnControlServiceProvided, providerTurnStartServiceProvided, runtimeRequestServiceProvided, + threadTitleRegenerationTestLayer, ), ), ); @@ -355,7 +361,6 @@ export function makeOrchestratorV2ReplayLayerWithRegistry( checkpointServiceProvided, commandPolicyLayer, contextHandoffServiceProvided, - effectWorkerProvided, persistenceLayer, registryLayer, runtimeLayer, diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 813492f10ea..7fa1a018fec 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -1138,6 +1138,9 @@ const makeWsRpcLayer = ( : { reuseExistingThread: input.reuseExistingThread }), projectId: input.projectId, title: input.title, + ...(input.generateTitle === undefined + ? {} + : { generateTitle: input.generateTitle }), modelSelection: input.modelSelection, runtimeMode: input.runtimeMode, interactionMode: input.interactionMode, diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index f108a56dfda..05ba2853bb1 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -5184,19 +5184,6 @@ function ChatViewContent(props: ChatViewProps) { ); let failure: AtomCommandResult | null = null; - // Auto-title from first message - if (isFirstMessage && isServerThread) { - const titleResult = await updateThreadMetadata({ - environmentId, - input: { - threadId: threadIdForSend, - title, - }, - }); - if (titleResult._tag === "Failure") { - failure = titleResult; - } - } if (failure === null && isServerThread) { const settingsResult = await persistThreadSettingsForNextTurn({ diff --git a/packages/client-runtime/src/operations/commands.test.ts b/packages/client-runtime/src/operations/commands.test.ts index cac8b3f39dc..0e52e444107 100644 --- a/packages/client-runtime/src/operations/commands.test.ts +++ b/packages/client-runtime/src/operations/commands.test.ts @@ -231,6 +231,7 @@ describe("V2 environment commands", () => { }, runtimeMode: "full-access", interactionMode: "default", + titleSeed: "Implement the plan", sourceProposedPlan: { threadId: ThreadId.make("thread-plan"), planId: PlanId.make("plan-1"), @@ -242,6 +243,7 @@ describe("V2 environment commands", () => { type: "message.dispatch", commandId: "implement-plan", threadId: v2ThreadId, + titleSeed: "Implement the plan", sourcePlanRef: { threadId: "thread-plan", planId: "plan-1" }, dispatchMode: { type: "start_immediately" }, }); @@ -264,6 +266,7 @@ describe("V2 environment commands", () => { }, runtimeMode: "full-access", interactionMode: "default", + titleSeed: "Continue here", bootstrap: { createThread: { projectId: ProjectId.make("project-1"), @@ -280,6 +283,8 @@ describe("V2 environment commands", () => { expect(launches[0]).toMatchObject({ threadId: v2ThreadId, + title: "Continue here", + generateTitle: true, workspaceStrategy: { type: "existing_worktree", worktreePath: "/workspace/project-worktrees/feature", diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index cd801766abd..b652025c13e 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -528,7 +528,8 @@ export const startThreadTurn = Effect.fn("EnvironmentCommands.startThreadTurn")( threadId: input.threadId, ...(bootstrap === undefined ? { reuseExistingThread: true } : {}), projectId: thread.projectId, - title: thread.title, + title: input.titleSeed ?? thread.title, + generateTitle: input.titleSeed !== undefined, modelSelection: input.modelSelection ?? thread.modelSelection, runtimeMode: input.runtimeMode, interactionMode: input.interactionMode, @@ -587,6 +588,9 @@ export const startThreadTurn = Effect.fn("EnvironmentCommands.startThreadTurn")( messageId: input.message.messageId, text: input.message.text, attachments, + ...(input.titleSeed === undefined || projection.messages.length > 0 + ? {} + : { titleSeed: input.titleSeed }), ...(input.modelSelection === undefined ? {} : { modelSelection: input.modelSelection }), ...(input.sourceProposedPlan === undefined ? {} : { sourcePlanRef: input.sourceProposedPlan }), dispatchMode, diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index 700160ad8c1..15edce1faf1 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -1939,6 +1939,13 @@ export const OrchestrationV2Command = Schema.Union([ worktreePath: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), expectedWorktreePath: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), }), + Schema.Struct({ + type: Schema.Literal("thread.title.regeneration.complete"), + commandId: CommandId, + threadId: ThreadId, + requestId: CommandId, + title: Schema.optional(TrimmedNonEmptyString), + }), Schema.Struct({ type: Schema.Literal("thread.runtime-mode.set"), commandId: CommandId, @@ -1972,6 +1979,8 @@ export const OrchestrationV2Command = Schema.Union([ messageId: MessageId, text: Schema.String, attachments: Schema.Array(ChatAttachment), + /** Seed the temporary title and generate a durable replacement for the first message. */ + titleSeed: Schema.optional(TrimmedNonEmptyString), modelSelection: Schema.optional(ModelSelection), sourcePlanRef: Schema.optional(Schema.Struct({ threadId: ThreadId, planId: PlanId })), dispatchMode: Schema.Union([ @@ -2179,6 +2188,7 @@ export const OrchestrationV2ThreadLaunchInput = Schema.Struct({ reuseExistingThread: Schema.optional(Schema.Boolean), projectId: ProjectId, title: TrimmedNonEmptyString, + generateTitle: Schema.optional(Schema.Boolean), modelSelection: ModelSelection, runtimeMode: RuntimeMode, interactionMode: ProviderInteractionMode, From 8b7c3357d7d0803693826518366d5da417d04586 Mon Sep 17 00:00:00 2001 From: Yusuf007R Date: Sat, 1 Aug 2026 02:32:07 -0500 Subject: [PATCH 2/4] refactor(server): expose title regeneration service make --- .../ThreadTitleRegenerationService.ts | 185 +++++++++--------- 1 file changed, 92 insertions(+), 93 deletions(-) diff --git a/apps/server/src/orchestration-v2/ThreadTitleRegenerationService.ts b/apps/server/src/orchestration-v2/ThreadTitleRegenerationService.ts index 9abd3aa63da..b955c160cee 100644 --- a/apps/server/src/orchestration-v2/ThreadTitleRegenerationService.ts +++ b/apps/server/src/orchestration-v2/ThreadTitleRegenerationService.ts @@ -95,101 +95,100 @@ export class ThreadTitleRegenerationService extends Context.Service< } >()("t3/orchestration-v2/ThreadTitleRegenerationService") {} -export const layer = Layer.effect( - ThreadTitleRegenerationService, - Effect.gen(function* () { - const threads = yield* ThreadManagementService; - const projects = yield* ProjectionProjectRepository; - const serverSettings = yield* ServerSettings.ServerSettingsService; - const textGeneration = yield* TextGeneration.TextGeneration; +export const make = Effect.gen(function* () { + const threads = yield* ThreadManagementService; + const projects = yield* ProjectionProjectRepository; + const serverSettings = yield* ServerSettings.ServerSettingsService; + const textGeneration = yield* TextGeneration.TextGeneration; - const complete = (input: { - readonly threadId: ThreadId; - readonly requestId: CommandId; - readonly title?: string; - }) => - threads - .dispatch({ - type: "thread.title.regeneration.complete", - commandId: CommandId.make(`${input.requestId}:title-complete`), - threadId: input.threadId, - requestId: input.requestId, - ...(input.title === undefined ? {} : { title: input.title }), - }) - .pipe(Effect.asVoid); - - const execute: ThreadTitleRegenerationService["Service"]["execute"] = Effect.fn( - "ThreadTitleRegenerationService.execute", - )(function* (input) { - const outcome: - | { readonly type: "stale" } - | { readonly type: "complete"; readonly title?: string } = yield* Effect.gen(function* () { - const projection = yield* threads.getThreadProjection(input.threadId); - if (projection.thread.titleRegeneration?.requestId !== input.requestId) { - return { type: "stale" as const }; - } - - const project = yield* projects.getById({ projectId: projection.thread.projectId }); - if (Option.isNone(project)) { - return { type: "complete" as const }; - } - - let context: { - readonly message: string; - readonly attachments: ReadonlyArray; - }; - if (input.kind.type === "initial") { - const messageId = input.kind.messageId; - const message = projection.messages.find( - (candidate) => candidate.id === messageId && !candidate.streaming, - ); - context = - message === undefined - ? { message: "", attachments: [] } - : { message: message.text, attachments: message.attachments }; - } else { - context = formatThreadTitleContext( - projection.messages.filter((message) => !message.streaming), - ); - } - if (context.message.length === 0 && context.attachments.length === 0) { - return { type: "complete" as const }; - } - - const settings = yield* serverSettings.getSettings; - const result = yield* textGeneration.generateThreadTitle({ - cwd: projection.thread.worktreePath ?? project.value.workspaceRoot, - message: context.message, - attachments: context.attachments, - ...(input.kind.type === "regenerate" ? { previousTitle: projection.thread.title } : {}), - modelSelection: settings.textGenerationModelSelection, - }); - const generatedTitle = result.title.trim(); - return generatedTitle === "New thread" || - (input.kind.type === "regenerate" && generatedTitle === projection.thread.title.trim()) - ? { type: "complete" as const } - : { type: "complete" as const, title: result.title }; - }).pipe( - Effect.catchCause((cause) => - Cause.hasInterruptsOnly(cause) - ? Effect.interrupt - : Effect.logWarning("Thread title generation failed", { - threadId: input.threadId, - requestId: input.requestId, - cause, - }).pipe(Effect.as({ type: "complete" as const })), - ), - ); - - if (outcome.type === "stale") { - return; + const complete = (input: { + readonly threadId: ThreadId; + readonly requestId: CommandId; + readonly title?: string; + }) => + threads + .dispatch({ + type: "thread.title.regeneration.complete", + commandId: CommandId.make(`${input.requestId}:title-complete`), + threadId: input.threadId, + requestId: input.requestId, + ...(input.title === undefined ? {} : { title: input.title }), + }) + .pipe(Effect.asVoid); + + const execute: ThreadTitleRegenerationService["Service"]["execute"] = Effect.fn( + "ThreadTitleRegenerationService.execute", + )(function* (input) { + const outcome: + | { readonly type: "stale" } + | { readonly type: "complete"; readonly title?: string } = yield* Effect.gen(function* () { + const projection = yield* threads.getThreadProjection(input.threadId); + if (projection.thread.titleRegeneration?.requestId !== input.requestId) { + return { type: "stale" as const }; } - yield* complete({ - ...input, - ...(outcome.title === undefined ? {} : { title: outcome.title }), + + const project = yield* projects.getById({ projectId: projection.thread.projectId }); + if (Option.isNone(project)) { + return { type: "complete" as const }; + } + + let context: { + readonly message: string; + readonly attachments: ReadonlyArray; + }; + if (input.kind.type === "initial") { + const messageId = input.kind.messageId; + const message = projection.messages.find( + (candidate) => candidate.id === messageId && !candidate.streaming, + ); + context = + message === undefined + ? { message: "", attachments: [] } + : { message: message.text, attachments: message.attachments }; + } else { + context = formatThreadTitleContext( + projection.messages.filter((message) => !message.streaming), + ); + } + if (context.message.length === 0 && context.attachments.length === 0) { + return { type: "complete" as const }; + } + + const settings = yield* serverSettings.getSettings; + const result = yield* textGeneration.generateThreadTitle({ + cwd: projection.thread.worktreePath ?? project.value.workspaceRoot, + message: context.message, + attachments: context.attachments, + ...(input.kind.type === "regenerate" ? { previousTitle: projection.thread.title } : {}), + modelSelection: settings.textGenerationModelSelection, }); + const generatedTitle = result.title.trim(); + return generatedTitle === "New thread" || + (input.kind.type === "regenerate" && generatedTitle === projection.thread.title.trim()) + ? { type: "complete" as const } + : { type: "complete" as const, title: result.title }; + }).pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.interrupt + : Effect.logWarning("Thread title generation failed", { + threadId: input.threadId, + requestId: input.requestId, + cause, + }).pipe(Effect.as({ type: "complete" as const })), + ), + ); + + if (outcome.type === "stale") { + return; + } + yield* complete({ + ...input, + ...(outcome.title === undefined ? {} : { title: outcome.title }), }); + }); + + return ThreadTitleRegenerationService.of({ execute }); +}); - return ThreadTitleRegenerationService.of({ execute }); - }), -); +export const layer = Layer.effect(ThreadTitleRegenerationService, make); From 93fffca5c5b2397004d165d0dbf97c78d8d937d9 Mon Sep 17 00:00:00 2001 From: Yusuf007R Date: Sun, 2 Aug 2026 02:27:47 -0500 Subject: [PATCH 3/4] fix(orchestration-v2): harden provisional title handling --- .../features/threads/NewTaskDraftScreen.tsx | 7 ++- apps/mobile/src/lib/projectThreadStartTurn.ts | 13 +---- .../mobile/src/state/use-pending-new-tasks.ts | 4 +- .../src/state/use-thread-outbox-drain.ts | 11 ++-- .../ThreadLaunchService.test.ts | 51 +++++++++++++++++++ .../orchestration-v2/ThreadLaunchService.ts | 1 - apps/web/src/components/ChatView.tsx | 33 +++++------- .../client-runtime/src/operations/index.ts | 1 + .../src/operations/threadTitle.test.ts | 47 +++++++++++++++++ .../src/operations/threadTitle.ts | 32 ++++++++++++ 10 files changed, 159 insertions(+), 41 deletions(-) create mode 100644 packages/client-runtime/src/operations/threadTitle.test.ts create mode 100644 packages/client-runtime/src/operations/threadTitle.ts diff --git a/apps/mobile/src/features/threads/NewTaskDraftScreen.tsx b/apps/mobile/src/features/threads/NewTaskDraftScreen.tsx index 6d204e3a8e6..8934838000e 100644 --- a/apps/mobile/src/features/threads/NewTaskDraftScreen.tsx +++ b/apps/mobile/src/features/threads/NewTaskDraftScreen.tsx @@ -12,6 +12,7 @@ import { isAtomCommandInterrupted, squashAtomCommandFailure, } from "@t3tools/client-runtime/state/runtime"; +import { deriveThreadTitleSeed } from "@t3tools/client-runtime/operations"; import { ComposerEditor, type ComposerEditorHandle } from "../../components/ComposerEditor"; import { @@ -44,7 +45,6 @@ import { } from "../../state/use-composer-drafts"; import { useEnvironmentServerConfig, useProjects } from "../../state/entities"; import { buildModelMenuActions, resolveSelectableModelSelection } from "../../lib/modelOptions"; -import { deriveThreadTitleFromPrompt } from "../../lib/projectThreadStartTurn"; import { armAgentAwarenessLiveActivityForLocalWork } from "../agent-awareness/remoteRegistration"; import { enqueueThreadOutboxMessage, removeThreadOutboxMessage } from "../../state/thread-outbox"; import { useRemoteConnectionStatus } from "../../state/use-remote-environment-registry"; @@ -853,7 +853,10 @@ export function NewTaskDraftScreen(props: { // -only Activity start. If creation fails, the token registration's replay // finds no work and ends the card within seconds. armAgentAwarenessLiveActivityForLocalWork({ - threadTitle: deriveThreadTitleFromPrompt(initialMessageText), + threadTitle: deriveThreadTitleSeed({ + text: initialMessageText, + attachments: draft.attachments, + }), projectTitle: selectedProject.title, }); const result = await createProjectThread({ diff --git a/apps/mobile/src/lib/projectThreadStartTurn.ts b/apps/mobile/src/lib/projectThreadStartTurn.ts index 310f1818590..f8461e494b2 100644 --- a/apps/mobile/src/lib/projectThreadStartTurn.ts +++ b/apps/mobile/src/lib/projectThreadStartTurn.ts @@ -7,19 +7,10 @@ import { type ProviderInteractionMode, type RuntimeMode, } from "@t3tools/contracts"; +import { deriveThreadTitleSeed } from "@t3tools/client-runtime/operations"; import { toUploadChatImageAttachments, type DraftComposerImageAttachment } from "./composerImages"; -export function deriveThreadTitleFromPrompt(value: string): string { - const trimmed = value.trim(); - if (trimmed.length === 0) { - return "New thread"; - } - - const compact = trimmed.replace(/\s+/g, " "); - return compact.length <= 72 ? compact : `${compact.slice(0, 69).trimEnd()}...`; -} - export interface ProjectThreadStartTurnSpec { readonly projectId: ProjectId; readonly projectCwd: string; @@ -46,7 +37,7 @@ export interface ProjectThreadStartTurnSpec { * offline outbox drain so both deliver identical commands. */ export function buildProjectThreadStartTurnInput(spec: ProjectThreadStartTurnSpec) { - const title = deriveThreadTitleFromPrompt(spec.text); + const title = deriveThreadTitleSeed({ text: spec.text, attachments: spec.attachments }); const isWorktree = spec.workspaceMode === "worktree"; return { commandId: CommandId.make(spec.commandId), diff --git a/apps/mobile/src/state/use-pending-new-tasks.ts b/apps/mobile/src/state/use-pending-new-tasks.ts index ccfe1527b3d..bbb508983ff 100644 --- a/apps/mobile/src/state/use-pending-new-tasks.ts +++ b/apps/mobile/src/state/use-pending-new-tasks.ts @@ -1,6 +1,6 @@ import { useMemo } from "react"; +import { deriveThreadTitleSeed } from "@t3tools/client-runtime/operations"; -import { deriveThreadTitleFromPrompt } from "../lib/projectThreadStartTurn"; import { flattenQueuedThreadMessages, type QueuedThreadCreation, @@ -26,7 +26,7 @@ export function usePendingNewTasks(): ReadonlyArray { tasks.push({ message, creation: message.creation, - title: deriveThreadTitleFromPrompt(message.text), + title: deriveThreadTitleSeed({ text: message.text, attachments: message.attachments }), }); } tasks.sort((left, right) => right.message.createdAt.localeCompare(left.message.createdAt)); diff --git a/apps/mobile/src/state/use-thread-outbox-drain.ts b/apps/mobile/src/state/use-thread-outbox-drain.ts index 706abdc1330..dcef23ce8c6 100644 --- a/apps/mobile/src/state/use-thread-outbox-drain.ts +++ b/apps/mobile/src/state/use-thread-outbox-drain.ts @@ -5,6 +5,7 @@ import { type EnvironmentThreadShell, } from "@t3tools/client-runtime/state/shell"; import type { AtomCommandResult } from "@t3tools/client-runtime/state/runtime"; +import { deriveThreadTitleSeed } from "@t3tools/client-runtime/operations"; import { CommandId, DEFAULT_PROVIDER_INTERACTION_MODE, @@ -17,10 +18,7 @@ import { AsyncResult, Atom } from "effect/unstable/reactivity"; import { useCallback, useEffect, useRef, useState } from "react"; import { scopedThreadKey } from "../lib/scopedEntities"; -import { - buildProjectThreadStartTurnInput, - deriveThreadTitleFromPrompt, -} from "../lib/projectThreadStartTurn"; +import { buildProjectThreadStartTurnInput } from "../lib/projectThreadStartTurn"; import { toUploadChatImageAttachments } from "../lib/composerImages"; import { randomHex } from "../lib/uuid"; import { appAtomRegistry } from "./atom-registry"; @@ -234,7 +232,10 @@ export function useThreadOutboxDrain(): void { attachments: toUploadChatImageAttachments(queuedMessage.attachments), }, modelSelection: settings.modelSelection, - titleSeed: deriveThreadTitleFromPrompt(queuedMessage.text), + titleSeed: deriveThreadTitleSeed({ + text: queuedMessage.text, + attachments: queuedMessage.attachments, + }), runtimeMode: settings.runtimeMode, interactionMode: settings.interactionMode, createdAt: queuedMessage.createdAt, diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts index ef9c057b2c4..869f440b1d5 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.test.ts @@ -533,6 +533,57 @@ it.effect("arms durable title generation after accepting the first message", () }), ); +it.effect("does not update a reused thread title when the initial message is rejected", () => + Effect.gen(function* () { + const harness = makeHarness(); + yield* Effect.gen(function* () { + const launches = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const outbox = yield* EffectOutbox.EffectOutboxV2; + const threadId = ThreadId.make("thread:launch:reused-title-failure"); + yield* threads.dispatch({ + type: "thread.create", + commandId: CommandId.make("command:launch:reused-title-failure:create"), + threadId, + projectId, + title: "Original title", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdBy: "user", + creationSource: "web", + }); + + const commandId = CommandId.make("command:launch:reused-title-failure"); + const failed = yield* launches + .launch({ + ...launchInput({ + command: commandId, + thread: threadId, + message: "Generate a provisional title", + }), + reuseExistingThread: true, + title: "Generate a provisional title", + generateTitle: true, + modelSelection: { + instanceId: ProviderInstanceId.make("missing-provider"), + model: "missing-model", + }, + }) + .pipe(Effect.exit); + + assert.isTrue(Exit.isFailure(failed)); + const projection = yield* threads.getThreadProjection(threadId); + assert.equal(projection.thread.title, "Original title"); + assert.isUndefined(projection.thread.titleRegeneration); + assert.isEmpty(projection.messages); + assert.isEmpty(yield* outbox.listByCommandId(CommandId.make(`${commandId}:initial-message`))); + }).pipe(Effect.provide(harness.layer)); + }), +); + it.effect("generates an initial title for an attachment-only message", () => Effect.gen(function* () { const harness = makeHarness(); diff --git a/apps/server/src/orchestration-v2/ThreadLaunchService.ts b/apps/server/src/orchestration-v2/ThreadLaunchService.ts index 600fe0b1df0..1e095d6bac4 100644 --- a/apps/server/src/orchestration-v2/ThreadLaunchService.ts +++ b/apps/server/src/orchestration-v2/ThreadLaunchService.ts @@ -425,7 +425,6 @@ export const make = Effect.gen(function* () { type: "thread.metadata.update", commandId: input.commandId, threadId: candidateThreadId, - title: input.title, }) : threads.dispatch({ type: "thread.create", diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 05ba2853bb1..57b55328583 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -25,6 +25,7 @@ import { connectionStatusTitle, type EnvironmentConnectionPresentation, } from "@t3tools/client-runtime/connection"; +import { deriveThreadTitleSeed } from "@t3tools/client-runtime/operations"; import { effectiveSettled, effectiveSnoozed } from "@t3tools/client-runtime/state/thread-settled"; import { deriveThreadActivityRun, @@ -5157,26 +5158,18 @@ function ChatViewContent(props: ChatViewProps) { clearComposerDraftContent(composerDraftTarget); composerRef.current?.resetCursorState(); - let firstComposerImageName: string | null = null; - if (composerImagesSnapshot.length > 0) { - const firstComposerImage = composerImagesSnapshot[0]; - if (firstComposerImage) { - firstComposerImageName = firstComposerImage.name; - } - } - let titleSeed = trimmed; - if (!titleSeed) { - if (firstComposerImageName) { - titleSeed = `Image: ${firstComposerImageName}`; - } else if (composerTerminalContextsSnapshot.length > 0) { - titleSeed = formatTerminalContextLabel(composerTerminalContextsSnapshot[0]!); - } else if (composerElementContextsSnapshot.length > 0) { - titleSeed = formatElementContextLabel(composerElementContextsSnapshot[0]!); - } else { - titleSeed = "New thread"; - } - } - const title = truncate(titleSeed); + const title = deriveThreadTitleSeed({ + text: trimmed, + attachments: composerImagesSnapshot, + fallbackLabels: [ + composerTerminalContextsSnapshot[0] === undefined + ? null + : formatTerminalContextLabel(composerTerminalContextsSnapshot[0]), + composerElementContextsSnapshot[0] === undefined + ? null + : formatElementContextLabel(composerElementContextsSnapshot[0]), + ], + }); const threadCreateModelSelection = createModelSelection( ctxSelectedModelSelection.instanceId, ctxSelectedModel || activeProject.defaultModelSelection?.model || DEFAULT_MODEL, diff --git a/packages/client-runtime/src/operations/index.ts b/packages/client-runtime/src/operations/index.ts index b7307fbb81f..2bafb27be27 100644 --- a/packages/client-runtime/src/operations/index.ts +++ b/packages/client-runtime/src/operations/index.ts @@ -1,2 +1,3 @@ export * from "./commands.ts"; export * from "./projects.ts"; +export * from "./threadTitle.ts"; diff --git a/packages/client-runtime/src/operations/threadTitle.test.ts b/packages/client-runtime/src/operations/threadTitle.test.ts new file mode 100644 index 00000000000..ab7aaff7ff4 --- /dev/null +++ b/packages/client-runtime/src/operations/threadTitle.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from "@effect/vitest"; + +import { deriveThreadTitleSeed } from "./threadTitle.ts"; + +describe("deriveThreadTitleSeed", () => { + it("prefers normalized message text", () => { + expect( + deriveThreadTitleSeed({ + text: " Investigate\n login failures ", + attachments: [{ name: "login-error.png" }], + fallbackLabels: ["Terminal 1"], + }), + ).toBe("Investigate login failures"); + }); + + it("uses the first attachment name when text is empty", () => { + expect( + deriveThreadTitleSeed({ + text: " \n ", + attachments: [{ name: "login-error.png" }, { name: "other.png" }], + }), + ).toBe("Image: login-error.png"); + }); + + it("uses the first non-empty fallback label after text and attachments", () => { + expect( + deriveThreadTitleSeed({ + text: "", + attachments: [], + fallbackLabels: [null, " ", "Terminal 1"], + }), + ).toBe("Terminal 1"); + }); + + it("falls back to New thread", () => { + expect(deriveThreadTitleSeed({ text: "", attachments: [] })).toBe("New thread"); + }); + + it("applies the shared title truncation", () => { + expect( + deriveThreadTitleSeed({ + text: "x".repeat(60), + attachments: [], + }), + ).toBe(`${"x".repeat(50)}...`); + }); +}); diff --git a/packages/client-runtime/src/operations/threadTitle.ts b/packages/client-runtime/src/operations/threadTitle.ts new file mode 100644 index 00000000000..36771e875f8 --- /dev/null +++ b/packages/client-runtime/src/operations/threadTitle.ts @@ -0,0 +1,32 @@ +import { truncate } from "@t3tools/shared/String"; + +export interface ThreadTitleSeedInput { + readonly text: string; + readonly attachments: ReadonlyArray<{ readonly name: string }>; + readonly fallbackLabels?: ReadonlyArray; +} + +function normalizeTitleSeed(value: string): string { + return value.trim().replace(/\s+/gu, " "); +} + +export function deriveThreadTitleSeed(input: ThreadTitleSeedInput): string { + const text = normalizeTitleSeed(input.text); + if (text.length > 0) { + return truncate(text); + } + + const attachmentName = normalizeTitleSeed(input.attachments[0]?.name ?? ""); + if (attachmentName.length > 0) { + return truncate(`Image: ${attachmentName}`); + } + + for (const label of input.fallbackLabels ?? []) { + const normalized = normalizeTitleSeed(label ?? ""); + if (normalized.length > 0) { + return truncate(normalized); + } + } + + return "New thread"; +} From c8a058fe282d8e2cb919e0aafa8ac23d0d68247b Mon Sep 17 00:00:00 2001 From: Yusuf007R Date: Sun, 2 Aug 2026 03:32:15 -0500 Subject: [PATCH 4/4] fix(orchestration-v2): keep title generation off provider lane --- .../src/orchestration-v2/EffectOutbox.ts | 13 +++ .../FoundationPersistence.test.ts | 105 ++++++++++++++++++ 2 files changed, 118 insertions(+) diff --git a/apps/server/src/orchestration-v2/EffectOutbox.ts b/apps/server/src/orchestration-v2/EffectOutbox.ts index dc6768c89d8..85180279fc7 100644 --- a/apps/server/src/orchestration-v2/EffectOutbox.ts +++ b/apps/server/src/orchestration-v2/EffectOutbox.ts @@ -273,6 +273,8 @@ export const layer: Layer.Layer = La available, Array.from({ length: Math.min(64, Math.max(0, Math.floor(count))) }, () => undefined), ).pipe(Effect.asVoid); + // Title generation is correlated metadata work, so it has its own + // per-thread lane and cannot delay provider lifecycle effects. const claimableCandidatePredicate = (availableBefore?: string) => sql` ${ @@ -286,6 +288,17 @@ export const layer: Layer.Layer = La FROM orchestration_v2_effect_outbox AS active WHERE active.thread_id = candidate.thread_id AND active.status = 'running' + AND ( + ( + candidate.effect_type = 'thread-title.generate' + AND active.effect_type = 'thread-title.generate' + ) + OR + ( + candidate.effect_type != 'thread-title.generate' + AND active.effect_type != 'thread-title.generate' + ) + ) ) `; diff --git a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts index 90df93815ec..e6159ee0772 100644 --- a/apps/server/src/orchestration-v2/FoundationPersistence.test.ts +++ b/apps/server/src/orchestration-v2/FoundationPersistence.test.ts @@ -1071,6 +1071,111 @@ it.layer(TestLayer)("orchestration V2 foundation persistence", (it) => { }), ); + it.effect("runs title generation beside critical work while serializing each lane", () => + Effect.gen(function* () { + const outbox = yield* EffectOutboxV2; + const commandId = CommandId.make("command:foundation-title-effect-lane"); + const threadId = ThreadId.make("thread:foundation-title-effect-lane"); + const titleEffectId = "effect:foundation-title-effect-lane:a-title"; + const providerEffectId = "effect:foundation-title-effect-lane:b-provider"; + const nextTitleEffectId = "effect:foundation-title-effect-lane:c-title"; + const nextCriticalEffectId = "effect:foundation-title-effect-lane:d-critical"; + yield* outbox.enqueue([ + { + id: titleEffectId, + commandId, + threadId, + request: { + type: "thread-title.generate", + kind: { + type: "initial", + messageId: MessageId.make("message:foundation-title-effect-lane"), + }, + }, + }, + { + id: providerEffectId, + commandId, + threadId, + request: { + type: "provider-turn.start", + runId: RunId.make("run:foundation-title-effect-lane"), + }, + }, + { + id: nextTitleEffectId, + commandId, + threadId, + request: { type: "thread-title.generate", kind: { type: "regenerate" } }, + }, + { + id: nextCriticalEffectId, + commandId, + threadId, + request: { type: "terminal.cleanup" }, + }, + ]); + + const title = yield* outbox.claimNext({ + workerId: "title-lane-worker", + leaseDurationMs: 30_000, + }); + assert.isTrue(Option.isSome(title)); + if (Option.isNone(title)) return; + assert.equal(title.value.id, titleEffectId); + + const provider = yield* outbox.claimNext({ + workerId: "critical-lane-worker", + leaseDurationMs: 30_000, + }); + assert.isTrue(Option.isSome(provider)); + if (Option.isNone(provider)) return; + assert.equal(provider.value.id, providerEffectId); + + assert.isTrue( + Option.isNone( + yield* outbox.claimNext({ + workerId: "blocked-lanes-worker", + leaseDurationMs: 30_000, + }), + ), + ); + + yield* outbox.succeed({ + effectId: provider.value.id, + workerId: "critical-lane-worker", + }); + const nextCritical = yield* outbox.claimNext({ + workerId: "critical-lane-worker", + leaseDurationMs: 30_000, + }); + assert.isTrue(Option.isSome(nextCritical)); + if (Option.isNone(nextCritical)) return; + assert.equal(nextCritical.value.id, nextCriticalEffectId); + yield* outbox.succeed({ + effectId: nextCritical.value.id, + workerId: "critical-lane-worker", + }); + + yield* outbox.succeed({ + effectId: title.value.id, + workerId: "title-lane-worker", + }); + const nextTitle = yield* outbox.claimNext({ + workerId: "title-lane-worker", + leaseDurationMs: 30_000, + }); + assert.isTrue(Option.isSome(nextTitle)); + if (Option.isSome(nextTitle)) { + assert.equal(nextTitle.value.id, nextTitleEffectId); + yield* outbox.succeed({ + effectId: nextTitle.value.id, + workerId: "title-lane-worker", + }); + } + }), + ); + it.effect("ignores deadlines blocked by a running effect on the same thread", () => Effect.gen(function* () { const outbox = yield* EffectOutboxV2;