diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 6774731a73e..663300cf48a 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -22,6 +22,8 @@ import { PreviewSnapshotToolkit, PreviewStandardToolkit, } from "./toolkits/preview/tools.ts"; +import { ThreadToolkitHandlersLayerLive } from "./toolkits/threads/handlers.ts"; +import { ThreadToolkit } from "./toolkits/threads/tools.ts"; const unauthorized = HttpServerResponse.jsonUnsafe( { @@ -208,10 +210,17 @@ export const PreviewToolkitRegistrationLive = Layer.mergeAll( PreviewSnapshotRegistrationLive, ); +export const ThreadToolkitRegistrationLive = McpServer.toolkit(ThreadToolkit).pipe( + Layer.provide(ThreadToolkitHandlersLayerLive), +); + const McpTransportLive = McpServer.layerHttp({ name: "T3 Code", version: packageJson.version, path: "/mcp", }).pipe(Layer.provide(McpAuthMiddlewareLive)); -export const layer = PreviewToolkitRegistrationLive.pipe(Layer.provideMerge(McpTransportLive)); +export const layer = Layer.mergeAll( + PreviewToolkitRegistrationLive, + ThreadToolkitRegistrationLive, +).pipe(Layer.provideMerge(McpTransportLive)); diff --git a/apps/server/src/mcp/McpInvocationContext.test.ts b/apps/server/src/mcp/McpInvocationContext.test.ts index 39c68689047..529518be0e8 100644 --- a/apps/server/src/mcp/McpInvocationContext.test.ts +++ b/apps/server/src/mcp/McpInvocationContext.test.ts @@ -37,3 +37,26 @@ it.effect("reports the scoped credential context when preview capability is unav expect(error.message).toBe("MCP credential does not grant the preview capability."); }); }); + +it.effect("reports a missing thread coordination capability", () => { + const invocation: McpInvocationContext.McpInvocationScope = { + environmentId: EnvironmentId.make("environment-1"), + threadId: ThreadId.make("thread-1"), + providerSessionId: "provider-session-1", + providerInstanceId: ProviderInstanceId.make("codex"), + capabilities: new Set(["preview"]), + issuedAt: 1, + expiresAt: 2, + }; + + return Effect.gen(function* () { + const error = yield* McpInvocationContext.requireMcpCapability("threads").pipe( + Effect.provideService(McpInvocationContext.McpInvocationContext, invocation), + Effect.flip, + ); + + expect(error).toBeInstanceOf(PreviewAutomationUnavailableError); + expect(error.capability).toBe("threads"); + expect(error.message).toBe("MCP credential does not grant the threads capability."); + }); +}); diff --git a/apps/server/src/mcp/McpInvocationContext.ts b/apps/server/src/mcp/McpInvocationContext.ts index b13bf2d312e..4603c42d43c 100644 --- a/apps/server/src/mcp/McpInvocationContext.ts +++ b/apps/server/src/mcp/McpInvocationContext.ts @@ -7,7 +7,7 @@ import { import * as Context from "effect/Context"; import * as Effect from "effect/Effect"; -export type McpCapability = "preview"; +export type McpCapability = "preview" | "threads"; export interface McpInvocationScope { readonly environmentId: EnvironmentId; diff --git a/apps/server/src/mcp/McpSessionRegistry.ts b/apps/server/src/mcp/McpSessionRegistry.ts index 67c4f2f0ff0..9c1e474fc6d 100644 --- a/apps/server/src/mcp/McpSessionRegistry.ts +++ b/apps/server/src/mcp/McpSessionRegistry.ts @@ -114,7 +114,7 @@ const makeWithOptions = Effect.fn("McpSessionRegistry.make")(function* ( threadId: ThreadId.make(request.threadId), providerSessionId, providerInstanceId: ProviderInstanceId.make(request.providerInstanceId), - capabilities: new Set(["preview"]), + capabilities: new Set(["preview", "threads"]), issuedAt, expiresAt, }; diff --git a/apps/server/src/mcp/toolkits/threads/handlers.ts b/apps/server/src/mcp/toolkits/threads/handlers.ts new file mode 100644 index 00000000000..59d73646848 --- /dev/null +++ b/apps/server/src/mcp/toolkits/threads/handlers.ts @@ -0,0 +1,249 @@ +import { + CommandId, + MessageId, + ThreadId, + type OrchestrationThread, + type OrchestrationThreadShell, +} from "@t3tools/contracts"; +import * as Cause from "effect/Cause"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Option from "effect/Option"; + +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import { OrchestrationEngineService } from "../../../orchestration/Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import { ThreadToolkit, ThreadToolError } from "./tools.ts"; + +const asToolFailure = (operation: string) => (cause: Cause.Cause) => + Effect.fail( + new ThreadToolError({ + operation, + detail: Cause.pretty(cause), + }), + ); + +const runThreadOperation = ( + operation: string, + effect: Effect.Effect, +): Effect.Effect => effect.pipe(Effect.catchCause(asToolFailure(operation))); + +const makeHandlers = Effect.gen(function* () { + const invocation = yield* McpInvocationContext.requireMcpCapability("threads"); + const engine = yield* OrchestrationEngineService; + const query = yield* ProjectionSnapshotQuery; + const crypto = yield* Crypto.Crypto; + + const newId = Effect.fn("ThreadToolkit.newId")(function* () { + return yield* crypto.randomUUIDv4.pipe(Effect.orDie); + }); + + const currentThread = Effect.fn("ThreadToolkit.currentThread")(function* () { + const thread = yield* query.getThreadDetailById(invocation.threadId); + if (Option.isNone(thread)) { + return yield* new ThreadToolError({ + operation: "thread_current", + detail: `Current thread '${invocation.threadId}' was not found.`, + }); + } + return thread.value; + }); + + const projectThread = Effect.fn("ThreadToolkit.projectThread")(function* (threadId: ThreadId) { + const [current, target] = yield* Effect.all([ + currentThread(), + query.getThreadDetailById(threadId), + ]); + if (Option.isNone(target) || target.value.projectId !== current.projectId) { + return yield* new ThreadToolError({ + operation: "thread_target", + detail: `Thread '${threadId}' is not an active thread in the current project.`, + }); + } + return target.value; + }); + + const startTurn = Effect.fn("ThreadToolkit.startTurn")(function* ( + thread: OrchestrationThread, + prompt: string, + ) { + const [commandUuid, messageUuid, createdAt] = yield* Effect.all([ + newId(), + newId(), + DateTime.now.pipe(Effect.map(DateTime.formatIso)), + ]); + yield* engine.dispatch({ + type: "thread.turn.start", + commandId: CommandId.make(commandUuid), + threadId: thread.id, + message: { + messageId: MessageId.make(messageUuid), + role: "user", + text: prompt, + attachments: [], + }, + modelSelection: thread.modelSelection, + runtimeMode: thread.runtimeMode, + interactionMode: thread.interactionMode, + createdAt, + }); + }); + + const resultFor = (thread: Pick, turnStarted: boolean) => ({ + threadId: thread.id, + title: thread.title, + turnStarted, + }); + + return { + thread_list: (input: { readonly includeArchived?: boolean | undefined }) => + Effect.gen(function* () { + const current = yield* currentThread(); + const snapshot = yield* query.getShellSnapshot(); + return snapshot.threads + .filter( + (thread) => + thread.projectId === current.projectId && + (input.includeArchived === true || thread.archivedAt === null), + ) + .map((thread: OrchestrationThreadShell) => ({ + threadId: thread.id, + title: thread.title, + status: thread.session?.status ?? "idle", + createdAt: thread.createdAt, + ...(thread.forkedFrom?.threadId + ? { forkedFromThreadId: thread.forkedFrom.threadId } + : {}), + ...(thread.forkedFrom?.turnId ? { forkedFromTurnId: thread.forkedFrom.turnId } : {}), + })); + }), + thread_create: (input: { + readonly title?: string | undefined; + readonly prompt?: string | undefined; + }) => + Effect.gen(function* () { + const current = yield* currentThread(); + const [threadUuid, commandUuid, createdAt] = yield* Effect.all([ + newId(), + newId(), + DateTime.now.pipe(Effect.map(DateTime.formatIso)), + ]); + const threadId = ThreadId.make(threadUuid); + const title = input.title ?? "New thread"; + yield* engine.dispatch({ + type: "thread.create", + commandId: CommandId.make(commandUuid), + threadId, + projectId: current.projectId, + title, + modelSelection: current.modelSelection, + runtimeMode: current.runtimeMode, + interactionMode: current.interactionMode, + branch: current.branch, + worktreePath: current.worktreePath, + createdAt, + }); + const created = { ...current, id: threadId, title }; + if (input.prompt !== undefined) yield* startTurn(created, input.prompt); + return resultFor(created, input.prompt !== undefined); + }), + thread_fork: (input: { + readonly sourceThreadId?: ThreadId | undefined; + readonly sourceTurnId?: import("@t3tools/contracts").TurnId | undefined; + readonly title?: string | undefined; + readonly prompt?: string | undefined; + }) => + Effect.gen(function* () { + const source = yield* projectThread(input.sourceThreadId ?? invocation.threadId); + const [threadUuid, commandUuid, createdAt] = yield* Effect.all([ + newId(), + newId(), + DateTime.now.pipe(Effect.map(DateTime.formatIso)), + ]); + const threadId = ThreadId.make(threadUuid); + const title = input.title ?? `${source.title} (fork)`; + yield* engine.dispatch({ + type: "thread.fork", + commandId: CommandId.make(commandUuid), + threadId, + sourceThreadId: source.id, + ...(input.sourceTurnId !== undefined ? { sourceTurnId: input.sourceTurnId } : {}), + title, + createdAt, + }); + const forked = { + ...source, + id: threadId, + title, + forkedFrom: { + threadId: source.id, + turnId: input.sourceTurnId ?? source.latestTurn?.turnId ?? null, + }, + session: null, + }; + if (input.prompt !== undefined) yield* startTurn(forked, input.prompt); + return resultFor(forked, input.prompt !== undefined); + }), + thread_send: (input: { readonly threadId: ThreadId; readonly prompt: string }) => + Effect.gen(function* () { + const target = yield* projectThread(input.threadId); + yield* startTurn(target, input.prompt); + return resultFor(target, true); + }), + thread_archive: (input: { readonly threadId: ThreadId }) => + Effect.gen(function* () { + if (input.threadId === invocation.threadId) { + return yield* new ThreadToolError({ + operation: "thread_archive", + detail: "The current agent thread cannot archive itself.", + }); + } + yield* projectThread(input.threadId); + yield* engine.dispatch({ + type: "thread.archive", + commandId: CommandId.make(yield* newId()), + threadId: input.threadId, + }); + return null; + }), + }; +}); + +export const ThreadToolkitHandlers = { + thread_list: (input: { readonly includeArchived?: boolean | undefined }) => + runThreadOperation( + "thread_list", + makeHandlers.pipe(Effect.flatMap((handlers) => handlers.thread_list(input))), + ), + thread_create: (input: { + readonly title?: string | undefined; + readonly prompt?: string | undefined; + }) => + runThreadOperation( + "thread_create", + makeHandlers.pipe(Effect.flatMap((handlers) => handlers.thread_create(input))), + ), + thread_fork: (input: { + readonly sourceThreadId?: ThreadId | undefined; + readonly sourceTurnId?: import("@t3tools/contracts").TurnId | undefined; + readonly title?: string | undefined; + readonly prompt?: string | undefined; + }) => + runThreadOperation( + "thread_fork", + makeHandlers.pipe(Effect.flatMap((handlers) => handlers.thread_fork(input))), + ), + thread_send: (input: { readonly threadId: ThreadId; readonly prompt: string }) => + runThreadOperation( + "thread_send", + makeHandlers.pipe(Effect.flatMap((handlers) => handlers.thread_send(input))), + ), + thread_archive: (input: { readonly threadId: ThreadId }) => + runThreadOperation( + "thread_archive", + makeHandlers.pipe(Effect.flatMap((handlers) => handlers.thread_archive(input))), + ), +} satisfies Parameters[0]; + +export const ThreadToolkitHandlersLayerLive = ThreadToolkit.toLayer(ThreadToolkitHandlers); diff --git a/apps/server/src/mcp/toolkits/threads/tools.test.ts b/apps/server/src/mcp/toolkits/threads/tools.test.ts new file mode 100644 index 00000000000..23f9bdadbe3 --- /dev/null +++ b/apps/server/src/mcp/toolkits/threads/tools.test.ts @@ -0,0 +1,26 @@ +import { expect, it } from "@effect/vitest"; +import { Tool } from "effect/unstable/ai"; + +import { ThreadToolkit } from "./tools.ts"; + +it("exports provider-compatible thread management schemas", () => { + expect(Object.keys(ThreadToolkit.tools).sort()).toEqual([ + "thread_archive", + "thread_create", + "thread_fork", + "thread_list", + "thread_send", + ]); + + for (const tool of Object.values(ThreadToolkit.tools)) { + const schema = Tool.getJsonSchema(tool) as { + readonly type?: unknown; + readonly anyOf?: unknown; + readonly oneOf?: unknown; + }; + expect(tool.description?.length ?? 0).toBeGreaterThan(40); + expect(schema.type, `${tool.name} must export a top-level object schema`).toBe("object"); + expect(schema.anyOf, `${tool.name} must not export a root anyOf`).toBeUndefined(); + expect(schema.oneOf, `${tool.name} must not export a root oneOf`).toBeUndefined(); + } +}); diff --git a/apps/server/src/mcp/toolkits/threads/tools.ts b/apps/server/src/mcp/toolkits/threads/tools.ts new file mode 100644 index 00000000000..a9b3fa96642 --- /dev/null +++ b/apps/server/src/mcp/toolkits/threads/tools.ts @@ -0,0 +1,131 @@ +import { + PROVIDER_SEND_TURN_MAX_INPUT_CHARS, + ThreadId, + TrimmedNonEmptyString, + TurnId, +} from "@t3tools/contracts"; +import * as Crypto from "effect/Crypto"; +import * as Schema from "effect/Schema"; +import { Tool, Toolkit } from "effect/unstable/ai"; + +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import { OrchestrationEngineService } from "../../../orchestration/Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../../../orchestration/Services/ProjectionSnapshotQuery.ts"; + +export class ThreadToolError extends Schema.TaggedErrorClass()("ThreadToolError", { + operation: Schema.String, + detail: Schema.String, +}) {} + +const OptionalPrompt = Schema.optional( + TrimmedNonEmptyString.check(Schema.isMaxLength(PROVIDER_SEND_TURN_MAX_INPUT_CHARS)), +); + +const ThreadSummary = Schema.Struct({ + threadId: ThreadId, + title: Schema.String, + status: Schema.String, + createdAt: Schema.String, + forkedFromThreadId: Schema.optional(ThreadId), + forkedFromTurnId: Schema.optional(TurnId), +}); + +const ThreadMutationResult = Schema.Struct({ + threadId: ThreadId, + title: Schema.String, + turnStarted: Schema.Boolean, +}); + +const dependencies = [ + Crypto.Crypto, + McpInvocationContext.McpInvocationContext, + OrchestrationEngineService, + ProjectionSnapshotQuery, +]; + +const readonlyThreadTool = (tool: T): T => + tool + .annotate(Tool.Readonly, true) + .annotate(Tool.Destructive, false) + .annotate(Tool.Idempotent, true) as T; + +const mutatingThreadTool = (tool: T): T => + tool.annotate(Tool.Readonly, false).annotate(Tool.Destructive, false) as T; + +export const ThreadListTool = readonlyThreadTool( + Tool.make("thread_list", { + description: + "List T3 Code threads in the current project, including fork lineage and runtime status. Archived threads are omitted unless requested.", + parameters: Schema.Struct({ + includeArchived: Schema.optional(Schema.Boolean), + }), + success: Schema.Array(ThreadSummary), + failure: ThreadToolError, + dependencies, + }).annotate(Tool.Title, "List project threads"), +); + +export const ThreadCreateTool = mutatingThreadTool( + Tool.make("thread_create", { + description: + "Create a sibling T3 Code thread in the current project. Supply prompt to immediately start its first turn.", + parameters: Schema.Struct({ + title: Schema.optional(TrimmedNonEmptyString), + prompt: OptionalPrompt, + }), + success: ThreadMutationResult, + failure: ThreadToolError, + dependencies, + }).annotate(Tool.Title, "Create project thread"), +); + +export const ThreadForkTool = mutatingThreadTool( + Tool.make("thread_fork", { + description: + "Fork a T3 Code conversation into a side thread with native provider history. Defaults to the current thread and its latest completed turn. Supply prompt to immediately start work in the fork.", + parameters: Schema.Struct({ + sourceThreadId: Schema.optional(ThreadId), + sourceTurnId: Schema.optional(TurnId), + title: Schema.optional(TrimmedNonEmptyString), + prompt: OptionalPrompt, + }), + success: ThreadMutationResult, + failure: ThreadToolError, + dependencies, + }).annotate(Tool.Title, "Fork conversation to side thread"), +); + +export const ThreadSendTool = mutatingThreadTool( + Tool.make("thread_send", { + description: + "Send a new prompt to another active T3 Code thread in the current project and start its turn.", + parameters: Schema.Struct({ + threadId: ThreadId, + prompt: TrimmedNonEmptyString.check(Schema.isMaxLength(PROVIDER_SEND_TURN_MAX_INPUT_CHARS)), + }), + success: ThreadMutationResult, + failure: ThreadToolError, + dependencies, + }).annotate(Tool.Title, "Send prompt to project thread"), +); + +export const ThreadArchiveTool = mutatingThreadTool( + Tool.make("thread_archive", { + description: + "Archive another inactive T3 Code thread in the current project. The current agent thread cannot archive itself.", + parameters: Schema.Struct({ threadId: ThreadId }), + success: Schema.Null, + failure: ThreadToolError, + dependencies, + }) + .annotate(Tool.Title, "Archive project thread") + .annotate(Tool.Destructive, true), +); + +export const ThreadToolkit = Toolkit.make( + ThreadListTool, + ThreadCreateTool, + ThreadForkTool, + ThreadSendTool, + ThreadArchiveTool, +); diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index f12df850941..60f939c8087 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -603,6 +603,8 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti interactionMode: event.payload.interactionMode, branch: event.payload.branch, worktreePath: event.payload.worktreePath, + forkedFromThreadId: null, + forkedFromTurnId: null, latestTurnId: null, createdAt: event.payload.createdAt, updatedAt: event.payload.updatedAt, @@ -615,6 +617,33 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti }); return; + case "thread.forked": + yield* projectionThreadRepository.upsert({ + threadId: event.payload.threadId, + projectId: event.payload.projectId, + title: event.payload.title, + modelSelection: event.payload.modelSelection, + runtimeMode: event.payload.runtimeMode, + interactionMode: event.payload.interactionMode, + branch: event.payload.branch, + worktreePath: event.payload.worktreePath, + forkedFromThreadId: event.payload.forkedFrom.threadId, + forkedFromTurnId: event.payload.forkedFrom.turnId, + latestTurnId: null, + createdAt: event.payload.createdAt, + updatedAt: event.payload.updatedAt, + archivedAt: null, + latestUserMessageAt: + event.payload.inheritedMessages + .toReversed() + .find((message) => message.role === "user")?.createdAt ?? null, + pendingApprovalCount: 0, + pendingUserInputCount: 0, + hasActionableProposedPlan: 0, + deletedAt: null, + }); + return; + case "thread.archived": { const existingRow = yield* projectionThreadRepository.getById({ threadId: event.payload.threadId, @@ -811,6 +840,27 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti "applyThreadMessagesProjection", )(function* (event, attachmentSideEffects) { switch (event.type) { + case "thread.forked": + yield* Effect.forEach( + event.payload.inheritedMessages, + (message) => + projectionThreadMessageRepository.upsert({ + messageId: message.id, + threadId: event.payload.threadId, + turnId: message.turnId, + role: message.role, + text: message.text, + ...(message.attachments !== undefined + ? { attachments: [...message.attachments] } + : {}), + isStreaming: message.streaming, + createdAt: message.createdAt, + updatedAt: message.updatedAt, + }), + { concurrency: 1, discard: true }, + ); + return; + case "thread.message-sent": { const existingMessage = yield* projectionThreadMessageRepository.getByMessageId({ messageId: event.payload.messageId, diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 32210436e67..9f99246bda7 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -224,6 +224,15 @@ function mapSessionRow( }; } +function mapForkedFrom(row: Schema.Schema.Type) { + return row.forkedFromThreadId == null + ? null + : { + threadId: row.forkedFromThreadId, + turnId: row.forkedFromTurnId ?? null, + }; +} + function mapProjectShellRow( row: Schema.Schema.Type, repositoryIdentity: OrchestrationProject["repositoryIdentity"], @@ -330,6 +339,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { interaction_mode AS "interactionMode", branch, worktree_path AS "worktreePath", + forked_from_thread_id AS "forkedFromThreadId", + forked_from_turn_id AS "forkedFromTurnId", latest_turn_id AS "latestTurnId", created_at AS "createdAt", updated_at AS "updatedAt", @@ -358,6 +369,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { interaction_mode AS "interactionMode", branch, worktree_path AS "worktreePath", + forked_from_thread_id AS "forkedFromThreadId", + forked_from_turn_id AS "forkedFromTurnId", latest_turn_id AS "latestTurnId", created_at AS "createdAt", updated_at AS "updatedAt", @@ -388,6 +401,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { interaction_mode AS "interactionMode", branch, worktree_path AS "worktreePath", + forked_from_thread_id AS "forkedFromThreadId", + forked_from_turn_id AS "forkedFromTurnId", latest_turn_id AS "latestTurnId", created_at AS "createdAt", updated_at AS "updatedAt", @@ -750,6 +765,8 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { interaction_mode AS "interactionMode", branch, worktree_path AS "worktreePath", + forked_from_thread_id AS "forkedFromThreadId", + forked_from_turn_id AS "forkedFromTurnId", latest_turn_id AS "latestTurnId", created_at AS "createdAt", updated_at AS "updatedAt", @@ -1182,6 +1199,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { interactionMode: row.interactionMode, branch: row.branch, worktreePath: row.worktreePath, + forkedFrom: mapForkedFrom(row), latestTurn: latestTurnByThread.get(row.threadId) ?? null, createdAt: row.createdAt, updatedAt: row.updatedAt, @@ -1380,6 +1398,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { interactionMode: row.interactionMode, branch: row.branch, worktreePath: row.worktreePath, + forkedFrom: mapForkedFrom(row), latestTurn: latestTurnByThread.get(row.threadId) ?? null, createdAt: row.createdAt, updatedAt: row.updatedAt, @@ -1509,6 +1528,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { interactionMode: row.interactionMode, branch: row.branch, worktreePath: row.worktreePath, + forkedFrom: mapForkedFrom(row), latestTurn: latestTurnByThread.get(row.threadId) ?? null, createdAt: row.createdAt, updatedAt: row.updatedAt, @@ -1643,6 +1663,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { interactionMode: row.interactionMode, branch: row.branch, worktreePath: row.worktreePath, + forkedFrom: mapForkedFrom(row), latestTurn: latestTurnByThread.get(row.threadId) ?? null, createdAt: row.createdAt, updatedAt: row.updatedAt, @@ -1883,6 +1904,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { interactionMode: threadRow.value.interactionMode, branch: threadRow.value.branch, worktreePath: threadRow.value.worktreePath, + forkedFrom: mapForkedFrom(threadRow.value), latestTurn: Option.isSome(latestTurnRow) ? mapLatestTurn(latestTurnRow.value) : null, createdAt: threadRow.value.createdAt, updatedAt: threadRow.value.updatedAt, @@ -1977,6 +1999,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { interactionMode: threadRow.value.interactionMode, branch: threadRow.value.branch, worktreePath: threadRow.value.worktreePath, + forkedFrom: mapForkedFrom(threadRow.value), latestTurn: Option.isSome(latestTurnRow) ? mapLatestTurn(latestTurnRow.value) : null, createdAt: threadRow.value.createdAt, updatedAt: threadRow.value.updatedAt, diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 9c7a7c94bb1..73bade43043 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -474,6 +474,7 @@ const make = Effect.gen(function* () { const startProviderSession = (input?: { readonly resumeCursor?: unknown; readonly provider?: ProviderDriverKind; + readonly includeForkSource?: boolean; }) => providerService.startSession(threadId, { threadId, @@ -482,6 +483,16 @@ const make = Effect.gen(function* () { ...(effectiveCwd ? { cwd: effectiveCwd } : {}), modelSelection: desiredModelSelection, ...(input?.resumeCursor !== undefined ? { resumeCursor: input.resumeCursor } : {}), + ...(input?.includeForkSource === true && thread.forkedFrom != null + ? { + forkFrom: { + threadId: thread.forkedFrom.threadId, + ...(thread.forkedFrom.turnId !== null + ? { sourceTurnId: thread.forkedFrom.turnId } + : {}), + }, + } + : {}), runtimeMode: desiredRuntimeMode, }); @@ -578,7 +589,7 @@ const make = Effect.gen(function* () { return restartedSession.threadId; } - const startedSession = yield* startProviderSession(undefined); + const startedSession = yield* startProviderSession({ includeForkSource: true }); yield* bindSessionToThread(startedSession); return startedSession.threadId; }); diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index f7ebf693440..be74153c8cc 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -3,6 +3,7 @@ import { ProjectMetaUpdatedPayload as ContractsProjectMetaUpdatedPayloadSchema, ProjectDeletedPayload as ContractsProjectDeletedPayloadSchema, ThreadCreatedPayload as ContractsThreadCreatedPayloadSchema, + ThreadForkedPayload as ContractsThreadForkedPayloadSchema, ThreadArchivedPayload as ContractsThreadArchivedPayloadSchema, ThreadMetaUpdatedPayload as ContractsThreadMetaUpdatedPayloadSchema, ThreadRuntimeModeSetPayload as ContractsThreadRuntimeModeSetPayloadSchema, @@ -28,6 +29,7 @@ export const ProjectMetaUpdatedPayload = ContractsProjectMetaUpdatedPayloadSchem export const ProjectDeletedPayload = ContractsProjectDeletedPayloadSchema; export const ThreadCreatedPayload = ContractsThreadCreatedPayloadSchema; +export const ThreadForkedPayload = ContractsThreadForkedPayloadSchema; export const ThreadArchivedPayload = ContractsThreadArchivedPayloadSchema; export const ThreadMetaUpdatedPayload = ContractsThreadMetaUpdatedPayloadSchema; export const ThreadRuntimeModeSetPayload = ContractsThreadRuntimeModeSetPayloadSchema; diff --git a/apps/server/src/orchestration/decider.fork.test.ts b/apps/server/src/orchestration/decider.fork.test.ts new file mode 100644 index 00000000000..adf006899f1 --- /dev/null +++ b/apps/server/src/orchestration/decider.fork.test.ts @@ -0,0 +1,267 @@ +import { + CommandId, + DEFAULT_PROVIDER_INTERACTION_MODE, + EventId, + MessageId, + ProjectId, + ProviderInstanceId, + ThreadId, + TurnId, + type OrchestrationEvent, + type OrchestrationReadModel, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { decideOrchestrationCommand } from "./decider.ts"; +import { createEmptyReadModel, projectEvent } from "./projector.ts"; + +const now = "2026-07-21T12:00:00.000Z"; +const sourceThreadId = ThreadId.make("thread-source"); +const forkThreadId = ThreadId.make("thread-fork"); +const turnOneId = TurnId.make("turn-1"); +const turnTwoId = TurnId.make("turn-2"); + +const seedReadModel = (): OrchestrationReadModel => ({ + ...createEmptyReadModel(now), + threads: [ + { + id: sourceThreadId, + projectId: ProjectId.make("project-1"), + title: "Source thread", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5.4", + }, + runtimeMode: "full-access", + interactionMode: DEFAULT_PROVIDER_INTERACTION_MODE, + branch: "main", + worktreePath: "/tmp/project", + forkedFrom: null, + latestTurn: { + turnId: turnTwoId, + state: "completed", + requestedAt: now, + startedAt: now, + completedAt: now, + assistantMessageId: MessageId.make("assistant-2"), + }, + createdAt: now, + updatedAt: now, + archivedAt: null, + deletedAt: null, + messages: [ + { + id: MessageId.make("user-1"), + role: "user", + text: "First question", + attachments: [], + turnId: turnOneId, + streaming: false, + createdAt: now, + updatedAt: now, + }, + { + id: MessageId.make("assistant-1"), + role: "assistant", + text: "First answer", + attachments: [], + turnId: turnOneId, + streaming: false, + createdAt: now, + updatedAt: now, + }, + { + id: MessageId.make("user-2"), + role: "user", + text: "Second question", + attachments: [], + turnId: turnTwoId, + streaming: false, + createdAt: now, + updatedAt: now, + }, + { + id: MessageId.make("assistant-2"), + role: "assistant", + text: "Second answer", + attachments: [], + turnId: turnTwoId, + streaming: false, + createdAt: now, + updatedAt: now, + }, + { + id: MessageId.make("assistant-streaming"), + role: "assistant", + text: "Unsettled answer", + attachments: [], + turnId: TurnId.make("turn-3"), + streaming: true, + createdAt: now, + updatedAt: now, + }, + ], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, + }, + ], +}); + +const forkCommand = (sourceTurnId?: TurnId) => ({ + type: "thread.fork" as const, + commandId: CommandId.make("command-fork"), + threadId: forkThreadId, + sourceThreadId, + ...(sourceTurnId === undefined ? {} : { sourceTurnId }), + title: "Forked thread", + createdAt: now, +}); + +type PlannedForkEvent = Omit< + Extract, + "sequence" +>; + +function requireForkEvent(result: unknown): PlannedForkEvent { + if ( + result === null || + typeof result !== "object" || + Array.isArray(result) || + !("type" in result) || + result.type !== "thread.forked" + ) { + throw new Error("Expected one thread.forked event."); + } + return result as PlannedForkEvent; +} + +it.layer(NodeServices.layer)("thread fork decider", (it) => { + it.effect("copies history through the requested turn and persists lineage", () => + Effect.gen(function* () { + const event = requireForkEvent( + yield* decideOrchestrationCommand({ + command: forkCommand(turnOneId), + readModel: seedReadModel(), + }), + ); + + expect(event.payload.forkedFrom).toEqual({ + threadId: sourceThreadId, + turnId: turnOneId, + }); + expect(event.payload.inheritedMessages.map((message) => message.text)).toEqual([ + "First question", + "First answer", + ]); + + const projected = yield* projectEvent(seedReadModel(), { + ...event, + sequence: 1, + eventId: EventId.make("event-fork"), + }); + expect(projected.threads.find((thread) => thread.id === forkThreadId)?.forkedFrom).toEqual({ + threadId: sourceThreadId, + turnId: turnOneId, + }); + }), + ); + + it.effect("defaults to the latest completed assistant turn and excludes streaming output", () => + Effect.gen(function* () { + const event = requireForkEvent( + yield* decideOrchestrationCommand({ + command: forkCommand(), + readModel: seedReadModel(), + }), + ); + + expect(event.payload.forkedFrom.turnId).toBe(turnTwoId); + expect(event.payload.inheritedMessages.map((message) => message.text)).toEqual([ + "First question", + "First answer", + "Second question", + "Second answer", + ]); + expect(event.payload.inheritedMessages.every((message) => !message.streaming)).toBe(true); + }), + ); + + it.effect("ignores finalized assistant segments from the currently running turn", () => + Effect.gen(function* () { + const readModel = seedReadModel(); + const source = readModel.threads[0]!; + const runningTurnId = TurnId.make("turn-3"); + const event = requireForkEvent( + yield* decideOrchestrationCommand({ + command: forkCommand(), + readModel: { + ...readModel, + threads: [ + { + ...source, + latestTurn: { + turnId: runningTurnId, + state: "running", + requestedAt: now, + startedAt: now, + completedAt: null, + assistantMessageId: MessageId.make("assistant-3"), + }, + messages: [ + ...source.messages, + { + id: MessageId.make("assistant-3"), + role: "assistant", + text: "I am invoking thread_fork now.", + attachments: [], + turnId: runningTurnId, + streaming: false, + createdAt: now, + updatedAt: now, + }, + ], + }, + ], + }, + }), + ); + + expect(event.payload.forkedFrom.turnId).toBe(turnTwoId); + expect(event.payload.inheritedMessages.map((message) => message.text)).toEqual([ + "First question", + "First answer", + "Second question", + "Second answer", + ]); + }), + ); + + it.effect("rejects a fork at a running turn", () => + Effect.gen(function* () { + const readModel = seedReadModel(); + const source = readModel.threads[0]!; + const error = yield* decideOrchestrationCommand({ + command: forkCommand(turnTwoId), + readModel: { + ...readModel, + threads: [ + { + ...source, + latestTurn: { + ...source.latestTurn!, + state: "running", + completedAt: null, + }, + }, + ], + }, + }).pipe(Effect.flip); + + expect(error.message).toContain("still running and cannot be forked"); + }), + ); +}); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 1730494ecc6..5d1d983920b 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -1,5 +1,6 @@ import { EventId, + MessageId, type OrchestrationCommand, type OrchestrationEvent, type OrchestrationReadModel, @@ -260,6 +261,90 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }; } + case "thread.fork": { + const sourceThread = yield* requireThread({ + readModel, + command, + threadId: command.sourceThreadId, + }); + yield* requireThreadAbsent({ + readModel, + command, + threadId: command.threadId, + }); + + const runningTurnId = + sourceThread.latestTurn?.state === "running" ? sourceThread.latestTurn.turnId : null; + const latestCompletedTurnId = sourceThread.messages + .toReversed() + .find( + (message) => + message.role === "assistant" && + message.turnId !== null && + message.turnId !== runningTurnId && + !message.streaming, + )?.turnId; + const sourceTurnId = command.sourceTurnId ?? latestCompletedTurnId ?? null; + if ( + sourceTurnId !== null && + !sourceThread.messages.some((message) => message.turnId === sourceTurnId) + ) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Turn '${sourceTurnId}' was not found in source thread '${sourceThread.id}'.`, + }); + } + if ( + sourceTurnId !== null && + sourceThread.latestTurn?.turnId === sourceTurnId && + sourceThread.latestTurn.state === "running" + ) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Turn '${sourceTurnId}' is still running and cannot be forked.`, + }); + } + + const cutoffIndex = + sourceTurnId === null + ? sourceThread.messages.length - 1 + : sourceThread.messages.findLastIndex((message) => message.turnId === sourceTurnId); + const inheritedMessages = sourceThread.messages + .slice(0, cutoffIndex + 1) + .filter((message) => !message.streaming) + .map((message, index) => ({ + ...message, + id: MessageId.make(`${command.threadId}:fork:${index}`), + })); + + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.forked", + payload: { + threadId: command.threadId, + projectId: sourceThread.projectId, + title: command.title, + modelSelection: sourceThread.modelSelection, + runtimeMode: sourceThread.runtimeMode, + interactionMode: sourceThread.interactionMode, + branch: sourceThread.branch, + worktreePath: sourceThread.worktreePath, + forkedFrom: { + threadId: sourceThread.id, + turnId: sourceTurnId, + }, + inheritedMessages, + createdAt: command.createdAt, + updatedAt: command.createdAt, + }, + }; + } + case "thread.delete": { yield* requireThread({ readModel, diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index fc6ab8f6fcf..5096adad583 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -17,6 +17,7 @@ import { ThreadActivityAppendedPayload, ThreadArchivedPayload, ThreadCreatedPayload, + ThreadForkedPayload, ThreadDeletedPayload, ThreadInteractionModeSetPayload, ThreadMetaUpdatedPayload, @@ -304,6 +305,48 @@ export function projectEvent( }; }); + case "thread.forked": + return Effect.gen(function* () { + const payload = yield* decodeForEvent( + ThreadForkedPayload, + event.payload, + event.type, + "payload", + ); + const thread: OrchestrationThread = yield* decodeForEvent( + OrchestrationThread, + { + id: payload.threadId, + projectId: payload.projectId, + title: payload.title, + modelSelection: payload.modelSelection, + runtimeMode: payload.runtimeMode, + interactionMode: payload.interactionMode, + branch: payload.branch, + worktreePath: payload.worktreePath, + forkedFrom: payload.forkedFrom, + latestTurn: null, + createdAt: payload.createdAt, + updatedAt: payload.updatedAt, + archivedAt: null, + deletedAt: null, + messages: payload.inheritedMessages, + activities: [], + checkpoints: [], + session: null, + }, + event.type, + "thread", + ); + const existing = nextBase.threads.find((entry) => entry.id === thread.id); + return { + ...nextBase, + threads: existing + ? nextBase.threads.map((entry) => (entry.id === thread.id ? thread : entry)) + : [...nextBase.threads, thread], + }; + }); + case "thread.deleted": return decodeForEvent(ThreadDeletedPayload, event.payload, event.type, "payload").pipe( Effect.map((payload) => ({ diff --git a/apps/server/src/persistence/Layers/ProjectionThreads.ts b/apps/server/src/persistence/Layers/ProjectionThreads.ts index 1baeb375c15..eb505143be2 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreads.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreads.ts @@ -39,6 +39,8 @@ const makeProjectionThreadRepository = Effect.gen(function* () { interaction_mode, branch, worktree_path, + forked_from_thread_id, + forked_from_turn_id, latest_turn_id, created_at, updated_at, @@ -58,6 +60,8 @@ const makeProjectionThreadRepository = Effect.gen(function* () { ${row.interactionMode}, ${row.branch}, ${row.worktreePath}, + ${row.forkedFromThreadId ?? null}, + ${row.forkedFromTurnId ?? null}, ${row.latestTurnId}, ${row.createdAt}, ${row.updatedAt}, @@ -77,6 +81,8 @@ const makeProjectionThreadRepository = Effect.gen(function* () { interaction_mode = excluded.interaction_mode, branch = excluded.branch, worktree_path = excluded.worktree_path, + forked_from_thread_id = excluded.forked_from_thread_id, + forked_from_turn_id = excluded.forked_from_turn_id, latest_turn_id = excluded.latest_turn_id, created_at = excluded.created_at, updated_at = excluded.updated_at, @@ -103,6 +109,8 @@ const makeProjectionThreadRepository = Effect.gen(function* () { interaction_mode AS "interactionMode", branch, worktree_path AS "worktreePath", + forked_from_thread_id AS "forkedFromThreadId", + forked_from_turn_id AS "forkedFromTurnId", latest_turn_id AS "latestTurnId", created_at AS "createdAt", updated_at AS "updatedAt", @@ -131,6 +139,8 @@ const makeProjectionThreadRepository = Effect.gen(function* () { interaction_mode AS "interactionMode", branch, worktree_path AS "worktreePath", + forked_from_thread_id AS "forkedFromThreadId", + forked_from_turn_id AS "forkedFromTurnId", latest_turn_id AS "latestTurnId", created_at AS "createdAt", updated_at AS "updatedAt", diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index d468838d5d4..b6e2b27becd 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -45,6 +45,7 @@ import Migration0029 from "./Migrations/029_ProjectionThreadDetailOrderingIndexe import Migration0030 from "./Migrations/030_ProjectionThreadShellArchiveIndexes.ts"; import Migration0031 from "./Migrations/031_AuthAuthorizationScopes.ts"; import Migration0032 from "./Migrations/032_AuthPairingProofKeyThumbprint.ts"; +import Migration0033 from "./Migrations/033_ProjectionThreadForkLineage.ts"; /** * Migration loader with all migrations defined inline. @@ -89,6 +90,7 @@ export const migrationEntries = [ [30, "ProjectionThreadShellArchiveIndexes", Migration0030], [31, "AuthAuthorizationScopes", Migration0031], [32, "AuthPairingProofKeyThumbprint", Migration0032], + [33, "ProjectionThreadForkLineage", Migration0033], ] as const; export const makeMigrationLoader = (throughId?: number) => diff --git a/apps/server/src/persistence/Migrations/033_ProjectionThreadForkLineage.test.ts b/apps/server/src/persistence/Migrations/033_ProjectionThreadForkLineage.test.ts new file mode 100644 index 00000000000..e0bbd6e0e1b --- /dev/null +++ b/apps/server/src/persistence/Migrations/033_ProjectionThreadForkLineage.test.ts @@ -0,0 +1,29 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; +import * as NodeSqliteClient from "../NodeSqliteClient.ts"; + +const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); + +layer("033_ProjectionThreadForkLineage", (it) => { + it.effect("adds nullable fork lineage columns and the parent lookup index", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* runMigrations({ toMigrationInclusive: 32 }); + const before = yield* sql<{ readonly name: string }>`PRAGMA table_info(projection_threads)`; + assert.isFalse(before.some((column) => column.name === "forked_from_thread_id")); + + yield* runMigrations({ toMigrationInclusive: 33 }); + const columns = yield* sql<{ readonly name: string }>`PRAGMA table_info(projection_threads)`; + const indexes = yield* sql<{ readonly name: string }>`PRAGMA index_list(projection_threads)`; + + assert.isTrue(columns.some((column) => column.name === "forked_from_thread_id")); + assert.isTrue(columns.some((column) => column.name === "forked_from_turn_id")); + assert.isTrue(indexes.some((index) => index.name === "idx_projection_threads_forked_from")); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/033_ProjectionThreadForkLineage.ts b/apps/server/src/persistence/Migrations/033_ProjectionThreadForkLineage.ts new file mode 100644 index 00000000000..e75a95a6bd4 --- /dev/null +++ b/apps/server/src/persistence/Migrations/033_ProjectionThreadForkLineage.ts @@ -0,0 +1,13 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + yield* sql`ALTER TABLE projection_threads ADD COLUMN forked_from_thread_id TEXT`; + yield* sql`ALTER TABLE projection_threads ADD COLUMN forked_from_turn_id TEXT`; + yield* sql` + CREATE INDEX IF NOT EXISTS idx_projection_threads_forked_from + ON projection_threads(forked_from_thread_id, created_at) + `; +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreads.ts b/apps/server/src/persistence/Services/ProjectionThreads.ts index 44fdc147a4a..04c99ea957d 100644 --- a/apps/server/src/persistence/Services/ProjectionThreads.ts +++ b/apps/server/src/persistence/Services/ProjectionThreads.ts @@ -32,6 +32,8 @@ export const ProjectionThread = Schema.Struct({ interactionMode: ProviderInteractionMode, branch: Schema.NullOr(Schema.String), worktreePath: Schema.NullOr(Schema.String), + forkedFromThreadId: Schema.optional(Schema.NullOr(ThreadId)), + forkedFromTurnId: Schema.optional(Schema.NullOr(TurnId)), latestTurnId: Schema.NullOr(TurnId), createdAt: IsoDateTime, updatedAt: IsoDateTime, diff --git a/apps/server/src/provider/CodexDeveloperInstructions.ts b/apps/server/src/provider/CodexDeveloperInstructions.ts index 35ffd1a4756..c54664753da 100644 --- a/apps/server/src/provider/CodexDeveloperInstructions.ts +++ b/apps/server/src/provider/CodexDeveloperInstructions.ts @@ -1,6 +1,6 @@ import type { ProviderInteractionMode } from "@t3tools/contracts"; -const T3_CODE_BROWSER_TOOL_INSTRUCTIONS = ` +const T3_CODE_MCP_TOOL_INSTRUCTIONS = ` ## T3 Code collaborative browser @@ -9,6 +9,10 @@ You are running inside T3 Code. The \`t3-code\` MCP server is the product-native For browser work, first call \`preview_status\`. If no automation-capable preview is attached, call \`preview_open\` before concluding that the browser is unavailable. Then use \`preview_navigate\`, \`preview_snapshot\`, and the focused interaction tools. Prefer snapshot-provided locators over coordinates. Do not switch to global browser skills, Chrome, Node REPL browser automation, standalone Playwright, or agent-browser merely because the preview is initially closed or a first call fails. Use an alternative browser system only when the T3 preview tools are absent, the user explicitly requests another browser, or \`preview_open\` returns an explicit unsupported/unavailable error. A failed T3 preview tool call should be inspected and retried with corrected arguments when the error is actionable. + +## T3 Code thread coordination + +When the \`t3-code\` MCP server exposes \`thread_*\` tools, use them to coordinate project threads. Use \`thread_fork\` when the new thread must inherit conversation history, \`thread_create\` for an independent sibling, \`thread_send\` to delegate work to another thread, \`thread_list\` to inspect active project threads, and \`thread_archive\` to retire an inactive thread. A fork created from the current thread opens as a side chat in the T3 Code interface. `; export const CODEX_PLAN_MODE_DEVELOPER_INSTRUCTIONS = `# Plan Mode (Conversational) @@ -131,7 +135,7 @@ plan content should be human and agent digestible. The final plan must be plan-o Do not ask "should I proceed?" in the final output. The user can easily switch out of Plan mode and request implementation if you have included a \`\` block in your response. Alternatively, they can decide to stay in Plan mode and continue refining the plan. Only produce at most one \`\` block per turn, and only when you are presenting a complete spec. -${T3_CODE_BROWSER_TOOL_INSTRUCTIONS} +${T3_CODE_MCP_TOOL_INSTRUCTIONS} `; export const CODEX_DEFAULT_MODE_DEVELOPER_INSTRUCTIONS = `# Collaboration Mode: Default @@ -145,7 +149,7 @@ Your active mode changes only when new developer instructions with a different \ The \`request_user_input\` tool is unavailable in Default mode. If you call it while in Default mode, it will return an error. In Default mode, strongly prefer making reasonable assumptions and executing the user's request rather than stopping to ask questions. If you absolutely must ask a question because the answer cannot be discovered from local context and a reasonable assumption would be risky, ask the user directly with a concise plain-text question. Never write a multiple choice question as a textual assistant message. -${T3_CODE_BROWSER_TOOL_INSTRUCTIONS} +${T3_CODE_MCP_TOOL_INSTRUCTIONS} `; export interface CodexRuntimeInfo { diff --git a/apps/server/src/provider/Layers/CodexAdapter.ts b/apps/server/src/provider/Layers/CodexAdapter.ts index 38a5887cdc3..13b9da7ae8c 100644 --- a/apps/server/src/provider/Layers/CodexAdapter.ts +++ b/apps/server/src/provider/Layers/CodexAdapter.ts @@ -1406,6 +1406,16 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( ...(isCodexResumeCursorSchema(input.resumeCursor) ? { resumeCursor: input.resumeCursor } : {}), + ...(input.forkFrom !== undefined && isCodexResumeCursorSchema(input.forkFrom.resumeCursor) + ? { + forkFrom: { + providerThreadId: input.forkFrom.resumeCursor.threadId, + ...(input.forkFrom.sourceTurnId !== undefined + ? { sourceTurnId: input.forkFrom.sourceTurnId } + : {}), + }, + } + : {}), runtimeMode: input.runtimeMode, ...(input.modelSelection?.instanceId === boundInstanceId ? { model: input.modelSelection.model } @@ -1703,6 +1713,7 @@ export const makeCodexAdapter = Effect.fn("makeCodexAdapter")(function* ( provider: PROVIDER, capabilities: { sessionModelSwitch: "in-session", + sessionFork: "native", }, startSession, sendTurn, diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index 119fa36303a..c40cc579295 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts @@ -275,6 +275,9 @@ describe("T3 browser developer instructions", () => { NodeAssert.match(instructions, /preview_status/); NodeAssert.match(instructions, /preview_open/); NodeAssert.match(instructions, /Do not switch to global browser skills/); + NodeAssert.match(instructions, /thread_fork/); + NodeAssert.match(instructions, /thread_create/); + NodeAssert.match(instructions, /side chat/); } }); }); @@ -365,12 +368,64 @@ describe("isRecoverableThreadResumeError", () => { }); describe("openCodexThread", () => { + it.effect("forks the native Codex thread at the selected turn", () => + Effect.gen(function* () { + const calls: Array<{ + method: "thread/start" | "thread/resume" | "thread/fork"; + payload: unknown; + }> = []; + const forked = makeThreadOpenResponse("forked-provider-thread"); + const client = { + request: ( + method: M, + payload: CodexRpc.ClientRequestParamsByMethod[M], + ) => { + calls.push({ method, payload }); + return Effect.succeed(forked as CodexRpc.ClientRequestResponsesByMethod[M]); + }, + }; + + const opened = yield* openCodexThread({ + client, + threadId: ThreadId.make("t3-fork-thread"), + runtimeMode: "full-access", + cwd: "/tmp/project", + requestedModel: "gpt-5.4", + serviceTier: undefined, + resumeThreadId: "ignored-target-resume-thread", + forkFrom: { + providerThreadId: "source-provider-thread", + sourceTurnId: "turn-2" as import("@t3tools/contracts").TurnId, + }, + }); + + NodeAssert.equal(opened.thread.id, "forked-provider-thread"); + NodeAssert.deepStrictEqual(calls, [ + { + method: "thread/fork", + payload: { + threadId: "source-provider-thread", + cwd: "/tmp/project", + approvalPolicy: "never", + sandbox: "danger-full-access", + model: "gpt-5.4", + lastTurnId: "turn-2", + threadSource: "t3-code", + }, + }, + ]); + }), + ); + it.effect("falls back to thread/start when resume fails recoverably", () => Effect.gen(function* () { - const calls: Array<{ method: "thread/start" | "thread/resume"; payload: unknown }> = []; + const calls: Array<{ + method: "thread/start" | "thread/resume" | "thread/fork"; + payload: unknown; + }> = []; const started = makeThreadOpenResponse("fresh-thread"); const client = { - request: ( + request: ( method: M, payload: CodexRpc.ClientRequestParamsByMethod[M], ) => { @@ -408,7 +463,7 @@ describe("openCodexThread", () => { it.effect("propagates non-recoverable resume failures", () => Effect.gen(function* () { const client = { - request: ( + request: ( method: M, _payload: CodexRpc.ClientRequestParamsByMethod[M], ) => { diff --git a/apps/server/src/provider/Layers/CodexSessionRuntime.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.ts index 5a81e915e34..6dde1ca8458 100644 --- a/apps/server/src/provider/Layers/CodexSessionRuntime.ts +++ b/apps/server/src/provider/Layers/CodexSessionRuntime.ts @@ -105,6 +105,10 @@ export interface CodexSessionRuntimeOptions { readonly model?: string; readonly serviceTier?: CodexServiceTier | undefined; readonly resumeCursor?: CodexResumeCursor; + readonly forkFrom?: { + readonly providerThreadId: string; + readonly sourceTurnId?: TurnId; + }; readonly appServerArgs?: ReadonlyArray; } @@ -428,9 +432,10 @@ export function isRecoverableThreadResumeError(error: unknown): boolean { type CodexThreadOpenResponse = | CodexRpc.ClientRequestResponsesByMethod["thread/start"] - | CodexRpc.ClientRequestResponsesByMethod["thread/resume"]; + | CodexRpc.ClientRequestResponsesByMethod["thread/resume"] + | CodexRpc.ClientRequestResponsesByMethod["thread/fork"]; -type CodexThreadOpenMethod = "thread/start" | "thread/resume"; +type CodexThreadOpenMethod = "thread/start" | "thread/resume" | "thread/fork"; interface CodexThreadOpenClient { readonly request: ( @@ -447,6 +452,12 @@ export const openCodexThread = (input: { readonly requestedModel: string | undefined; readonly serviceTier: CodexServiceTier | undefined; readonly resumeThreadId: string | undefined; + readonly forkFrom?: + | { + readonly providerThreadId: string; + readonly sourceTurnId?: TurnId; + } + | undefined; }): Effect.Effect => { const resumeThreadId = input.resumeThreadId; const startParams = buildThreadStartParams({ @@ -456,6 +467,22 @@ export const openCodexThread = (input: { serviceTier: input.serviceTier, }); + if (input.forkFrom !== undefined) { + const config = runtimeModeToThreadConfig(input.runtimeMode); + return input.client.request("thread/fork", { + threadId: input.forkFrom.providerThreadId, + cwd: input.cwd, + approvalPolicy: config.approvalPolicy, + sandbox: config.sandbox, + ...(startParams.model ? { model: startParams.model } : {}), + ...(startParams.serviceTier ? { serviceTier: startParams.serviceTier } : {}), + ...(input.forkFrom.sourceTurnId !== undefined + ? { lastTurnId: input.forkFrom.sourceTurnId } + : {}), + threadSource: "t3-code", + }); + } + if (resumeThreadId === undefined) { return input.client.request("thread/start", startParams); } @@ -1212,6 +1239,7 @@ export const makeCodexSessionRuntime = ( requestedModel, serviceTier: options.serviceTier, resumeThreadId: readResumeCursorThreadId(options.resumeCursor), + forkFrom: options.forkFrom, }); const providerThreadId = opened.thread.id; diff --git a/apps/server/src/provider/Layers/ProviderService.ts b/apps/server/src/provider/Layers/ProviderService.ts index 2eaaeb8ce3c..287349959b7 100644 --- a/apps/server/src/provider/Layers/ProviderService.ts +++ b/apps/server/src/provider/Layers/ProviderService.ts @@ -559,12 +559,61 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( `Provider instance '${resolvedInstanceId}' is disabled in T3 Code settings.`, ); } + const adapter = yield* registry.getByInstance(resolvedInstanceId); const persistedBinding = Option.getOrUndefined(yield* directory.getBinding(threadId)); + const requestedForkFrom = input.forkFrom; + const effectiveForkFrom = + requestedForkFrom === undefined + ? undefined + : yield* Effect.gen(function* () { + const sourceBinding = Option.getOrUndefined( + yield* directory.getBinding(requestedForkFrom.threadId), + ); + if (!sourceBinding) { + if (requestedForkFrom.sourceTurnId === undefined) { + return undefined; + } + return yield* toValidationError( + "ProviderService.startSession", + `Cannot fork thread '${requestedForkFrom.threadId}' because it has no persisted provider binding.`, + ); + } + if (adapter.capabilities.sessionFork !== "native") { + return yield* toValidationError( + "ProviderService.startSession", + `Provider '${resolvedProvider}' does not support native thread forks.`, + ); + } + if (sourceBinding.providerInstanceId !== resolvedInstanceId) { + return yield* toValidationError( + "ProviderService.startSession", + `Cannot fork thread '${requestedForkFrom.threadId}' into provider instance '${resolvedInstanceId}' because its provider resume state belongs to a different instance.`, + ); + } + if ( + sourceBinding.resumeCursor === null || + sourceBinding.resumeCursor === undefined + ) { + return yield* toValidationError( + "ProviderService.startSession", + `Cannot fork thread '${requestedForkFrom.threadId}' because no provider resume state is persisted.`, + ); + } + return { + threadId: requestedForkFrom.threadId, + ...(requestedForkFrom.sourceTurnId !== undefined + ? { sourceTurnId: requestedForkFrom.sourceTurnId } + : {}), + resumeCursor: sourceBinding.resumeCursor, + }; + }); const effectiveResumeCursor = - input.resumeCursor ?? - (persistedBinding?.providerInstanceId === resolvedInstanceId - ? persistedBinding.resumeCursor - : undefined); + effectiveForkFrom === undefined + ? (input.resumeCursor ?? + (persistedBinding?.providerInstanceId === resolvedInstanceId + ? persistedBinding.resumeCursor + : undefined)) + : undefined; const effectiveCwd = input.cwd ?? (persistedBinding?.providerInstanceId === resolvedInstanceId @@ -580,6 +629,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( ? "persisted" : "none", "provider.resume_cursor.present": effectiveResumeCursor !== undefined, + "provider.fork.present": effectiveForkFrom !== undefined, "provider.cwd.source": input.cwd !== undefined ? "request" @@ -589,7 +639,6 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( : "none", "provider.cwd.effective": effectiveCwd ?? "", }); - const adapter = yield* registry.getByInstance(resolvedInstanceId); yield* prepareMcpSession(threadId, resolvedInstanceId); const session = yield* adapter .startSession({ @@ -597,6 +646,7 @@ const makeProviderService = Effect.fn("makeProviderService")(function* ( providerInstanceId: resolvedInstanceId, ...(effectiveCwd !== undefined ? { cwd: effectiveCwd } : {}), ...(effectiveResumeCursor !== undefined ? { resumeCursor: effectiveResumeCursor } : {}), + ...(effectiveForkFrom !== undefined ? { forkFrom: effectiveForkFrom } : {}), }) .pipe(Effect.onError(() => clearMcpSession(threadId))); diff --git a/apps/server/src/provider/Services/ProviderAdapter.ts b/apps/server/src/provider/Services/ProviderAdapter.ts index 01eeae7b7bd..941e40af3e7 100644 --- a/apps/server/src/provider/Services/ProviderAdapter.ts +++ b/apps/server/src/provider/Services/ProviderAdapter.ts @@ -24,12 +24,14 @@ import type * as Effect from "effect/Effect"; import type * as Stream from "effect/Stream"; export type ProviderSessionModelSwitchMode = "in-session" | "unsupported"; +export type ProviderSessionForkMode = "native" | "unsupported"; export interface ProviderAdapterCapabilities { /** * Declares whether changing the model on an existing session is supported. */ readonly sessionModelSwitch: ProviderSessionModelSwitchMode; + readonly sessionFork?: ProviderSessionForkMode; } export interface ProviderThreadTurnSnapshot { diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index c296c717066..1d91aebbdbe 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -200,6 +200,7 @@ import { useThread, useThreadProposedPlans, useThreadRefs, + useThreadShells, } from "../state/entities"; import { environmentShell } from "../state/shell"; import { ChatComposer, type ChatComposerHandle } from "./chat/ChatComposer"; @@ -423,6 +424,7 @@ type ChatViewProps = onDiffPanelOpen?: () => void; reserveTitleBarControlInset?: boolean; forceExpandedMobileComposer?: boolean; + embedded?: boolean; routeKind: "server"; draftId?: never; } @@ -432,6 +434,7 @@ type ChatViewProps = onDiffPanelOpen?: () => void; reserveTitleBarControlInset?: boolean; forceExpandedMobileComposer?: boolean; + embedded?: boolean; routeKind: "draft"; draftId: DraftId; }; @@ -1084,6 +1087,7 @@ function ChatViewContent(props: ChatViewProps) { onDiffPanelOpen, reserveTitleBarControlInset = true, forceExpandedMobileComposer = false, + embedded = false, } = props; const draftId = routeKind === "draft" ? props.draftId : null; const routeThreadRef = useMemo( @@ -1099,6 +1103,7 @@ function ChatViewContent(props: ChatViewProps) { const writeTerminal = useAtomCommand(terminalEnvironment.write, "terminal write"); const closeTerminalMutation = useAtomCommand(terminalEnvironment.close, "terminal close"); const createThread = useAtomCommand(threadEnvironment.create, { reportFailure: false }); + const forkThread = useAtomCommand(threadEnvironment.fork, { reportFailure: false }); const deleteThread = useAtomCommand(threadEnvironment.delete, { reportFailure: false }); const updateThreadMetadata = useAtomCommand(threadEnvironment.updateMetadata, { reportFailure: false, @@ -1208,6 +1213,7 @@ function ChatViewContent(props: ChatViewProps) { >({}); const [isConnecting, _setIsConnecting] = useState(false); const [isRevertingCheckpoint, setIsRevertingCheckpoint] = useState(false); + const [isForkingToSide, setIsForkingToSide] = useState(false); const [maximizedRightPanelThreadKey, setMaximizedRightPanelThreadKey] = useState( null, ); @@ -1253,6 +1259,8 @@ function ChatViewContent(props: ChatViewProps) { const attachmentPreviewHandoffByMessageIdRef = useRef>({}); const attachmentPreviewPromotionInFlightByMessageIdRef = useRef>({}); const sendInFlightRef = useRef(false); + const knownThreadIdsRef = useRef>(new Set()); + const mountedAtRef = useRef(Date.now()); const terminalUiOpenByThreadRef = useRef>({}); useLayoutEffect(() => { @@ -1293,6 +1301,7 @@ function ChatViewContent(props: ChatViewProps) { const storeSetActiveTerminal = useTerminalUiStateStore((s) => s.setActiveTerminal); const storeCloseTerminal = useTerminalUiStateStore((s) => s.closeTerminal); const serverThreadRefs = useThreadRefs(); + const threadShells = useThreadShells(); const serverThreadKeys = useMemo(() => serverThreadRefs.map(scopedThreadKey), [serverThreadRefs]); const draftThreadsByThreadKey = useComposerDraftStore((store) => store.draftThreadsByThreadKey); const draftThreadKeys = useMemo( @@ -1422,6 +1431,26 @@ function ChatViewContent(props: ChatViewProps) { [activeThread], ); const activeThreadKey = activeThreadRef ? scopedThreadKey(activeThreadRef) : null; + useEffect(() => { + const currentIds = new Set( + threadShells.map((thread) => `${thread.environmentId}:${thread.id}`), + ); + if (!embedded && activeThread && activeThreadRef) { + for (const thread of threadShells) { + const key = `${thread.environmentId}:${thread.id}`; + if ( + !knownThreadIdsRef.current.has(key) && + thread.environmentId === activeThread.environmentId && + thread.projectId === activeThread.projectId && + thread.forkedFrom?.threadId === activeThread.id && + Date.parse(thread.createdAt) >= mountedAtRef.current - 2_000 + ) { + useRightPanelStore.getState().openThread(activeThreadRef, thread.id, thread.title); + } + } + } + knownThreadIdsRef.current = currentIds; + }, [activeThread, activeThreadRef, embedded, threadShells]); const [timelineAnchor, setTimelineAnchor] = useState<{ readonly threadKey: string | null; readonly messageId: MessageId | null; @@ -1453,7 +1482,7 @@ function ChatViewContent(props: ChatViewProps) { [rightPanelState.surfaces], ); const previewPanelOpen = activeRightPanelKind === "preview" && isPreviewSupportedInRuntime(); - const rightPanelOpen = rightPanelState.isOpen; + const rightPanelOpen = !embedded && rightPanelState.isOpen; const canMaximizeRightPanel = rightPanelOpen && !shouldUsePlanSidebarSheet; const rightPanelMaximized = canMaximizeRightPanel && maximizedRightPanelThreadKey === routeThreadKey; @@ -4758,6 +4787,40 @@ function ChatViewContent(props: ChatViewProps) { ], ); + const onForkToSide = useCallback(async () => { + if (!activeThread || !activeThreadRef || !isServerThread || isForkingToSide) return; + + const nextThreadId = newThreadId(); + const title = truncate(`${activeThread.title} (fork)`); + setIsForkingToSide(true); + const result = await forkThread({ + environmentId: activeThread.environmentId, + input: { + threadId: nextThreadId, + sourceThreadId: activeThread.id, + title, + createdAt: new Date().toISOString(), + }, + }); + setIsForkingToSide(false); + + if (result._tag === "Failure") { + if (!isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + toastManager.add( + stackedThreadToast({ + type: "error", + title: "Could not fork side chat", + description: error instanceof Error ? error.message : "The thread could not be forked.", + }), + ); + } + return; + } + + useRightPanelStore.getState().openThread(activeThreadRef, nextThreadId, title); + }, [activeThread, activeThreadRef, forkThread, isForkingToSide, isServerThread]); + const onImplementPlanInNewThread = useCallback(async () => { if ( !activeThread || @@ -5113,7 +5176,15 @@ function ChatViewContent(props: ChatViewProps) { ); const rightPanelContent = activeThreadRef ? ( - activeRightPanelSurface?.kind === "preview" ? ( + activeRightPanelSurface?.kind === "thread" ? ( + + ) : activeRightPanelSurface?.kind === "preview" ? ( {/* Top bar */} -
- {!rightPanelOpen ? panelLayoutControls : null} - -
+ {!embedded ? ( +
+ {!rightPanelOpen ? panelLayoutControls : null} + +
+ ) : null} {/* Error banner */} @@ -5509,24 +5583,30 @@ function ChatViewContent(props: ChatViewProps) { {/* end horizontal flex container */} - {mountedTerminalThreadRefs.map(({ key: mountedThreadKey, threadRef: mountedThreadRef }) => ( - - ))} + {!embedded + ? mountedTerminalThreadRefs.map( + ({ key: mountedThreadKey, threadRef: mountedThreadRef }) => ( + + ), + ) + : null} {!shouldUsePlanSidebarSheet && rightPanelOpen && activeThreadRef ? ( diff --git a/apps/web/src/components/RightPanelTabs.tsx b/apps/web/src/components/RightPanelTabs.tsx index f5bc9880d74..960fd08fcab 100644 --- a/apps/web/src/components/RightPanelTabs.tsx +++ b/apps/web/src/components/RightPanelTabs.tsx @@ -1,6 +1,15 @@ import type { ContextMenuItem, PreviewSessionSnapshot } from "@t3tools/contracts"; import { getTerminalLabel } from "@t3tools/shared/terminalLabels"; -import { ClipboardList, FileDiff, Files, Globe2, Plus, TerminalSquare, X } from "lucide-react"; +import { + ClipboardList, + FileDiff, + Files, + GitFork, + Globe2, + Plus, + TerminalSquare, + X, +} from "lucide-react"; import { type MouseEvent as ReactMouseEvent, type ReactElement, @@ -205,6 +214,8 @@ function surfaceTitle( ); case "plan": return "Plan"; + case "thread": + return surface.title; case "preview": { const snapshot = surface.resourceId ? sessions[surface.resourceId] : null; if (!snapshot || snapshot.navStatus._tag === "Idle") return "Browser"; @@ -266,6 +277,8 @@ function SurfaceIcon({ return ; case "plan": return ; + case "thread": + return ; } } diff --git a/apps/web/src/components/chat/ChatHeader.tsx b/apps/web/src/components/chat/ChatHeader.tsx index ef3ec863d0b..10e2b2fcccf 100644 --- a/apps/web/src/components/chat/ChatHeader.tsx +++ b/apps/web/src/components/chat/ChatHeader.tsx @@ -7,6 +7,7 @@ import { } from "@t3tools/contracts"; import { scopeThreadRef } from "@t3tools/client-runtime/environment"; import { memo } from "react"; +import { GitFork } from "lucide-react"; import GitActionsControl from "../GitActionsControl"; import { type DraftId } from "~/composerDraftStore"; import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; @@ -38,6 +39,8 @@ interface ChatHeaderProps { input: NewProjectScriptInput, ) => Promise; onDeleteProjectScript: (scriptId: string) => Promise; + onForkToSide?: (() => void) | undefined; + forkToSideDisabled?: boolean | undefined; } export function shouldShowOpenInPicker(input: { @@ -69,6 +72,8 @@ export const ChatHeader = memo(function ChatHeader({ onAddProjectScript, onUpdateProjectScript, onDeleteProjectScript, + onForkToSide, + forkToSideDisabled = false, }: ChatHeaderProps) { const primaryEnvironmentId = usePrimaryEnvironmentId(); const showOpenInPicker = shouldShowOpenInPicker({ @@ -100,6 +105,24 @@ export const ChatHeader = memo(function ChatHeader({ rightPanelOpen ? "pr-0" : "pr-16", )} > + {onForkToSide ? ( + + + + + } + /> + Fork to side chat + + ) : null} {activeProjectScripts && ( { }); }); + it("opens a forked thread as a peer surface and refreshes its title", () => { + const forkId = ThreadId.make("thread-fork"); + useRightPanelStore.getState().openThread(refA, forkId, "Initial fork"); + useRightPanelStore.getState().openThread(refA, forkId, "Renamed fork"); + + expect(selectThreadRightPanelState(useRightPanelStore.getState().byThreadKey, refA)).toEqual({ + isOpen: true, + activeSurfaceId: "thread:thread-fork", + surfaces: [ + { + id: "thread:thread-fork", + kind: "thread", + threadId: forkId, + title: "Renamed fork", + }, + ], + }); + }); + it("tracks one surface per terminal session", () => { useRightPanelStore.getState().openTerminal(refA, "term-1"); useRightPanelStore.getState().openTerminal(refA, "term-2"); diff --git a/apps/web/src/rightPanelStore.ts b/apps/web/src/rightPanelStore.ts index 70d163306cc..2e6f7c0c13a 100644 --- a/apps/web/src/rightPanelStore.ts +++ b/apps/web/src/rightPanelStore.ts @@ -8,13 +8,21 @@ * workspace paths, and diff/plan/files remain singleton surfaces. */ import { scopedThreadKey } from "@t3tools/client-runtime/environment"; -import type { ScopedThreadRef } from "@t3tools/contracts"; +import type { ScopedThreadRef, ThreadId } from "@t3tools/contracts"; import { create } from "zustand"; import { createJSONStorage, persist } from "zustand/middleware"; import { resolveStorage } from "./lib/storage"; -export const RIGHT_PANEL_KINDS = ["plan", "diff", "files", "file", "preview", "terminal"] as const; +export const RIGHT_PANEL_KINDS = [ + "plan", + "diff", + "files", + "file", + "preview", + "terminal", + "thread", +] as const; export type RightPanelKind = (typeof RIGHT_PANEL_KINDS)[number]; export type RightPanelSurface = @@ -37,10 +45,11 @@ export type RightPanelSurface = revealLine: number | null; revealRequestId: number; } - | { id: "plan"; kind: "plan" }; + | { id: "plan"; kind: "plan" } + | { id: `thread:${string}`; kind: "thread"; threadId: ThreadId; title: string }; const RIGHT_PANEL_STORAGE_KEY = "t3code:right-panel-state:v2"; -const RIGHT_PANEL_STORAGE_VERSION = 7; +const RIGHT_PANEL_STORAGE_VERSION = 8; export interface ThreadRightPanelState { isOpen: boolean; @@ -50,7 +59,11 @@ export interface ThreadRightPanelState { interface RightPanelStoreState { byThreadKey: Record; - open: (ref: ScopedThreadRef, kind: Exclude) => void; + open: ( + ref: ScopedThreadRef, + kind: Exclude, + ) => void; + openThread: (ref: ScopedThreadRef, threadId: ThreadId, title: string) => void; openBrowser: (ref: ScopedThreadRef, tabId: string | null) => void; openFile: (ref: ScopedThreadRef, relativePath: string, line?: number) => void; openTerminal: (ref: ScopedThreadRef, terminalId: string) => void; @@ -72,7 +85,10 @@ interface RightPanelStoreState { show: (ref: ScopedThreadRef) => void; close: (ref: ScopedThreadRef) => void; toggleVisibility: (ref: ScopedThreadRef) => void; - toggle: (ref: ScopedThreadRef, kind: Exclude) => void; + toggle: ( + ref: ScopedThreadRef, + kind: Exclude, + ) => void; removeThread: (ref: ScopedThreadRef) => void; } @@ -83,7 +99,7 @@ const EMPTY_THREAD_STATE: ThreadRightPanelState = { }; const singletonSurface = ( - kind: Exclude, + kind: Exclude, ): RightPanelSurface => { switch (kind) { case "diff": @@ -120,6 +136,13 @@ const terminalSurface = (terminalId: string): RightPanelSurface => ({ activeTerminalId: terminalId, }); +const threadSurface = (threadId: ThreadId, title: string): RightPanelSurface => ({ + id: `thread:${threadId}`, + kind: "thread", + threadId, + title, +}); + const upsertSurface = ( current: ThreadRightPanelState, surface: RightPanelSurface, @@ -259,6 +282,20 @@ export const useRightPanelStore = create()( return upsertSurface({ ...current, surfaces: withoutPlaceholder }, surface); }), })), + openThread: (ref, threadId, title) => + set((state) => ({ + byThreadKey: updateThread(state.byThreadKey, scopedThreadKey(ref), (current) => { + const surface = threadSurface(threadId, title); + const existing = current.surfaces.some((entry) => entry.id === surface.id); + return { + isOpen: true, + activeSurfaceId: surface.id, + surfaces: existing + ? current.surfaces.map((entry) => (entry.id === surface.id ? surface : entry)) + : [...current.surfaces, surface], + }; + }), + })), openFile: (ref, relativePath, line) => set((state) => ({ byThreadKey: updateThread(state.byThreadKey, scopedThreadKey(ref), (current) => { diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index a0c3cbe771f..1eb32753139 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -32,6 +32,7 @@ export type CreateProjectInput = CommandInput<"project.create">; export type UpdateProjectInput = CommandInput<"project.meta.update">; export type DeleteProjectInput = CommandInput<"project.delete">; export type CreateThreadInput = CommandInput<"thread.create">; +export type ForkThreadInput = CommandInput<"thread.fork">; export type DeleteThreadInput = CommandInput<"thread.delete">; export type ArchiveThreadInput = CommandInput<"thread.archive">; export type UnarchiveThreadInput = CommandInput<"thread.unarchive">; @@ -123,6 +124,18 @@ export const createThread: (input: CreateThreadInput) => CommandEffect = Effect. }); }); +export const forkThread: (input: ForkThreadInput) => CommandEffect = Effect.fn( + "EnvironmentCommands.forkThread", +)(function* (input) { + const metadata = yield* timestampedCommandMetadata(input); + return yield* dispatch({ + ...input, + type: "thread.fork", + commandId: metadata.commandId, + createdAt: metadata.createdAt, + }); +}); + export const deleteThread: (input: DeleteThreadInput) => CommandEffect = Effect.fn( "EnvironmentCommands.deleteThread", )(function* (input) { diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index aab5110e9cf..c73541ff101 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -5,6 +5,7 @@ import { createAtomCommandScheduler, createEnvironmentCommand } from "./runtime. import { type ArchiveThreadInput, type CreateThreadInput, + type ForkThreadInput, type DeleteThreadInput, type InterruptThreadTurnInput, type RespondToThreadApprovalInput, @@ -18,6 +19,7 @@ import { type UpdateThreadMetadataInput, archiveThread, createThread, + forkThread, deleteThread, interruptThreadTurn, respondToThreadApproval, @@ -35,6 +37,7 @@ import type { EnvironmentRegistry } from "../connection/registry.ts"; export type { ArchiveThreadInput, CreateThreadInput, + ForkThreadInput, DeleteThreadInput, InterruptThreadTurnInput, RespondToThreadApprovalInput, @@ -64,6 +67,12 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), + fork: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:fork", + execute: (input: ForkThreadInput) => forkThread(input), + scheduler, + concurrency, + }), delete: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:delete", execute: (input: DeleteThreadInput) => deleteThread(input), diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index c0c9c62d080..60acfb712b9 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -341,6 +341,12 @@ export const OrchestrationLatestTurn = Schema.Struct({ }); export type OrchestrationLatestTurn = typeof OrchestrationLatestTurn.Type; +export const ThreadForkReference = Schema.Struct({ + threadId: ThreadId, + turnId: Schema.NullOr(TurnId), +}); +export type ThreadForkReference = typeof ThreadForkReference.Type; + export const OrchestrationThread = Schema.Struct({ id: ThreadId, projectId: ProjectId, @@ -352,6 +358,7 @@ export const OrchestrationThread = Schema.Struct({ ), branch: Schema.NullOr(TrimmedNonEmptyString), worktreePath: Schema.NullOr(TrimmedNonEmptyString), + forkedFrom: Schema.optional(Schema.NullOr(ThreadForkReference)), latestTurn: Schema.NullOr(OrchestrationLatestTurn), createdAt: IsoDateTime, updatedAt: IsoDateTime, @@ -398,6 +405,7 @@ export const OrchestrationThreadShell = Schema.Struct({ ), branch: Schema.NullOr(TrimmedNonEmptyString), worktreePath: Schema.NullOr(TrimmedNonEmptyString), + forkedFrom: Schema.optional(Schema.NullOr(ThreadForkReference)), latestTurn: Schema.NullOr(OrchestrationLatestTurn), createdAt: IsoDateTime, updatedAt: IsoDateTime, @@ -540,6 +548,16 @@ const ThreadCreateCommand = Schema.Struct({ createdAt: IsoDateTime, }); +const ThreadForkCommand = Schema.Struct({ + type: Schema.Literal("thread.fork"), + commandId: CommandId, + threadId: ThreadId, + sourceThreadId: ThreadId, + sourceTurnId: Schema.optional(TurnId), + title: TrimmedNonEmptyString, + createdAt: IsoDateTime, +}); + const ThreadDeleteCommand = Schema.Struct({ type: Schema.Literal("thread.delete"), commandId: CommandId, @@ -697,6 +715,7 @@ const DispatchableClientOrchestrationCommand = Schema.Union([ ProjectMetaUpdateCommand, ProjectDeleteCommand, ThreadCreateCommand, + ThreadForkCommand, ThreadDeleteCommand, ThreadArchiveCommand, ThreadUnarchiveCommand, @@ -718,6 +737,7 @@ export const ClientOrchestrationCommand = Schema.Union([ ProjectMetaUpdateCommand, ProjectDeleteCommand, ThreadCreateCommand, + ThreadForkCommand, ThreadDeleteCommand, ThreadArchiveCommand, ThreadUnarchiveCommand, @@ -820,6 +840,7 @@ export const OrchestrationEventType = Schema.Literals([ "project.meta-updated", "project.deleted", "thread.created", + "thread.forked", "thread.deleted", "thread.archived", "thread.unarchived", @@ -886,6 +907,21 @@ export const ThreadCreatedPayload = Schema.Struct({ updatedAt: IsoDateTime, }); +export const ThreadForkedPayload = Schema.Struct({ + threadId: ThreadId, + projectId: ProjectId, + title: TrimmedNonEmptyString, + modelSelection: ModelSelection, + runtimeMode: RuntimeMode, + interactionMode: ProviderInteractionMode, + branch: Schema.NullOr(TrimmedNonEmptyString), + worktreePath: Schema.NullOr(TrimmedNonEmptyString), + forkedFrom: ThreadForkReference, + inheritedMessages: Schema.Array(OrchestrationMessage), + createdAt: IsoDateTime, + updatedAt: IsoDateTime, +}); + export const ThreadDeletedPayload = Schema.Struct({ threadId: ThreadId, deletedAt: IsoDateTime, @@ -1054,6 +1090,11 @@ export const OrchestrationEvent = Schema.Union([ type: Schema.Literal("thread.created"), payload: ThreadCreatedPayload, }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.forked"), + payload: ThreadForkedPayload, + }), Schema.Struct({ ...EventBaseFields, type: Schema.Literal("thread.deleted"), diff --git a/packages/contracts/src/previewAutomation.ts b/packages/contracts/src/previewAutomation.ts index 6431dd0dcfd..04baf477df6 100644 --- a/packages/contracts/src/previewAutomation.ts +++ b/packages/contracts/src/previewAutomation.ts @@ -608,7 +608,7 @@ export type PreviewAutomationResponse = typeof PreviewAutomationResponse.Type; export class PreviewAutomationUnavailableError extends Schema.TaggedErrorClass()( "PreviewAutomationUnavailableError", { - capability: Schema.Literal("preview"), + capability: Schema.Literals(["preview", "threads"]), environmentId: EnvironmentId, threadId: ThreadId, providerSessionId: TrimmedNonEmptyString, diff --git a/packages/contracts/src/provider.ts b/packages/contracts/src/provider.ts index 94fb007a7bc..77941ab78ee 100644 --- a/packages/contracts/src/provider.ts +++ b/packages/contracts/src/provider.ts @@ -50,6 +50,13 @@ export const ProviderSession = Schema.Struct({ }); export type ProviderSession = typeof ProviderSession.Type; +export const ProviderSessionForkSource = Schema.Struct({ + threadId: ThreadId, + sourceTurnId: Schema.optional(TurnId), + resumeCursor: Schema.optional(Schema.Unknown), +}); +export type ProviderSessionForkSource = typeof ProviderSessionForkSource.Type; + export const ProviderSessionStartInput = Schema.Struct({ threadId: ThreadId, provider: Schema.optional(ProviderDriverKind), @@ -58,6 +65,7 @@ export const ProviderSessionStartInput = Schema.Struct({ cwd: Schema.optional(TrimmedNonEmptyString), modelSelection: Schema.optional(ModelSelection), resumeCursor: Schema.optional(Schema.Unknown), + forkFrom: Schema.optional(ProviderSessionForkSource), approvalPolicy: Schema.optional(ProviderApprovalPolicy), sandboxMode: Schema.optional(ProviderSandboxMode), runtimeMode: RuntimeMode,