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/decider.fork.test.ts b/apps/server/src/orchestration/decider.fork.test.ts index df7b2a8c01c..adf006899f1 100644 --- a/apps/server/src/orchestration/decider.fork.test.ts +++ b/apps/server/src/orchestration/decider.fork.test.ts @@ -111,12 +111,12 @@ const seedReadModel = (): OrchestrationReadModel => ({ ], }); -const forkCommand = (sourceTurnId: TurnId) => ({ +const forkCommand = (sourceTurnId?: TurnId) => ({ type: "thread.fork" as const, commandId: CommandId.make("command-fork"), threadId: forkThreadId, sourceThreadId, - sourceTurnId, + ...(sourceTurnId === undefined ? {} : { sourceTurnId }), title: "Forked thread", createdAt: now, }); @@ -170,6 +170,76 @@ it.layer(NodeServices.layer)("thread fork decider", (it) => { }), ); + 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(); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index f2908153922..5d1d983920b 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -273,17 +273,31 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" threadId: command.threadId, }); - const sourceTurnId = command.sourceTurnId; - if (!sourceThread.messages.some((message) => message.turnId === sourceTurnId)) { + 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}'.`, }); } - const latestTurn = sourceThread.latestTurn; if ( - latestTurn?.turnId === sourceTurnId && - latestTurn.state === "running" + sourceTurnId !== null && + sourceThread.latestTurn?.turnId === sourceTurnId && + sourceThread.latestTurn.state === "running" ) { return yield* new OrchestrationCommandInvariantError({ commandType: command.type, @@ -291,9 +305,10 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }); } - const cutoffIndex = sourceThread.messages.findLastIndex( - (message) => message.turnId === sourceTurnId, - ); + 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) 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/CodexSessionRuntime.test.ts b/apps/server/src/provider/Layers/CodexSessionRuntime.test.ts index 262f74f6b29..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/); } }); }); diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index feb0f71dc43..2e3753f7b86 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"; @@ -1259,6 +1260,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(() => { @@ -1299,6 +1302,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( @@ -1428,6 +1432,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; diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index ee9cc7dbc9d..60acfb712b9 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -553,7 +553,7 @@ const ThreadForkCommand = Schema.Struct({ commandId: CommandId, threadId: ThreadId, sourceThreadId: ThreadId, - sourceTurnId: TurnId, + sourceTurnId: Schema.optional(TurnId), title: TrimmedNonEmptyString, createdAt: IsoDateTime, }); 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,