From b4e23c92992a5ce66f070892dd59a485480afdce Mon Sep 17 00:00:00 2001 From: Serge Alex Serbinenko <129629680+SergeSerb2@users.noreply.github.com> Date: Mon, 27 Jul 2026 19:25:30 -0700 Subject: [PATCH] feat: add thread branching --- .../Sources/SergeCodeMac/ContentView.swift | 9 ++ .../Sources/SergeCodeMac/Model/AppModel.swift | 12 ++ .../SergeCodeMac/Model/BackendService.swift | 7 ++ .../SergeCodeMac/Model/LiveBackend.swift | 25 ++++ .../SergeCodeMac/Model/MockBackend.swift | 18 +++ .../SergeCodeMac/UI/Shell/SidebarView.swift | 10 ++ .../Sources/T3Kit/OrchestrationModels.swift | 22 ++++ apps/mac/Sources/T3Kit/T3Client.swift | 11 ++ .../T3KitTests/ActivityPayloadTests.swift | 16 +++ apps/mac/docs/wire-protocol.md | 1 + .../agents/ThreadBranchCoordinator.ts | 97 ++++++++++++++ .../src/mcp/toolkits/agents/handlers.ts | 24 +++- apps/server/src/mcp/toolkits/agents/tools.ts | 22 +++- .../Layers/ProviderCommandReactor.ts | 13 +- .../src/orchestration/ThreadBranching.test.ts | 73 +++++++++++ .../src/orchestration/ThreadBranching.ts | 102 +++++++++++++++ .../src/orchestration/decider.branch.test.ts | 118 ++++++++++++++++++ apps/server/src/orchestration/decider.ts | 115 +++++++++++++++++ packages/contracts/src/index.ts | 1 + packages/contracts/src/orchestration.test.ts | 19 +++ packages/contracts/src/orchestration.ts | 16 +++ packages/contracts/src/threadBranching.ts | 41 ++++++ 22 files changed, 768 insertions(+), 4 deletions(-) create mode 100644 apps/server/src/mcp/toolkits/agents/ThreadBranchCoordinator.ts create mode 100644 apps/server/src/orchestration/ThreadBranching.test.ts create mode 100644 apps/server/src/orchestration/ThreadBranching.ts create mode 100644 apps/server/src/orchestration/decider.branch.test.ts create mode 100644 packages/contracts/src/threadBranching.ts diff --git a/apps/mac/Sources/SergeCodeMac/ContentView.swift b/apps/mac/Sources/SergeCodeMac/ContentView.swift index 80e0452477b..d782d3ce7ee 100644 --- a/apps/mac/Sources/SergeCodeMac/ContentView.swift +++ b/apps/mac/Sources/SergeCodeMac/ContentView.swift @@ -219,6 +219,15 @@ struct RootView: View { ComposerPickerSectionLabel(title: project.name) .padding(.top, 4) + AlpineMenuRow( + icon: "arrow.triangle.branch", + title: "Branch Current Session", + detail: "Continue from this conversation in a new session" + ) { + isPresented.wrappedValue = false + Task { await model.branchThread(thread) } + } + AlpineMenuRow( title: "New \(thread.provider.displayName) Session", detail: "Same project and provider as this session" diff --git a/apps/mac/Sources/SergeCodeMac/Model/AppModel.swift b/apps/mac/Sources/SergeCodeMac/Model/AppModel.swift index df0f00006ff..09121a2cc18 100644 --- a/apps/mac/Sources/SergeCodeMac/Model/AppModel.swift +++ b/apps/mac/Sources/SergeCodeMac/Model/AppModel.swift @@ -1705,6 +1705,18 @@ public final class AppModel { } } + @discardableResult + public func branchThread(_ source: ChatThread) async -> ChatThread? { + do { + let thread = try await backend.branchThread(source: source) + selectedThreadID = thread.id + return thread + } catch { + report(error) + return nil + } + } + public func respond(to approval: ApprovalRequest, decision: ApprovalDecision) async { do { try await backend.respondToApproval(id: approval.id, decision: decision) diff --git a/apps/mac/Sources/SergeCodeMac/Model/BackendService.swift b/apps/mac/Sources/SergeCodeMac/Model/BackendService.swift index 571ae2b17d6..87ef2aa3f4c 100644 --- a/apps/mac/Sources/SergeCodeMac/Model/BackendService.swift +++ b/apps/mac/Sources/SergeCodeMac/Model/BackendService.swift @@ -82,6 +82,7 @@ public protocol BackendService: Sendable { /// `title` nil means the backend picks its generic default. func createThread(projectID: String, provider: ProviderKind, title: String?) async throws -> ChatThread + func branchThread(source: ChatThread) async throws -> ChatThread func archiveThread(id: String) async throws func unarchiveThread(id: String) async throws func settleThread(id: String) async throws @@ -186,6 +187,12 @@ public protocol BackendService: Sendable { } public extension BackendService { + func branchThread(source: ChatThread) async throws -> ChatThread { + try await createThread( + projectID: source.projectID, provider: source.provider, + title: "Branch of \(source.title)") + } + func archivedThreadsPage(cursor: Int?, limit: Int) async throws -> ArchivedThreadsPage { // Default for backends without a separate archived query: paginate // the full thread list client-side. diff --git a/apps/mac/Sources/SergeCodeMac/Model/LiveBackend.swift b/apps/mac/Sources/SergeCodeMac/Model/LiveBackend.swift index db23d1a3d11..7596ded9fc1 100644 --- a/apps/mac/Sources/SergeCodeMac/Model/LiveBackend.swift +++ b/apps/mac/Sources/SergeCodeMac/Model/LiveBackend.swift @@ -1636,6 +1636,31 @@ public actor LiveBackend: BackendService { return thread } + public func branchThread(source: ChatThread) async throws -> ChatThread { + guard let client = currentClient else { throw LiveBackendError.notConnected } + let threadID = UUID().uuidString + let title = "Branch of \(source.title)" + _ = try await client.branchThread( + sourceThreadId: source.id, threadId: threadID, title: title) + let thread = ChatThread( + id: threadID, projectID: source.projectID, title: title, provider: source.provider, + status: .idle, updatedAt: Date(), latestUserMessageAt: source.latestUserMessageAt, + runtimeMode: source.runtimeMode, interactionMode: source.interactionMode, + modelInstanceID: source.modelInstanceID, modelID: source.modelID, + executorModelInstanceID: source.executorModelInstanceID, + executorModelID: source.executorModelID, + executorMaxSubAgents: source.executorMaxSubAgents, + reasoningEffort: source.reasoningEffort, serviceTier: source.serviceTier) + threadsByID[threadID] = thread + if let selection = modelSelectionsByThread[source.id] { + modelSelectionsByThread[threadID] = selection + } + titleSeedsByThread[threadID] = title + threadEnvByThread[threadID] = threadEnvByThread[source.id] + emitOrdered(threadID: threadID, event: .threadUpserted(thread)) + return thread + } + public func archiveThread(id: String) async throws { guard let client = currentClient else { throw LiveBackendError.notConnected } _ = try await client.archiveThread(threadId: id) diff --git a/apps/mac/Sources/SergeCodeMac/Model/MockBackend.swift b/apps/mac/Sources/SergeCodeMac/Model/MockBackend.swift index 8d62adbbd7f..fad1a70f640 100644 --- a/apps/mac/Sources/SergeCodeMac/Model/MockBackend.swift +++ b/apps/mac/Sources/SergeCodeMac/Model/MockBackend.swift @@ -107,6 +107,10 @@ public final class MockBackend: BackendService, @unchecked Sendable { await state.createThread(projectID: projectID, provider: provider, title: title) } + public func branchThread(source: ChatThread) async throws -> ChatThread { + await state.branchThread(source: source) + } + public func archiveThread(id: String) async throws { await state.archiveThread(id: id) } @@ -689,6 +693,20 @@ private actor MockState { return thread } + func branchThread(source: ChatThread) -> ChatThread { + var thread = source + thread.id = nextID("thread") + thread.title = "Branch of \(source.title)" + thread.status = .idle + thread.updatedAt = Date() + threadsByID[thread.id] = thread + timelinesByThread[thread.id] = timelinesByThread[source.id] ?? [] + diffsByThread[thread.id] = diffsByThread[source.id] ?? [] + checkpointsByThread[thread.id] = checkpointsByThread[source.id] ?? [] + emit(.threadUpserted(thread)) + return thread + } + func archiveThread(id: String) { guard var thread = threadsByID[id] else { return } cancelStreamingTasks(for: id) diff --git a/apps/mac/Sources/SergeCodeMac/UI/Shell/SidebarView.swift b/apps/mac/Sources/SergeCodeMac/UI/Shell/SidebarView.swift index 5bf17714a7f..5b173145b89 100644 --- a/apps/mac/Sources/SergeCodeMac/UI/Shell/SidebarView.swift +++ b/apps/mac/Sources/SergeCodeMac/UI/Shell/SidebarView.swift @@ -1232,6 +1232,16 @@ private struct SidebarThreadMenu: View { } .disabled(!item.isSelectable || providers.isEmpty) + AlpineMenuRow( + icon: "arrow.triangle.branch", + title: "Branch Session" + ) { + Haptics.play(.commit) + dismiss() + Task { await model.branchThread(item.thread) } + } + .disabled(!item.isSelectable) + AlpineMenuSeparator() lifecycleRow diff --git a/apps/mac/Sources/T3Kit/OrchestrationModels.swift b/apps/mac/Sources/T3Kit/OrchestrationModels.swift index da1175016d8..b5783d5a45d 100644 --- a/apps/mac/Sources/T3Kit/OrchestrationModels.swift +++ b/apps/mac/Sources/T3Kit/OrchestrationModels.swift @@ -1073,6 +1073,26 @@ public struct ThreadCreateCommand: Codable, Sendable { } } +public struct ThreadBranchCommand: Encodable, Sendable { + public let type: String = "thread.branch" + public var commandId: String + public var sourceThreadId: String + public var threadId: String + public var title: String? + public var createdAt: String + + public init( + commandId: String, sourceThreadId: String, threadId: String, title: String? = nil, + createdAt: String + ) { + self.commandId = commandId + self.sourceThreadId = sourceThreadId + self.threadId = threadId + self.title = title + self.createdAt = createdAt + } +} + public struct ThreadDeleteCommand: Encodable, Sendable { public let type: String = "thread.delete" public var commandId: String @@ -1490,6 +1510,7 @@ public enum ClientOrchestrationCommand: Encodable, Sendable { case projectMetaUpdate(ProjectMetaUpdateCommand) case projectDelete(ProjectDeleteCommand) case threadCreate(ThreadCreateCommand) + case threadBranch(ThreadBranchCommand) case threadDelete(ThreadDeleteCommand) case threadArchive(ThreadArchiveCommand) case threadUnarchive(ThreadUnarchiveCommand) @@ -1514,6 +1535,7 @@ public enum ClientOrchestrationCommand: Encodable, Sendable { case .projectMetaUpdate(let c): try container.encode(c) case .projectDelete(let c): try container.encode(c) case .threadCreate(let c): try container.encode(c) + case .threadBranch(let c): try container.encode(c) case .threadDelete(let c): try container.encode(c) case .threadArchive(let c): try container.encode(c) case .threadUnarchive(let c): try container.encode(c) diff --git a/apps/mac/Sources/T3Kit/T3Client.swift b/apps/mac/Sources/T3Kit/T3Client.swift index 5a04c0fecc0..88dffc3b845 100644 --- a/apps/mac/Sources/T3Kit/T3Client.swift +++ b/apps/mac/Sources/T3Kit/T3Client.swift @@ -298,6 +298,17 @@ public actor T3Client { createdAt: T3Clock.nowISO8601()))) } + @discardableResult + public func branchThread( + sourceThreadId: String, threadId: String, title: String? = nil + ) async throws -> DispatchResult { + try await dispatch( + .threadBranch( + ThreadBranchCommand( + commandId: T3Ids.newCommandId(), sourceThreadId: sourceThreadId, + threadId: threadId, title: title, createdAt: T3Clock.nowISO8601()))) + } + @discardableResult public func deleteThread(threadId: String) async throws -> DispatchResult { try await dispatch(.threadDelete(ThreadDeleteCommand(commandId: T3Ids.newCommandId(), threadId: threadId))) diff --git a/apps/mac/Tests/T3KitTests/ActivityPayloadTests.swift b/apps/mac/Tests/T3KitTests/ActivityPayloadTests.swift index e697801ba71..ea2ff27189a 100644 --- a/apps/mac/Tests/T3KitTests/ActivityPayloadTests.swift +++ b/apps/mac/Tests/T3KitTests/ActivityPayloadTests.swift @@ -346,6 +346,22 @@ struct ActivityPayloadTests { #expect(interactionObject["interactionMode"] as? String == "plan") } + @Test("thread branch command encodes source and target identities") + func threadBranchCommand() throws { + let command = ClientOrchestrationCommand.threadBranch( + ThreadBranchCommand( + commandId: "branch-command", sourceThreadId: "source-thread", + threadId: "target-thread", title: "Alternative", + createdAt: "2026-07-04T12:00:00.000Z")) + let data = try WireCoding.encoder.encode(command) + let object = try JSONSerialization.jsonObject(with: data) as? [String: Any] ?? [:] + + #expect(object["type"] as? String == "thread.branch") + #expect(object["sourceThreadId"] as? String == "source-thread") + #expect(object["threadId"] as? String == "target-thread") + #expect(object["title"] as? String == "Alternative") + } + @Test("tool.completed decodes the typed skill presentation") func toolPresentationSkill() throws { let activity = try activity( diff --git a/apps/mac/docs/wire-protocol.md b/apps/mac/docs/wire-protocol.md index 1197c070262..2d9bb3dc6c2 100644 --- a/apps/mac/docs/wire-protocol.md +++ b/apps/mac/docs/wire-protocol.md @@ -372,6 +372,7 @@ discriminated by the `type` string field**. Client-dispatchable members: | `project.meta.update` | `projectId, title?, workspaceRoot?, defaultModelSelection?, scripts?` | `:476-484` | | `project.delete` | `projectId, force?` | `:486-491` | | `thread.create` | `threadId, projectId, title, modelSelection, runtimeMode, interactionMode(dflt "default"), branch:string\|null, worktreePath:string\|null, createdAt` | `:493-507` | +| `thread.branch` | `sourceThreadId, threadId, title?, createdAt`; server snapshots bounded history, remaps message ids, and creates an independent provider session | `packages/contracts/src/orchestration.ts` | | `thread.delete` | `threadId` | `:509-513` | | `thread.archive` | `threadId` | `:515-519` | | `thread.unarchive` | `threadId` | `:521-525` | diff --git a/apps/server/src/mcp/toolkits/agents/ThreadBranchCoordinator.ts b/apps/server/src/mcp/toolkits/agents/ThreadBranchCoordinator.ts new file mode 100644 index 00000000000..571fcd063bb --- /dev/null +++ b/apps/server/src/mcp/toolkits/agents/ThreadBranchCoordinator.ts @@ -0,0 +1,97 @@ +import { + BranchThreadError, + type BranchThreadInput, + type BranchThreadResult, + CommandId, + ThreadId, +} from "@t3tools/contracts"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; + +import { selectThreadBranchContext } from "../../../orchestration/ThreadBranching.ts"; +import { OrchestrationEngineService } from "../../../orchestration/Services/OrchestrationEngine.ts"; +import { ProjectionSnapshotQuery } from "../../../orchestration/Services/ProjectionSnapshotQuery.ts"; +import type { McpInvocationScope } from "../../McpInvocationContext.ts"; + +export interface ThreadBranchCoordinatorShape { + readonly branch: ( + scope: McpInvocationScope, + input: BranchThreadInput, + ) => Effect.Effect; +} + +export class ThreadBranchCoordinator extends Context.Service< + ThreadBranchCoordinator, + ThreadBranchCoordinatorShape +>()("t3/mcp/toolkits/agents/ThreadBranchCoordinator") {} + +const makeThreadBranchCoordinator = Effect.gen(function* () { + const crypto = yield* Crypto.Crypto; + const engine = yield* OrchestrationEngineService; + const snapshots = yield* ProjectionSnapshotQuery; + + const branch: ThreadBranchCoordinatorShape["branch"] = Effect.fn( + "ThreadBranchCoordinator.branch", + )(function* (scope, input) { + const source = yield* snapshots.getThreadDetailById(scope.threadId).pipe( + Effect.mapError( + (cause) => + new BranchThreadError({ + reason: "dispatch-failed", + description: `Failed to read the current thread: ${String(cause)}`, + }), + ), + ); + if (Option.isNone(source)) { + return yield* new BranchThreadError({ + reason: "caller-thread-not-found", + description: "The calling session's thread no longer exists; it cannot be branched.", + }); + } + + const threadUuid = yield* crypto.randomUUIDv4.pipe(Effect.orDie); + const commandUuid = yield* crypto.randomUUIDv4.pipe(Effect.orDie); + const createdAt = yield* DateTime.now.pipe(Effect.map(DateTime.formatIso)); + const title = input.title ?? `Branch of ${source.value.title}`; + const context = selectThreadBranchContext(source.value.messages); + const threadId = ThreadId.make(threadUuid); + + yield* engine + .dispatch({ + type: "thread.branch", + commandId: CommandId.make(`server:thread-branch:${commandUuid}`), + sourceThreadId: scope.threadId, + threadId, + title, + createdAt, + }) + .pipe( + Effect.mapError( + (cause) => + new BranchThreadError({ + reason: "dispatch-failed", + description: `Failed to branch the current thread: ${String(cause)}`, + }), + ), + ); + + return { + threadId, + sourceThreadId: scope.threadId, + title, + copiedMessageCount: context.messages.length, + omittedMessageCount: context.omittedMessageCount, + }; + }); + + return ThreadBranchCoordinator.of({ branch }); +}); + +export const ThreadBranchCoordinatorLive = Layer.effect( + ThreadBranchCoordinator, + makeThreadBranchCoordinator, +); diff --git a/apps/server/src/mcp/toolkits/agents/handlers.ts b/apps/server/src/mcp/toolkits/agents/handlers.ts index 9eaa32973e0..c2266ebdd2a 100644 --- a/apps/server/src/mcp/toolkits/agents/handlers.ts +++ b/apps/server/src/mcp/toolkits/agents/handlers.ts @@ -1,9 +1,10 @@ -import { DelegateError } from "@t3tools/contracts"; +import { BranchThreadError, DelegateError } from "@t3tools/contracts"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; import { DelegateCoordinator, DelegateCoordinatorLive } from "./DelegateCoordinator.ts"; +import { ThreadBranchCoordinator, ThreadBranchCoordinatorLive } from "./ThreadBranchCoordinator.ts"; import { DelegateToolkit } from "./tools.ts"; const requireAgentsScope = Effect.fn("DelegateToolkit.requireAgentsScope")(function* () { @@ -17,6 +18,19 @@ const requireAgentsScope = Effect.fn("DelegateToolkit.requireAgentsScope")(funct return invocation; }); +const requireThreadBranchScope = Effect.fn("DelegateToolkit.requireThreadBranchScope")( + function* () { + const invocation = yield* McpInvocationContext.McpInvocationContext; + if (!invocation.capabilities.has("agents")) { + return yield* new BranchThreadError({ + reason: "capability-unavailable", + description: "This MCP credential does not grant the agents capability.", + }); + } + return invocation; + }, +); + const DelegateToolkitHandlers = DelegateToolkit.toLayer({ delegate_task: (input) => Effect.gen(function* () { @@ -30,8 +44,14 @@ const DelegateToolkitHandlers = DelegateToolkit.toLayer({ const coordinator = yield* DelegateCoordinator; return yield* coordinator.waitForDelegate(scope, input); }), + branch_thread: (input) => + Effect.gen(function* () { + const scope = yield* requireThreadBranchScope(); + const coordinator = yield* ThreadBranchCoordinator; + return yield* coordinator.branch(scope, input); + }), }); export const DelegateToolkitHandlersLive = DelegateToolkitHandlers.pipe( - Layer.provide(DelegateCoordinatorLive), + Layer.provide(Layer.mergeAll(DelegateCoordinatorLive, ThreadBranchCoordinatorLive)), ); diff --git a/apps/server/src/mcp/toolkits/agents/tools.ts b/apps/server/src/mcp/toolkits/agents/tools.ts index 82ebf3e869f..e1e89217c2e 100644 --- a/apps/server/src/mcp/toolkits/agents/tools.ts +++ b/apps/server/src/mcp/toolkits/agents/tools.ts @@ -1,4 +1,7 @@ import { + BranchThreadError, + BranchThreadInput, + BranchThreadResult, DelegateError, DelegateTaskInput, DelegateTaskResult, @@ -8,6 +11,7 @@ import { Tool, Toolkit } from "effect/unstable/ai"; import * as McpInvocationContext from "../../McpInvocationContext.ts"; import { DelegateCoordinator } from "./DelegateCoordinator.ts"; +import { ThreadBranchCoordinator } from "./ThreadBranchCoordinator.ts"; const dependencies = [McpInvocationContext.McpInvocationContext, DelegateCoordinator]; @@ -35,4 +39,20 @@ export const WaitForDelegateTool = Tool.make("wait_for_delegate", { .annotate(Tool.OpenWorld, false) .annotate(Tool.Destructive, false); -export const DelegateToolkit = Toolkit.make(DelegateTaskTool, WaitForDelegateTool); +export const BranchThreadTool = Tool.make("branch_thread", { + description: + "Create a new independent top-level thread from the current conversation. The new thread inherits a bounded snapshot of this conversation, its project, provider/model settings, permissions, and workspace, but starts a separate provider session. Use this when the user asks to explore or continue an alternative path in a new thread. Returns the new thread id; it does not start a turn in that thread.", + parameters: BranchThreadInput, + success: BranchThreadResult, + failure: BranchThreadError, + dependencies: [McpInvocationContext.McpInvocationContext, ThreadBranchCoordinator], +}) + .annotate(Tool.Title, "Branch current thread") + .annotate(Tool.OpenWorld, false) + .annotate(Tool.Destructive, false); + +export const DelegateToolkit = Toolkit.make( + DelegateTaskTool, + WaitForDelegateTool, + BranchThreadTool, +); diff --git a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts index 2baf2421d91..95be92cf244 100644 --- a/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts +++ b/apps/server/src/orchestration/Layers/ProviderCommandReactor.ts @@ -46,6 +46,7 @@ import { import { ServerSettingsService } from "../../serverSettings.ts"; import { VcsStatusBroadcaster } from "../../vcs/VcsStatusBroadcaster.ts"; import { GitWorkflowService } from "../../git/GitWorkflowService.ts"; +import { formatThreadBranchPrompt, readThreadBranchActivity } from "../ThreadBranching.ts"; const isProviderAdapterRequestError = Schema.is(ProviderAdapterRequestError); const isProviderDriverKind = Schema.is(ProviderDriverKind); @@ -896,6 +897,8 @@ const make = Effect.gen(function* () { new Error(`Thread '${input.threadId}' was not found in read model.`), ); } + const branchActivity = readThreadBranchActivity(thread.activities); + const needsBranchBootstrap = thread.session === null && branchActivity !== null; yield* ensureSessionForThread(input.threadId, input.createdAt, { ...(input.modelSelection !== undefined ? { modelSelection: input.modelSelection } : {}), ...(input.interactionMode !== undefined ? { interactionMode: input.interactionMode } : {}), @@ -903,7 +906,15 @@ const make = Effect.gen(function* () { if (input.modelSelection !== undefined) { threadModelSelections.set(input.threadId, input.modelSelection); } - const normalizedInput = toNonEmptyProviderInput(input.messageText); + const messageText = + needsBranchBootstrap && branchActivity !== null + ? formatThreadBranchPrompt({ + sourceTitle: branchActivity.sourceTitle, + messages: thread.messages.slice(0, -1), + currentMessage: input.messageText, + }) + : input.messageText; + const normalizedInput = toNonEmptyProviderInput(messageText); const normalizedAttachments = input.attachments ?? []; const activeSession = yield* providerService .listSessions() diff --git a/apps/server/src/orchestration/ThreadBranching.test.ts b/apps/server/src/orchestration/ThreadBranching.test.ts new file mode 100644 index 00000000000..b06ac7e45ae --- /dev/null +++ b/apps/server/src/orchestration/ThreadBranching.test.ts @@ -0,0 +1,73 @@ +import { MessageId, type OrchestrationMessage } from "@t3tools/contracts"; +import { describe, expect, it } from "@effect/vitest"; + +import { + formatThreadBranchPrompt, + selectThreadBranchContext, + THREAD_BRANCH_MAX_CONTEXT_CHARS, + THREAD_BRANCH_MAX_MESSAGES, +} from "./ThreadBranching.ts"; + +const message = (index: number, text = `message ${index}`): OrchestrationMessage => ({ + id: MessageId.make(`message-${index}`), + role: index % 2 === 0 ? "user" : "assistant", + text, + turnId: null, + streaming: false, + createdAt: `2026-01-01T00:00:${String(index).padStart(2, "0")}.000Z`, + updatedAt: `2026-01-01T00:00:${String(index).padStart(2, "0")}.000Z`, +}); + +describe("thread branching context", () => { + it("keeps the newest bounded messages in chronological order", () => { + const messages = Array.from({ length: THREAD_BRANCH_MAX_MESSAGES + 4 }, (_, index) => + message(index), + ); + + const selected = selectThreadBranchContext(messages); + + expect(selected.messages).toHaveLength(THREAD_BRANCH_MAX_MESSAGES); + expect(selected.messages[0]?.text).toBe("message 4"); + expect(selected.messages.at(-1)?.text).toBe(`message ${THREAD_BRANCH_MAX_MESSAGES + 3}`); + expect(selected.omittedMessageCount).toBe(4); + }); + + it("retains the tail of an oversized newest message", () => { + const selected = selectThreadBranchContext([ + message(0, "old"), + message(1, `prefix-${"x".repeat(THREAD_BRANCH_MAX_CONTEXT_CHARS)}-tail`), + ]); + + expect(selected.messages).toHaveLength(1); + expect(selected.messages[0]?.text).toContain("[Earlier content omitted]"); + expect(selected.messages[0]?.text.endsWith("-tail")).toBe(true); + expect(selected.omittedMessageCount).toBe(1); + }); + + it("does not exceed the character budget when only marker space remains", () => { + const selected = selectThreadBranchContext([ + message(0, "older context"), + message(1, "x".repeat(THREAD_BRANCH_MAX_CONTEXT_CHARS - 10)), + ]); + + expect(selected.messages).toHaveLength(1); + expect(selected.messages[0]?.id).toBe(MessageId.make("message-1")); + expect( + selected.messages.reduce((total, entry) => total + entry.text.length, 0), + ).toBeLessThanOrEqual(THREAD_BRANCH_MAX_CONTEXT_CHARS); + expect(selected.omittedMessageCount).toBe(1); + }); + + it("labels inherited context separately from the new user message", () => { + const prompt = formatThreadBranchPrompt({ + sourceTitle: "Original", + messages: [message(0, "first question"), message(1, "first answer")], + currentMessage: "try another approach", + }); + + expect(prompt).toContain('branched from "Original"'); + expect(prompt).toContain(""); + expect(prompt).toContain("Assistant:\nfirst answer"); + expect(prompt).toContain("\n\ntry another approach"); + }); +}); diff --git a/apps/server/src/orchestration/ThreadBranching.ts b/apps/server/src/orchestration/ThreadBranching.ts new file mode 100644 index 00000000000..858986c8e40 --- /dev/null +++ b/apps/server/src/orchestration/ThreadBranching.ts @@ -0,0 +1,102 @@ +import type { OrchestrationMessage, OrchestrationThreadActivity } from "@t3tools/contracts"; + +export const THREAD_BRANCH_ACTIVITY_KIND = "thread.branched"; +export const THREAD_BRANCH_MAX_MESSAGES = 80; +export const THREAD_BRANCH_MAX_CONTEXT_CHARS = 48_000; + +export interface ThreadBranchActivityPayload { + readonly version: 1; + readonly sourceThreadId: string; + readonly sourceTitle: string; + readonly copiedMessageCount: number; + readonly omittedMessageCount: number; +} + +export interface ThreadBranchContextSelection { + readonly messages: ReadonlyArray; + readonly omittedMessageCount: number; +} + +/** + * Retain the newest complete messages that fit the branch budget. The newest + * message may be truncated from the front so a single oversized tool dump + * cannot leave the branch with no useful conversational context. + */ +export function selectThreadBranchContext( + messages: ReadonlyArray, +): ThreadBranchContextSelection { + const candidates = messages.slice(-THREAD_BRANCH_MAX_MESSAGES); + const selected: OrchestrationMessage[] = []; + let remaining = THREAD_BRANCH_MAX_CONTEXT_CHARS; + + for (let index = candidates.length - 1; index >= 0 && remaining > 0; index -= 1) { + const message = candidates[index]; + if (!message) continue; + if (message.text.length <= remaining) { + selected.push(message); + remaining -= message.text.length; + continue; + } + const marker = "[Earlier content omitted]\n"; + if (remaining <= marker.length) break; + const tailLength = remaining - marker.length; + selected.push({ + ...message, + text: `${marker}${message.text.slice(-tailLength)}`, + }); + remaining = 0; + } + + selected.reverse(); + return { + messages: selected, + omittedMessageCount: messages.length - selected.length, + }; +} + +export function readThreadBranchActivity( + activities: ReadonlyArray, +): ThreadBranchActivityPayload | null { + const activity = activities.findLast((entry) => entry.kind === THREAD_BRANCH_ACTIVITY_KIND); + if (!activity || typeof activity.payload !== "object" || activity.payload === null) return null; + const payload = activity.payload as Record; + if ( + payload.version !== 1 || + typeof payload.sourceThreadId !== "string" || + typeof payload.sourceTitle !== "string" || + typeof payload.copiedMessageCount !== "number" || + typeof payload.omittedMessageCount !== "number" + ) { + return null; + } + return payload as unknown as ThreadBranchActivityPayload; +} + +export function formatThreadBranchPrompt(input: { + readonly sourceTitle: string; + readonly messages: ReadonlyArray; + readonly currentMessage: string; +}): string { + const transcript = input.messages + .map((message) => { + const role = + message.role === "assistant" ? "Assistant" : message.role === "system" ? "System" : "User"; + const attachmentNote = + message.attachments && message.attachments.length > 0 + ? `\n[${message.attachments.length} attachment(s) were present in this message]` + : ""; + return `${role}:\n${message.text}${attachmentNote}`; + }) + .join("\n\n"); + + return [ + `This conversation was branched from "${input.sourceTitle}".`, + "Use the following read-only transcript snapshot as prior conversation context. Continue independently; do not assume that work performed later in the source thread also happened here.", + "", + transcript, + "", + "", + input.currentMessage, + "", + ].join("\n\n"); +} diff --git a/apps/server/src/orchestration/decider.branch.test.ts b/apps/server/src/orchestration/decider.branch.test.ts new file mode 100644 index 00000000000..d97fba0649d --- /dev/null +++ b/apps/server/src/orchestration/decider.branch.test.ts @@ -0,0 +1,118 @@ +import { + CommandId, + MessageId, + ProjectId, + ProviderInstanceId, + ThreadId, + 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"; + +const NOW = "2026-01-01T00:00:00.000Z"; + +const readModel: OrchestrationReadModel = { + snapshotSequence: 0, + projects: [], + threads: [ + { + id: ThreadId.make("source"), + projectId: ProjectId.make("project"), + title: "Original", + modelSelection: { + instanceId: ProviderInstanceId.make("codex"), + model: "gpt-5.4", + }, + runtimeMode: "approval-required", + interactionMode: "advisor", + executorModelSelection: { + instanceId: ProviderInstanceId.make("claude"), + model: "claude-sonnet", + }, + executorMaxSubAgents: 5, + branch: "feature/original", + worktreePath: "/tmp/original", + parentThreadId: null, + latestTurn: null, + createdAt: NOW, + updatedAt: NOW, + archivedAt: null, + settledOverride: null, + settledAt: null, + deletedAt: null, + autoReviewPhase: null, + messages: [ + { + id: MessageId.make("source-user"), + role: "user", + text: "original question", + turnId: null, + streaming: false, + createdAt: NOW, + updatedAt: NOW, + }, + { + id: MessageId.make("source-assistant"), + role: "assistant", + text: "original answer", + turnId: null, + streaming: false, + createdAt: NOW, + updatedAt: NOW, + }, + ], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, + }, + ], + updatedAt: NOW, +}; + +it.layer(NodeServices.layer)("thread branch decider", (it) => { + it.effect("creates an independent thread and remaps copied message ids", () => + Effect.gen(function* () { + const result = yield* decideOrchestrationCommand({ + command: { + type: "thread.branch", + commandId: CommandId.make("branch-command"), + sourceThreadId: ThreadId.make("source"), + threadId: ThreadId.make("target"), + createdAt: NOW, + }, + readModel, + }); + const events = Array.isArray(result) ? result : [result]; + + expect(events.map((event) => event.type)).toEqual([ + "thread.created", + "thread.executor-model-set", + "thread.activity-appended", + "thread.message-sent", + "thread.message-sent", + ]); + const created = events[0]; + expect(created?.type).toBe("thread.created"); + if (created?.type === "thread.created") { + expect(created.payload.title).toBe("Branch of Original"); + expect(created.payload.parentThreadId).toBeNull(); + expect(created.payload.branch).toBe("feature/original"); + expect(created.payload.worktreePath).toBe("/tmp/original"); + } + + const copied = events.filter((event) => event.type === "thread.message-sent"); + expect(copied.map((event) => event.payload.text)).toEqual([ + "original question", + "original answer", + ]); + expect(copied.map((event) => event.payload.messageId)).not.toContain( + MessageId.make("source-user"), + ); + expect(new Set(copied.map((event) => event.payload.messageId)).size).toBe(2); + }), + ); +}); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index d2a115a6916..c78c803bbec 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, @@ -26,6 +27,7 @@ import { isSubAgentThreadTitle, withSubAgentThreadTitle, } from "./subAgentModelPolicy.ts"; +import { selectThreadBranchContext, THREAD_BRANCH_ACTIVITY_KIND } from "./ThreadBranching.ts"; const nowIso = Effect.map(DateTime.now, DateTime.formatIso); @@ -320,6 +322,119 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }; } + case "thread.branch": { + const source = yield* requireThreadNotArchived({ + readModel, + command, + threadId: command.sourceThreadId, + }); + yield* requireThreadAbsent({ + readModel, + command, + threadId: command.threadId, + }); + + const title = command.title ?? `Branch of ${source.title}`; + const context = selectThreadBranchContext(source.messages); + const createdEvent: PlannedOrchestrationEvent = { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.created", + payload: { + threadId: command.threadId, + projectId: source.projectId, + title, + modelSelection: source.modelSelection, + runtimeMode: source.runtimeMode, + interactionMode: source.interactionMode, + executorModelSelection: source.executorModelSelection, + branch: source.branch, + worktreePath: source.worktreePath, + parentThreadId: null, + createdAt: command.createdAt, + updatedAt: command.createdAt, + }, + }; + const executorSettingsEvent: PlannedOrchestrationEvent = { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.executor-model-set", + payload: { + threadId: command.threadId, + executorModelSelection: source.executorModelSelection, + executorMaxSubAgents: source.executorMaxSubAgents, + updatedAt: command.createdAt, + }, + }; + const activityId = yield* Crypto.Crypto.pipe( + Effect.flatMap((crypto) => crypto.randomUUIDv4), + Effect.map(EventId.make), + ); + const branchActivityEvent: PlannedOrchestrationEvent = { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.activity-appended", + payload: { + threadId: command.threadId, + activity: { + id: activityId, + tone: "info", + kind: THREAD_BRANCH_ACTIVITY_KIND, + summary: `Branched from ${source.title}`, + payload: { + version: 1, + sourceThreadId: source.id, + sourceTitle: source.title, + copiedMessageCount: context.messages.length, + omittedMessageCount: context.omittedMessageCount, + }, + turnId: null, + createdAt: command.createdAt, + }, + }, + }; + const copiedMessageEvents: PlannedOrchestrationEvent[] = []; + for (const message of context.messages) { + const messageId = yield* Crypto.Crypto.pipe( + Effect.flatMap((crypto) => crypto.randomUUIDv4), + Effect.map(MessageId.make), + ); + copiedMessageEvents.push({ + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt: command.createdAt, + commandId: command.commandId, + })), + type: "thread.message-sent", + payload: { + threadId: command.threadId, + messageId, + role: message.role, + text: message.text, + ...(message.attachments !== undefined ? { attachments: message.attachments } : {}), + turnId: null, + streaming: false, + createdAt: message.createdAt, + updatedAt: command.createdAt, + }, + }); + } + return [createdEvent, executorSettingsEvent, branchActivityEvent, ...copiedMessageEvents]; + } + case "thread.delete": { yield* requireThread({ readModel, diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index c7bd4184724..b6f29000a6b 100644 --- a/packages/contracts/src/index.ts +++ b/packages/contracts/src/index.ts @@ -25,6 +25,7 @@ export * from "./assets.ts"; export * from "./review.ts"; export * from "./autoReview.ts"; export * from "./subAgents.ts"; +export * from "./threadBranching.ts"; export * from "./customInstructions.ts"; export * from "./extensions.ts"; export * from "./rpc.ts"; diff --git a/packages/contracts/src/orchestration.test.ts b/packages/contracts/src/orchestration.test.ts index 2edcf0541fd..eb52f6a7534 100644 --- a/packages/contracts/src/orchestration.test.ts +++ b/packages/contracts/src/orchestration.test.ts @@ -294,6 +294,25 @@ it.effect("accepts bootstrap metadata in thread.turn.start", () => }), ); +it.effect("decodes thread.branch commands with an optional title", () => + Effect.gen(function* () { + const parsed = yield* decodeOrchestrationCommand({ + type: "thread.branch", + commandId: "cmd-branch", + sourceThreadId: "thread-source", + threadId: "thread-branch", + title: "Alternative approach", + createdAt: "2026-01-01T00:00:00.000Z", + }); + + assert.strictEqual(parsed.type, "thread.branch"); + if (parsed.type !== "thread.branch") return; + assert.strictEqual(parsed.sourceThreadId, "thread-source"); + assert.strictEqual(parsed.threadId, "thread-branch"); + assert.strictEqual(parsed.title, "Alternative approach"); + }), +); + it.effect("decodes thread.created runtime mode for historical events", () => Effect.gen(function* () { const parsed = yield* decodeThreadCreatedPayload({ diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index feb5bb7e917..c413a21fbd0 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -724,6 +724,20 @@ const ThreadCreateCommand = Schema.Struct({ createdAt: IsoDateTime, }); +/** + * Creates an independent top-level thread from a bounded snapshot of another + * thread. The server owns context selection and message-id remapping so clients + * cannot forge copied history or alias projection rows. + */ +const ThreadBranchCommand = Schema.Struct({ + type: Schema.Literal("thread.branch"), + commandId: CommandId, + sourceThreadId: ThreadId, + threadId: ThreadId, + title: Schema.optional(TrimmedNonEmptyString), + createdAt: IsoDateTime, +}); + const ThreadDeleteCommand = Schema.Struct({ type: Schema.Literal("thread.delete"), commandId: CommandId, @@ -919,6 +933,7 @@ const DispatchableClientOrchestrationCommand = Schema.Union([ ProjectMetaUpdateCommand, ProjectDeleteCommand, ThreadCreateCommand, + ThreadBranchCommand, ThreadDeleteCommand, ThreadArchiveCommand, ThreadUnarchiveCommand, @@ -944,6 +959,7 @@ export const ClientOrchestrationCommand = Schema.Union([ ProjectMetaUpdateCommand, ProjectDeleteCommand, ThreadCreateCommand, + ThreadBranchCommand, ThreadDeleteCommand, ThreadArchiveCommand, ThreadUnarchiveCommand, diff --git a/packages/contracts/src/threadBranching.ts b/packages/contracts/src/threadBranching.ts new file mode 100644 index 00000000000..c8ada52d353 --- /dev/null +++ b/packages/contracts/src/threadBranching.ts @@ -0,0 +1,41 @@ +import * as Schema from "effect/Schema"; + +import { NonNegativeInt, ThreadId, TrimmedNonEmptyString } from "./baseSchemas.ts"; + +export const BranchThreadInput = Schema.Struct({ + title: Schema.optional( + TrimmedNonEmptyString.annotate({ + description: + "Optional title for the new thread. When omitted, the server derives one from the current thread.", + }), + ), +}); +export type BranchThreadInput = typeof BranchThreadInput.Type; + +export const BranchThreadResult = Schema.Struct({ + threadId: ThreadId, + sourceThreadId: ThreadId, + title: TrimmedNonEmptyString, + copiedMessageCount: NonNegativeInt, + omittedMessageCount: NonNegativeInt, +}); +export type BranchThreadResult = typeof BranchThreadResult.Type; + +export const BranchThreadErrorReason = Schema.Literals([ + "capability-unavailable", + "caller-thread-not-found", + "dispatch-failed", +]); +export type BranchThreadErrorReason = typeof BranchThreadErrorReason.Type; + +export class BranchThreadError extends Schema.TaggedErrorClass()( + "BranchThreadError", + { + reason: BranchThreadErrorReason, + description: Schema.String, + }, +) { + override get message(): string { + return this.description; + } +}