diff --git a/apps/cli/src/core/threads/coordinator.start-thread.test.ts b/apps/cli/src/core/threads/coordinator.start-thread.test.ts new file mode 100644 index 0000000..f9bce36 --- /dev/null +++ b/apps/cli/src/core/threads/coordinator.start-thread.test.ts @@ -0,0 +1,425 @@ +import { beforeEach, describe, expect, mock, test } from "bun:test"; +import type { RuntimeSession } from "@acp-kit/core"; +import { Result } from "better-result"; +import type { AgentPool } from "@/core/acp/pool"; +import { ThreadCoordinator } from "./coordinator"; + +type ThreadRow = { + id: string; + projectId: string; + name: string; + agentName: string | undefined; + sessionId: string | undefined; + agentLocked: true | undefined; + branch: string | null; + worktreePath: string | null; + createdAt: string; + updatedAt: string; +}; + +const threads = new Map(); +const conversations: Array<{ + threadId: string; + turnId: string; + event: { type: string; message?: string; content?: string }; +}> = []; +const ops: string[] = []; + +let projectCwd = "/tmp/project"; +let sessionCreateError: Error | null = null; +let setModelError: Error | null = null; +const sessions: RuntimeSession[] = []; +let nextSessionCwd: string | undefined; + +function createMockSession(sessionId: string): RuntimeSession { + return { + sessionId, + transcript: { + session: { + models: { + availableModels: [{ modelId: "model-1", name: "Model 1" }], + }, + modes: { availableModes: [{ id: "mode-1", name: "Mode 1" }] }, + configOptions: [], + }, + }, + close: mock(async () => undefined), + setModel: mock(() => { + ops.push("preferences"); + if (setModelError) return Promise.reject(setModelError); + return Promise.resolve(); + }), + setMode: mock(async () => undefined), + on: () => () => undefined, + prompt: mock(async () => ({})), + cancel: mock(async () => undefined), + } as unknown as RuntimeSession; +} + +mock.module("@cyrus/database/repositories/projects", () => ({ + resolveProjectCwd: () => Promise.resolve(Result.ok(projectCwd)), + getProject: () => + Promise.resolve( + Result.ok({ id: "project-1", name: "Project", cwd: projectCwd }) + ), +})); + +mock.module("@cyrus/database/repositories/git", () => ({ + resolveThreadGitCwd: (threadId: string) => { + const thread = threads.get(threadId); + if (!thread) + return Promise.resolve( + Result.err(new Error(`thread ${threadId} not found`)) + ); + return Promise.resolve(Result.ok(thread.worktreePath ?? projectCwd)); + }, +})); + +mock.module("@cyrus/database/repositories/threads", () => ({ + createThread: ( + projectId: string, + options?: { branch?: string; worktreePath?: string } + ) => { + ops.push("createThread"); + const id = `thread-${threads.size + 1}`; + const now = "2026-07-17T00:00:00.000Z"; + const row: ThreadRow = { + id, + projectId, + name: options?.branch ?? "New thread", + agentName: undefined, + sessionId: undefined, + agentLocked: undefined, + branch: options?.branch ?? null, + worktreePath: options?.worktreePath ?? null, + createdAt: now, + updatedAt: now, + }; + threads.set(id, row); + return Promise.resolve(Result.ok({ ...row })); + }, + getThread: (threadId: string) => { + const row = threads.get(threadId); + return Promise.resolve(Result.ok(row ? { ...row } : undefined)); + }, + bindThreadAgent: ( + threadId: string, + _projectId: string, + data: { agentName: string; sessionId: string } + ) => { + ops.push("bindThreadAgent"); + const row = threads.get(threadId); + if (!row) return Promise.resolve(Result.err(new Error("not found"))); + row.agentName = data.agentName; + row.sessionId = data.sessionId; + return Promise.resolve(Result.ok({ ...row })); + }, + setAgentLocked: (threadId: string) => { + ops.push("setAgentLocked"); + const row = threads.get(threadId); + if (!row) return Promise.resolve(Result.err(new Error("not found"))); + row.agentLocked = true; + return Promise.resolve(Result.ok({ ...row })); + }, + updateThreadWorktreePath: (threadId: string, worktreePath: string | null) => { + ops.push("git"); + const row = threads.get(threadId); + if (!row) return Promise.resolve(Result.err(new Error("not found"))); + row.worktreePath = worktreePath; + return Promise.resolve(Result.ok({ ...row })); + }, + clearThreadDraftBinding: () => Promise.resolve(Result.ok(undefined)), +})); + +mock.module("@cyrus/database/repositories/conversations", () => ({ + appendConversation: ( + threadId: string, + entry: { + threadId: string; + turnId: string; + event: { type: string; message?: string; content?: string }; + } + ) => { + conversations.push({ + threadId, + turnId: entry.turnId, + event: entry.event, + }); + return Promise.resolve( + Result.ok({ + chunk: { + threadId, + turnId: entry.turnId, + seq: conversations.length, + event: entry.event, + }, + }) + ); + }, +})); + +mock.module("@/git/worktree", () => ({ + createGitWorktree: (_projectCwd: string, refName: string, path?: string) => { + ops.push("git"); + return Promise.resolve(Result.ok(path ?? `/tmp/worktrees/${refName}`)); + }, + removeGitWorktree: () => Promise.resolve(Result.ok(undefined)), +})); + +mock.module("@/git/checkout", () => ({ + tryCheckoutGitRef: () => { + ops.push("git"); + return Promise.resolve(Result.ok(undefined)); + }, + checkoutGitRef: () => Promise.resolve(Result.ok(undefined)), +})); + +function createCoordinator() { + const pool = { + getState: () => "ready", + getRuntime: async () => ({ + newSession: ({ cwd }: { cwd: string }) => { + ops.push("createBoundSession"); + nextSessionCwd = cwd; + if (sessionCreateError) { + return Promise.reject(sessionCreateError); + } + const session = createMockSession(`session-${sessions.length + 1}`); + sessions.push(session); + return Promise.resolve(session); + }, + agentCapabilities: { loadSession: true }, + }), + getSdkConnection: () => ({ + setSessionConfigOption: mock(async () => undefined), + }), + } as unknown as AgentPool; + + return new ThreadCoordinator(pool); +} + +describe("startThread", () => { + beforeEach(() => { + threads.clear(); + conversations.length = 0; + ops.length = 0; + sessions.length = 0; + sessionCreateError = null; + setModelError = null; + nextSessionCwd = undefined; + projectCwd = "/tmp/project"; + }); + + test("births a thread with a bound session ready to prompt", async () => { + const coordinator = createCoordinator(); + + const started = await coordinator.startThread({ + projectId: "project-1", + agentName: "mock-agent", + message: [{ type: "text", text: "hello" }], + branch: "main", + preferences: { modelId: "model-1" }, + turnId: "turn-1", + }); + + expect(started.isOk()).toBe(true); + if (started.isErr()) throw new Error("expected startThread to succeed"); + expect(started.value.bound).not.toBeNull(); + expect(started.value.turnId).toBe("turn-1"); + expect(ops).toEqual([ + "createThread", + "git", + "createBoundSession", + "preferences", + "bindThreadAgent", + "setAgentLocked", + ]); + + const thread = threads.get(started.value.threadId); + expect(thread?.agentName).toBe("mock-agent"); + expect(thread?.sessionId).toBe("session-1"); + expect(thread?.agentLocked).toBe(true); + + const prompt = await coordinator.prompt( + "mock-agent", + started.value.threadId, + "project-1", + [{ type: "text", text: "hello" }], + "turn-1" + ); + expect(prompt.isOk()).toBe(true); + if (prompt.isErr()) throw new Error("expected prompt to succeed"); + for await (const _event of prompt.value) { + /* drain */ + } + expect(sessions[0]?.prompt).toHaveBeenCalledWith("hello"); + }); + + test("session failure after row creation leaves the message and an error entry", async () => { + sessionCreateError = new Error("session boom"); + const coordinator = createCoordinator(); + + const started = await coordinator.startThread({ + projectId: "project-1", + agentName: "mock-agent", + message: [{ type: "text", text: "keep me" }], + turnId: "turn-fail", + }); + + expect(started.isOk()).toBe(true); + if (started.isErr()) throw new Error("expected ok with bound null"); + expect(started.value.bound).toBeNull(); + expect(threads.has(started.value.threadId)).toBe(true); + + const thread = threads.get(started.value.threadId); + expect(thread?.sessionId).toBeUndefined(); + expect(thread?.agentLocked).toBeUndefined(); + + expect(conversations).toHaveLength(3); + expect(conversations[0]).toMatchObject({ + threadId: started.value.threadId, + turnId: "turn-fail", + event: { type: "user_message", content: "keep me" }, + }); + expect(conversations[1]).toMatchObject({ + threadId: started.value.threadId, + turnId: "turn-fail", + event: { type: "thread_error", code: "coordinator.runtime" }, + }); + expect(conversations[1]?.event.message).toContain("session boom"); + expect(conversations[2]).toMatchObject({ + threadId: started.value.threadId, + turnId: "turn-fail", + event: { type: "turn_interrupted" }, + }); + }); + + test("creates the session at the worktree cwd when a worktree is requested", async () => { + const coordinator = createCoordinator(); + + const started = await coordinator.startThread({ + projectId: "project-1", + agentName: "mock-agent", + message: [{ type: "text", text: "in worktree" }], + branch: "feature", + worktree: true, + turnId: "turn-wt", + }); + + expect(started.isOk()).toBe(true); + if (started.isErr()) throw new Error("expected ok"); + expect(started.value.bound?.cwd).toBe("/tmp/worktrees/feature"); + expect(nextSessionCwd).toBe("/tmp/worktrees/feature"); + expect(threads.get(started.value.threadId)?.worktreePath).toBe( + "/tmp/worktrees/feature" + ); + }); + + test("rejects worktree without a branch before creating a session", async () => { + const coordinator = createCoordinator(); + + const started = await coordinator.startThread({ + projectId: "project-1", + agentName: "mock-agent", + message: [{ type: "text", text: "no branch" }], + worktree: true, + turnId: "turn-wt-missing-branch", + }); + + expect(started.isOk()).toBe(true); + if (started.isErr()) throw new Error("expected ok with bound null"); + expect(started.value.bound).toBeNull(); + expect(sessions).toHaveLength(0); + expect(conversations[1]?.event.message).toContain( + "worktree requires a branch" + ); + }); + + test("applies draft model preference on the new session", async () => { + const coordinator = createCoordinator(); + + const started = await coordinator.startThread({ + projectId: "project-1", + agentName: "mock-agent", + message: [{ type: "text", text: "with model" }], + preferences: { modelId: "model-1" }, + turnId: "turn-prefs", + }); + + expect(started.isOk()).toBe(true); + if (started.isErr()) throw new Error("expected ok"); + expect(started.value.bound).not.toBeNull(); + expect(sessions[0]?.setModel).toHaveBeenCalledWith("model-1"); + }); + + test("skips unsupported model preference and still binds the session", async () => { + setModelError = new Error( + 'Unhandled exception: "Method not found": session/set_model' + ); + const coordinator = createCoordinator(); + + const started = await coordinator.startThread({ + projectId: "project-1", + agentName: "mock-agent", + message: [{ type: "text", text: "unsupported model" }], + preferences: { modelId: "model-1" }, + turnId: "turn-prefs-skip", + }); + + expect(started.isOk()).toBe(true); + if (started.isErr()) throw new Error("expected ok"); + expect(started.value.bound).not.toBeNull(); + expect(sessions[0]?.setModel).toHaveBeenCalledWith("model-1"); + expect( + conversations.some((entry) => entry.event.type === "thread_error") + ).toBe(false); + }); + + test("after a mid-flight failure the thread can be bound and prompted for retry", async () => { + sessionCreateError = new Error("session boom"); + const coordinator = createCoordinator(); + + const started = await coordinator.startThread({ + projectId: "project-1", + agentName: "mock-agent", + message: [{ type: "text", text: "retry me" }], + turnId: "turn-1", + }); + expect(started.isOk()).toBe(true); + if (started.isErr()) throw new Error("expected ok"); + expect(started.value.bound).toBeNull(); + expect( + conversations.some((entry) => entry.event.type === "user_message") + ).toBe(true); + + sessionCreateError = null; + const rebound = await coordinator.bindAgent( + started.value.threadId, + "project-1", + "mock-agent" + ); + expect(rebound.isOk()).toBe(true); + if (rebound.isErr()) throw new Error("expected bind to succeed"); + + const persisted = await coordinator.persistBoundSession( + started.value.threadId, + "project-1", + "mock-agent" + ); + expect(persisted.isOk()).toBe(true); + + const prompt = await coordinator.prompt( + "mock-agent", + started.value.threadId, + "project-1", + [{ type: "text", text: "retry me" }], + "turn-retry" + ); + expect(prompt.isOk()).toBe(true); + if (prompt.isErr()) throw new Error("expected prompt to succeed"); + for await (const _event of prompt.value) { + /* drain */ + } + expect(sessions[0]?.prompt).toHaveBeenCalledWith("retry me"); + }); +}); diff --git a/apps/cli/src/core/threads/coordinator.ts b/apps/cli/src/core/threads/coordinator.ts index f8132d4..ba64307 100644 --- a/apps/cli/src/core/threads/coordinator.ts +++ b/apps/cli/src/core/threads/coordinator.ts @@ -21,6 +21,11 @@ import { type DraftCatalog, getDraftCatalog as getDraftCatalogFn, } from "./draft-catalog"; +import { + type StartThreadInput, + type StartThreadResult, + startThread as startThreadFn, +} from "./start-thread"; import { cancel as cancelTurn, closeAnyThreadSession, @@ -38,6 +43,7 @@ export class ThreadCoordinator implements CoordinatorHost { private readonly agents = new Map(); private readonly pool: AgentPool; private readonly threadMutexes = new Map(); + private readonly projectMutexes = new Map(); constructor(pool: AgentPool) { this.pool = pool; @@ -52,6 +58,15 @@ export class ThreadCoordinator implements CoordinatorHost { return mutex; } + private projectMutexFor(projectId: string): Mutex { + let mutex = this.projectMutexes.get(projectId); + if (!mutex) { + mutex = new Mutex(); + this.projectMutexes.set(projectId, mutex); + } + return mutex; + } + private withThreadLock( threadId: string, fn: () => Promise @@ -59,6 +74,13 @@ export class ThreadCoordinator implements CoordinatorHost { return this.mutexFor(threadId).runExclusive(fn); } + private withProjectLock( + projectId: string, + fn: () => Promise + ): Promise { + return this.projectMutexFor(projectId).runExclusive(fn); + } + getAgent(agentName: string): AgentRuntime { let runtime = this.agents.get(agentName); if (!runtime) { @@ -202,6 +224,16 @@ export class ThreadCoordinator implements CoordinatorHost { return getDraftCatalogFn(this, agentName, projectId); } + /** Birth a thread from a first message: row, session, prefs, binding. */ + startThread( + input: StartThreadInput + ): Promise> { + // Serialize checkout/worktree mutations that share a project cwd. + return this.withProjectLock(input.projectId, () => + startThreadFn(this, input) + ); + } + prompt( agentName: string, threadId: string, diff --git a/apps/cli/src/core/threads/start-thread.ts b/apps/cli/src/core/threads/start-thread.ts new file mode 100644 index 0000000..42b38f5 --- /dev/null +++ b/apps/cli/src/core/threads/start-thread.ts @@ -0,0 +1,266 @@ +import { appendConversation } from "@cyrus/database/repositories/conversations"; +import { resolveProjectCwd } from "@cyrus/database/repositories/projects"; +import { + bindThreadAgent, + createThread, + setAgentLocked, + updateThreadWorktreePath, +} from "@cyrus/database/repositories/threads"; +import { + type CoordinatorError, + coordinatorRepositoryError, + coordinatorRuntimeError, +} from "@cyrus/errors/coordinator"; +import type { ChatMessage } from "@cyrus/schemas/rtc/chat"; +import { formatPromptBlocks } from "@cyrus/schemas/rtc/chat"; +import { randomId } from "@cyrus/utils/identity"; +import { Result } from "better-result"; +import { tryCheckoutGitRef } from "@/git/checkout"; +import { createGitWorktree, removeGitWorktree } from "@/git/worktree"; +import { + coordinatorErrorCode, + coordinatorErrorMessage, +} from "@/utils/thread-errors"; +import type { BoundThread, CoordinatorHost } from "./types"; + +export type StartThreadPreferences = { + modelId?: string; + modeId?: string; + effortId?: string; + personaId?: string; +}; + +export type StartThreadInput = { + projectId: string; + agentName: string; + message: ChatMessage; + preferences?: StartThreadPreferences; + branch?: string; + /** When true with a branch, create a worktree instead of checking out in-place. */ + worktree?: boolean; + worktreePath?: string; + turnId?: string; +}; + +export type StartThreadResult = { + threadId: string; + turnId: string; + /** Bound and ready to prompt; null if setup failed after the row was created. */ + bound: BoundThread | null; +}; + +async function persistFailure( + threadId: string, + turnId: string, + message: ChatMessage, + error: unknown +): Promise> { + const userMessage = await appendConversation(threadId, { + threadId, + turnId, + event: { + type: "user_message", + content: formatPromptBlocks(message), + blocks: message, + }, + }); + if (userMessage.isErr()) { + return Result.err(coordinatorRepositoryError(userMessage.error)); + } + + const threadError = await appendConversation(threadId, { + threadId, + turnId, + event: { + type: "thread_error", + message: coordinatorErrorMessage(error), + ...(coordinatorErrorCode(error) + ? { code: coordinatorErrorCode(error) } + : {}), + }, + }); + if (threadError.isErr()) { + return Result.err(coordinatorRepositoryError(threadError.error)); + } + + const interrupted = await appendConversation(threadId, { + threadId, + turnId, + event: { type: "turn_interrupted" }, + }); + if (interrupted.isErr()) { + return Result.err(coordinatorRepositoryError(interrupted.error)); + } + + return Result.ok(undefined); +} + +async function failAfterRow( + threadId: string, + turnId: string, + message: ChatMessage, + error: unknown +): Promise> { + const persisted = await persistFailure(threadId, turnId, message, error); + if (persisted.isErr()) return Result.err(persisted.error); + return Result.ok({ threadId, turnId, bound: null }); +} + +async function applyBranchOrWorktree( + threadId: string, + projectId: string, + branch: string | undefined, + worktree: boolean | undefined, + worktreePath: string | undefined +): Promise> { + const wantsWorktree = Boolean(worktree || worktreePath); + if (wantsWorktree && !branch) { + return Result.err(coordinatorRuntimeError("worktree requires a branch")); + } + if (!branch) return Result.ok(undefined); + + const projectCwd = await resolveProjectCwd(projectId); + if (projectCwd.isErr()) { + return Result.err(coordinatorRepositoryError(projectCwd.error)); + } + + if (wantsWorktree) { + const created = await createGitWorktree( + projectCwd.value, + branch, + worktreePath + ); + if (created.isErr()) { + return Result.err(coordinatorRuntimeError(created.error.message)); + } + const updated = await updateThreadWorktreePath(threadId, created.value); + if (updated.isErr()) { + await removeGitWorktree(projectCwd.value, created.value); + return Result.err(coordinatorRepositoryError(updated.error)); + } + return Result.ok(undefined); + } + + const checkedOut = await tryCheckoutGitRef(projectCwd.value, branch); + if (checkedOut.isErr()) { + return Result.err(coordinatorRuntimeError(checkedOut.error.message)); + } + return Result.ok(undefined); +} + +async function applyPreferences( + host: CoordinatorHost, + bound: BoundThread, + preferences: StartThreadPreferences | undefined +): Promise> { + if (!preferences) return Result.ok(undefined); + + const runtime = host.getAgent(bound.agentName); + const fields: Array<{ + field: "model" | "mode" | "effort" | "persona"; + value: string; + }> = []; + if (preferences.modelId) + fields.push({ field: "model", value: preferences.modelId }); + if (preferences.modeId) + fields.push({ field: "mode", value: preferences.modeId }); + if (preferences.effortId) + fields.push({ field: "effort", value: preferences.effortId }); + if (preferences.personaId) + fields.push({ field: "persona", value: preferences.personaId }); + + for (const { field, value } of fields) { + const setResult = await host.withRuntime(() => + runtime.setCatalogField( + field, + bound.threadId, + bound.projectId, + bound.cwd, + bound.sessionId, + value + ) + ); + if (setResult.isErr()) { + // Some agents advertise catalog options but do not implement the + // matching ACP setter (e.g. session/set_model → Method not found). + // Skip that preference so first-send can still bind and prompt. + const message = coordinatorErrorMessage(setResult.error); + if (message.includes("Method not found")) continue; + return Result.err(setResult.error); + } + } + return Result.ok(undefined); +} + +/** + * Compound first-message operation: row → git → session → prefs → binding. + * On failure after the row exists, persists the user message and a thread_error + * entry and returns `bound: null` so the controller can navigate and retry. + */ +export async function startThread( + host: CoordinatorHost, + input: StartThreadInput +): Promise> { + const turnId = input.turnId ?? randomId(); + + // Persist branch only; worktree path is written after successful creation. + const created = await createThread(input.projectId, { + branch: input.branch, + }); + if (created.isErr()) { + return Result.err(coordinatorRepositoryError(created.error)); + } + const threadId = created.value.id; + + const git = await applyBranchOrWorktree( + threadId, + input.projectId, + input.branch, + input.worktree, + input.worktreePath + ); + if (git.isErr()) { + return failAfterRow(threadId, turnId, input.message, git.error); + } + + const cwd = await host.resolveCwd(threadId); + if (cwd.isErr()) { + return failAfterRow(threadId, turnId, input.message, cwd.error); + } + + const runtime = host.getAgent(input.agentName); + const session = await host.withRuntime(() => + runtime.createBoundSession(threadId, input.projectId, cwd.value) + ); + if (session.isErr()) { + return failAfterRow(threadId, turnId, input.message, session.error); + } + + const bound: BoundThread = { + threadId, + projectId: input.projectId, + agentName: input.agentName, + sessionId: session.value.sessionId, + cwd: cwd.value, + }; + + const prefs = await applyPreferences(host, bound, input.preferences); + if (prefs.isErr()) { + return failAfterRow(threadId, turnId, input.message, prefs.error); + } + + const persisted = await bindThreadAgent(threadId, input.projectId, { + agentName: input.agentName, + sessionId: session.value.sessionId, + }); + if (persisted.isErr()) { + return failAfterRow(threadId, turnId, input.message, persisted.error); + } + + const locked = await setAgentLocked(threadId); + if (locked.isErr()) { + return failAfterRow(threadId, turnId, input.message, locked.error); + } + + return Result.ok({ threadId, turnId, bound }); +} diff --git a/apps/cli/src/handlers/controller/chat.ts b/apps/cli/src/handlers/controller/chat.ts index b45d9e5..0a4a4d5 100644 --- a/apps/cli/src/handlers/controller/chat.ts +++ b/apps/cli/src/handlers/controller/chat.ts @@ -1,25 +1,11 @@ -import { appendConversation } from "@cyrus/database/repositories/conversations"; -import { - applyAgentThreadTitle, - ensureThread, - getThread, - setAgentLocked, -} from "@cyrus/database/repositories/threads"; +import { ensureThread, getThread } from "@cyrus/database/repositories/threads"; import { CoordinatorAgentNotBoundError } from "@cyrus/errors/coordinator"; import { throwOrpc } from "@cyrus/errors/orpc"; -import type { ChatChunk } from "@cyrus/schemas/rtc/chat"; import { formatPromptBlocks } from "@cyrus/schemas/rtc/chat"; import { randomId } from "@cyrus/utils/identity"; import { ORPCError } from "@orpc/server"; -import { log } from "evlog"; -import { runTurn } from "@/utils/run-turn"; -import { - isStreamingDelta, - resolvePersistEvent, - trackDelta, -} from "@/utils/streams"; -import { maybeApplyAutoThreadTitle } from "@/utils/thread-title"; import type { ControllerDeps } from "./deps"; +import { launchTurn } from "./turn-emit"; export function chatHandlers({ os, runtime }: ControllerDeps) { return { @@ -58,112 +44,46 @@ export function chatHandlers({ os, runtime }: ControllerDeps) { }); if (thread.isErr()) throwOrpc(thread.error); - const messageBuffers = new Map(); - const thoughtBuffers = new Map(); - - context.eventBus.ensureWatch(context.peerId, threadId); - - function publishChunk(chunk: ChatChunk): void { - context.eventBus.publish(chunk); - } - - async function applySessionTitleUpdate( - event: ChatChunk["event"] - ): Promise { - if ( - event.type !== "session_update" || - event.sessionUpdate !== "session_info_update" - ) { - return false; - } - const raw = event.raw as { title?: string | null } | undefined; - if (typeof raw?.title === "string" && raw.title.trim()) { - await applyAgentThreadTitle(threadId, raw.title); - } - return true; - } - - async function emit(event: ChatChunk["event"]): Promise { - if (await applySessionTitleUpdate(event)) return; - - trackDelta(event, messageBuffers, thoughtBuffers); - - if (isStreamingDelta(event)) - return publishChunk({ threadId, turnId, seq: 0, event }); - - const persistEvent = resolvePersistEvent( - event, - messageBuffers, - thoughtBuffers - ); - const entry = await appendConversation(threadId, { - threadId, - turnId, - event: persistEvent, - }); - if (entry.isErr()) throwOrpc(entry.error); - if (persistEvent.type === "user_message") { - const locked = await setAgentLocked(threadId); - if (locked.isErr()) throwOrpc(locked.error); - } - publishChunk(entry.value.chunk); - } - - async function emitTerminal( - event: Extract< - ChatChunk["event"], - { type: "turn_completed" | "turn_interrupted" } - > - ): Promise { - trackDelta(event, messageBuffers, thoughtBuffers); - - const persistEvent = resolvePersistEvent( - event, - messageBuffers, - thoughtBuffers - ); - const entry = await appendConversation(threadId, { - threadId, - turnId, - event: persistEvent, - }); - - if (entry.isOk()) { - publishChunk(entry.value.chunk); - if (event.type === "turn_completed") { - await maybeApplyAutoThreadTitle(threadId, turnId); - } - return; - } - - log.error({ - kind: "terminal_event_persist", - error: entry.error, - threadId, - turnId, - event: event.type, - }); - publishChunk({ threadId, turnId, seq: 0, event }); - } - - runTurn({ + launchTurn({ agentName, threadId, projectId, turnId, message, - emit, - emitTerminal, + context, runtime, - }) - .then((result) => { - result.tapError((error) => { - log.error({ kind: "chat_turn_failed", error, threadId, turnId }); - }); - }) - .catch((error) => { - log.error({ kind: "chat_turn_failed", error, threadId, turnId }); + }); + + return { threadId, turnId }; + }), + + startThread: os.startThread.handler(async ({ input, context }) => { + const turnId = input.turnId ?? randomId(); + const started = await runtime.threadCoordinator.startThread({ + projectId: input.projectId, + agentName: input.agentName, + message: input.message, + preferences: input.preferences, + branch: input.branch, + worktree: input.worktree, + worktreePath: input.worktreePath, + turnId, + }); + if (started.isErr()) throwOrpc(started.error); + + const { threadId, bound } = started.value; + + if (bound) + launchTurn({ + agentName: input.agentName, + threadId, + projectId: input.projectId, + turnId, + message: input.message, + context, + runtime, }); + else context.eventBus.ensureWatch(context.peerId, threadId); return { threadId, turnId }; }), diff --git a/apps/cli/src/handlers/controller/turn-emit.ts b/apps/cli/src/handlers/controller/turn-emit.ts new file mode 100644 index 0000000..d0f39d8 --- /dev/null +++ b/apps/cli/src/handlers/controller/turn-emit.ts @@ -0,0 +1,182 @@ +import type { RtcContext } from "@cyrus/connections/rtc/peer"; +import { appendConversation } from "@cyrus/database/repositories/conversations"; +import { + applyAgentThreadTitle, + setAgentLocked, +} from "@cyrus/database/repositories/threads"; +import { throwOrpc } from "@cyrus/errors/orpc"; +import type { ChatChunk, ChatMessage } from "@cyrus/schemas/rtc/chat"; +import { log } from "evlog"; +import type { WorkerRuntime } from "@/core"; +import { runTurn } from "@/utils/run-turn"; +import { + isStreamingDelta, + resolvePersistEvent, + trackDelta, +} from "@/utils/streams"; +import { maybeApplyAutoThreadTitle } from "@/utils/thread-title"; + +type TurnEmitters = { + emit: (event: ChatChunk["event"]) => Promise; + emitTerminal: ( + event: Extract< + ChatChunk["event"], + { type: "turn_completed" | "turn_interrupted" } + > + ) => Promise; +}; + +export function createTurnEmitters( + context: RtcContext, + threadId: string, + turnId: string +): TurnEmitters { + const messageBuffers = new Map(); + const thoughtBuffers = new Map(); + + function publishChunk(chunk: ChatChunk): void { + context.eventBus.publish(chunk); + } + + async function applySessionTitleUpdate( + event: ChatChunk["event"] + ): Promise { + if ( + event.type !== "session_update" || + event.sessionUpdate !== "session_info_update" + ) { + return false; + } + const raw = event.raw as { title?: string | null } | undefined; + if (typeof raw?.title === "string" && raw.title.trim()) { + await applyAgentThreadTitle(threadId, raw.title); + } + return true; + } + + async function emit(event: ChatChunk["event"]): Promise { + if (await applySessionTitleUpdate(event)) return; + + trackDelta(event, messageBuffers, thoughtBuffers); + + if (isStreamingDelta(event)) + return publishChunk({ threadId, turnId, seq: 0, event }); + + const persistEvent = resolvePersistEvent( + event, + messageBuffers, + thoughtBuffers + ); + const entry = await appendConversation(threadId, { + threadId, + turnId, + event: persistEvent, + }); + if (entry.isErr()) throwOrpc(entry.error); + if (persistEvent.type === "user_message") { + const locked = await setAgentLocked(threadId); + if (locked.isErr()) throwOrpc(locked.error); + } + publishChunk(entry.value.chunk); + } + + async function emitTerminal( + event: Extract< + ChatChunk["event"], + { type: "turn_completed" | "turn_interrupted" } + > + ): Promise { + trackDelta(event, messageBuffers, thoughtBuffers); + + const persistEvent = resolvePersistEvent( + event, + messageBuffers, + thoughtBuffers + ); + const entry = await appendConversation(threadId, { + threadId, + turnId, + event: persistEvent, + }); + + if (entry.isOk()) { + publishChunk(entry.value.chunk); + if (event.type === "turn_completed") { + await maybeApplyAutoThreadTitle(threadId, turnId); + } + return; + } + + log.error({ + kind: "terminal_event_persist", + error: entry.error, + threadId, + turnId, + event: event.type, + }); + publishChunk({ threadId, turnId, seq: 0, event }); + } + + return { emit, emitTerminal }; +} + +export function launchTurn(options: { + agentName: string; + threadId: string; + projectId: string; + turnId: string; + message: ChatMessage; + context: RtcContext; + runtime: WorkerRuntime; +}): void { + const { emit, emitTerminal } = createTurnEmitters( + options.context, + options.threadId, + options.turnId + ); + + options.context.eventBus.ensureWatch( + options.context.peerId, + options.threadId + ); + + let terminalPublished = false; + const trackedEmitTerminal: typeof emitTerminal = async (event) => { + await emitTerminal(event); + terminalPublished = true; + }; + + runTurn({ + agentName: options.agentName, + threadId: options.threadId, + projectId: options.projectId, + turnId: options.turnId, + message: options.message, + emit, + emitTerminal: trackedEmitTerminal, + runtime: options.runtime, + }) + .then((result) => { + result.tapError((error) => { + log.error({ + kind: "chat_turn_failed", + error, + threadId: options.threadId, + turnId: options.turnId, + }); + }); + }) + .catch((error) => { + log.error({ + kind: "chat_turn_failed", + error, + threadId: options.threadId, + turnId: options.turnId, + }); + if (!terminalPublished) { + trackedEmitTerminal({ type: "turn_interrupted" }).catch( + () => undefined + ); + } + }); +} diff --git a/apps/web/src/components/chat/composer/composer-branch-toolbar.tsx b/apps/web/src/components/chat/composer/composer-branch-toolbar.tsx index 35478d8..a59f95c 100644 --- a/apps/web/src/components/chat/composer/composer-branch-toolbar.tsx +++ b/apps/web/src/components/chat/composer/composer-branch-toolbar.tsx @@ -3,7 +3,10 @@ import { useCreateWorktree, useGitStatus, useListGitRefs, + useListProjectGitRefs, + useProjectGitStatus, } from "@cyrus/hooks/queries/use-git"; +import { useLocalDraftStore } from "@cyrus/hooks/stores/local-draft"; import type { Thread } from "@cyrus/schemas/rtc/threads"; import { cn } from "cnfast"; import { @@ -36,6 +39,7 @@ type WorkspaceMode = "local" | "worktree"; type ComposerBranchToolbarProps = { thread: Thread; + localDraft?: boolean; }; function resolveWorkspaceLabel( @@ -103,28 +107,56 @@ function BranchListItems({ )); } -export function ComposerBranchToolbar({ thread }: ComposerBranchToolbarProps) { - const gitStatus = useGitStatus(thread.id); - const gitRefs = useListGitRefs(thread.id); +export function ComposerBranchToolbar({ + thread, + localDraft = false, +}: ComposerBranchToolbarProps) { + const threadGitStatus = useGitStatus(localDraft ? undefined : thread.id); + const projectGitStatus = useProjectGitStatus( + localDraft ? thread.projectId : undefined + ); + const gitStatus = localDraft ? projectGitStatus : threadGitStatus; + const threadGitRefs = useListGitRefs(localDraft ? undefined : thread.id); + const projectGitRefs = useListProjectGitRefs( + localDraft ? thread.projectId : undefined + ); + const gitRefs = localDraft ? projectGitRefs : threadGitRefs; const checkoutRef = useCheckoutRef(); const createWorktree = useCreateWorktree(); - const [workspaceMode, setWorkspaceMode] = useState( - thread.worktreePath ? "worktree" : "local" - ); + const draftGit = useLocalDraftStore((state) => state.gitByDraft[thread.id]); + const setDraftBranch = useLocalDraftStore((state) => state.setBranch); + const setDraftWorktree = useLocalDraftStore((state) => state.setWorktree); + + const [workspaceMode, setWorkspaceMode] = useState(() => { + if (localDraft) return draftGit?.worktree ? "worktree" : "local"; + return thread.worktreePath ? "worktree" : "local"; + }); const [branchQuery, setBranchQuery] = useState(""); useEffect(() => { + if (localDraft) return; if (thread.worktreePath) setWorkspaceMode("worktree"); - }, [thread.worktreePath]); + }, [localDraft, thread.worktreePath]); + + useEffect(() => { + if (!localDraft) return; + setDraftWorktree(thread.id, workspaceMode === "worktree"); + }, [localDraft, setDraftWorktree, thread.id, workspaceMode]); const isRepo = gitStatus.data?.isRepo === true; - const refName = + const statusRefName = gitStatus.data?.isRepo === true ? gitStatus.data.refName : null; - const hasWorktree = Boolean(thread.worktreePath); + const refName = localDraft + ? (draftGit?.branch ?? statusRefName) + : statusRefName; + const hasWorktree = localDraft ? false : Boolean(thread.worktreePath); const envLocked = hasWorktree; - const branchMutationError = checkoutRef.error ?? createWorktree.error; - const isBranchActionPending = - checkoutRef.isPending || createWorktree.isPending; + const branchMutationError = localDraft + ? null + : (checkoutRef.error ?? createWorktree.error); + const isBranchActionPending = localDraft + ? false + : checkoutRef.isPending || createWorktree.isPending; const filteredRefs = useMemo(() => { const refs = gitRefs.data?.refs ?? []; @@ -136,9 +168,15 @@ export function ComposerBranchToolbar({ thread }: ComposerBranchToolbarProps) { if (!isRepo) return null; function handleBranchSelect(name: string) { + setBranchQuery(""); + + if (localDraft) { + setDraftBranch(thread.id, name); + return; + } + checkoutRef.reset(); createWorktree.reset(); - setBranchQuery(""); if (workspaceMode === "worktree" && !hasWorktree) { createWorktree.mutate({ threadId: thread.id, refName: name }); diff --git a/apps/web/src/components/chat/composer/index.tsx b/apps/web/src/components/chat/composer/index.tsx index 9df2fb1..7988c88 100644 --- a/apps/web/src/components/chat/composer/index.tsx +++ b/apps/web/src/components/chat/composer/index.tsx @@ -1,5 +1,8 @@ import { useAgentCatalog } from "@cyrus/hooks/agent-catalog/use-agent-catalog"; -import { useGitStatus } from "@cyrus/hooks/queries/use-git"; +import { + useGitStatus, + useProjectGitStatus, +} from "@cyrus/hooks/queries/use-git"; import { useListAgents } from "@cyrus/hooks/queries/use-list-agents"; import { useProjects } from "@cyrus/hooks/queries/use-projects"; import { useSearchEntries } from "@cyrus/hooks/queries/use-search-entries"; @@ -79,6 +82,7 @@ export function Composer({ threadError = null, pendingApprovals = EMPTY_APPROVALS, pendingElicitations = EMPTY_ELICITATIONS, + localDraft = false, }: { projectId: string; threadId: string; @@ -90,6 +94,7 @@ export function Composer({ threadError?: ErrorView | null; pendingApprovals?: ApprovalView[]; pendingElicitations?: ElicitationView[]; + localDraft?: boolean; }) { const agentsQuery = useListAgents(); const agents = agentsQuery.data?.agents ?? []; @@ -102,7 +107,12 @@ export function Composer({ projects.find((project) => project.id === projectId)?.cwd ?? ""; const threadCwd = thread.worktreePath ?? projectCwd; - const catalog = useAgentCatalog({ agents, projectId, threadId }); + const catalog = useAgentCatalog({ + agents, + localDraft, + projectId, + threadId, + }); const supportsEmbeddedContext = catalog.capabilities == null || catalog.promptCapabilities.embeddedContext !== false; @@ -110,8 +120,12 @@ export function Composer({ const canPasteUrls = supportsEmbeddedContext; const composerBlocked = Boolean(threadError ?? catalog.bindError); - const gitStatus = useGitStatus(threadId); - const isGitRepo = gitStatus.data?.isRepo === true; + const threadGitStatus = useGitStatus(localDraft ? undefined : threadId); + const projectGitStatus = useProjectGitStatus( + localDraft ? projectId : undefined + ); + const isGitRepo = + (localDraft ? projectGitStatus : threadGitStatus).data?.isRepo === true; const { setValue: setDraft, clear: clearDraft } = useComposerDraft(threadId); const draftHydrated = useComposerDraftHydrated(); const [plainText, setPlainText] = useState(""); @@ -548,7 +562,13 @@ export function Composer({ : "pb-[calc(env(safe-area-inset-bottom)+0.75rem)] sm:pb-[calc(env(safe-area-inset-bottom)+1rem)]" )} > - {isGitRepo ? : null} + {isGitRepo ? ( + + ) : null} ); diff --git a/apps/web/src/components/chat/main/draft-workspace.tsx b/apps/web/src/components/chat/main/draft-workspace.tsx new file mode 100644 index 0000000..c4a3fef --- /dev/null +++ b/apps/web/src/components/chat/main/draft-workspace.tsx @@ -0,0 +1,140 @@ +import { coordinatorRuntimeError } from "@cyrus/errors/coordinator"; +import { useStartThread } from "@cyrus/hooks/queries/use-start-thread"; +import { useAgentCatalogStore } from "@cyrus/hooks/stores/agent-catalog"; +import { useComposerDraftStore } from "@cyrus/hooks/stores/composer-draft"; +import { + discardLocalDraft, + useLocalDraftStore, +} from "@cyrus/hooks/stores/local-draft"; +import type { ChatMessage } from "@cyrus/schemas/rtc/chat"; +import type { Thread } from "@cyrus/schemas/rtc/threads"; +import { randomId } from "@cyrus/utils/identity"; +import { useNavigate } from "@tanstack/react-router"; +import { useEffect, useMemo } from "react"; +import { Composer } from "@/components/chat/composer"; +import { ChatFeed } from "@/components/chat/feed/chat-feed"; +import { ThreadHeader } from "@/components/chat/main/thread-header"; + +type DraftWorkspaceProps = { + workerId: string; + projectId: string; + draftId: string; +}; + +const EMPTY_CONVERSATION = { + approvals: [], + diffs: [], + elicitations: [], + errors: [], + messages: [], + thoughts: [], + toolCalls: [], + turns: [], +}; + +function clearDraftControllerState(draftId: string): void { + discardLocalDraft(draftId); + useComposerDraftStore.getState().clearDraft(draftId); + useAgentCatalogStore.getState().clearPendingAgent(draftId); +} + +export function DraftWorkspace({ + workerId, + projectId, + draftId, +}: DraftWorkspaceProps) { + const navigate = useNavigate(); + const startThread = useStartThread(projectId); + const gitChoice = useLocalDraftStore((state) => state.gitByDraft[draftId]); + + useEffect( + () => () => { + clearDraftControllerState(draftId); + }, + [draftId] + ); + + const draftThread = useMemo( + () => ({ + id: draftId, + projectId, + name: "New thread", + agentName: undefined, + sessionId: undefined, + agentLocked: undefined, + titleSource: null, + createdAt: new Date().toISOString(), + updatedAt: new Date().toISOString(), + }), + [draftId, projectId] + ); + + async function handleSend(message: ChatMessage) { + const catalog = useAgentCatalogStore.getState(); + const agentName = + catalog.pendingAgentByThread[draftId] ?? + catalog.liveBindingByThread[draftId]?.agentName; + if (!agentName) { + // TanStack mutation / composer boundary — throw TaggedError. + throw coordinatorRuntimeError("no agent selected"); + } + + const selection = catalog.selectionByThread[draftId] ?? {}; + const turnId = randomId(); + // mutateAsync already rejects with the oRPC/domain error — do not wrap + // in Result.tryPromise (that would erase the TaggedError into UnhandledException). + const started = await startThread.mutateAsync({ + projectId, + agentName, + message, + turnId, + branch: gitChoice?.branch, + worktree: gitChoice?.worktree, + preferences: { + modelId: selection.modelId, + modeId: selection.modeId, + effortId: selection.effortId, + personaId: selection.personaId, + }, + }); + + clearDraftControllerState(draftId); + await navigate({ + to: "/workers/$workerId/p/$projectId/t/$threadId", + params: { + workerId, + projectId, + threadId: started.threadId, + }, + }); + } + + return ( + <> + + +
+
+ + +
+
+ + ); +} diff --git a/apps/web/src/components/chat/main/thread-header.tsx b/apps/web/src/components/chat/main/thread-header.tsx index 2b4b6be..9cfe42b 100644 --- a/apps/web/src/components/chat/main/thread-header.tsx +++ b/apps/web/src/components/chat/main/thread-header.tsx @@ -1,6 +1,7 @@ import { useGitStatus, useInitGitRepository, + useProjectGitStatus, } from "@cyrus/hooks/queries/use-git"; import { useProjects } from "@cyrus/hooks/queries/use-projects"; import type { Thread } from "@cyrus/schemas/rtc/threads"; @@ -22,22 +23,30 @@ type ThreadHeaderProps = { thread: Thread; workerId: string; projectId: string; + localDraft?: boolean; }; export function ThreadHeader({ thread, workerId, projectId, + localDraft = false, }: ThreadHeaderProps) { const { diffOpen, toggleDiffOpen } = useChatUiStore(); const { projects } = useProjects(); const project = projects.find((item) => item.id === projectId); - const gitStatus = useGitStatus(thread.id); + const threadGitStatus = useGitStatus(localDraft ? undefined : thread.id); + const projectGitStatus = useProjectGitStatus( + localDraft ? projectId : undefined + ); + const gitStatus = localDraft ? projectGitStatus : threadGitStatus; const initGitRepository = useInitGitRepository(); const isRepo = gitStatus.data?.isRepo === true; function renderGitAction() { + if (localDraft) return null; + if (isRepo) return (