diff --git a/apps/cli/__tests__/integration/draft-session-lifecycle.test.ts b/apps/cli/__tests__/integration/draft-session-lifecycle.test.ts index 11209fd..4dc80c5 100644 --- a/apps/cli/__tests__/integration/draft-session-lifecycle.test.ts +++ b/apps/cli/__tests__/integration/draft-session-lifecycle.test.ts @@ -39,8 +39,8 @@ function createMockSession(sessionId: string): RuntimeSession { const sessions: RuntimeSession[] = []; -mock.module("@cyrus/database/repositories/projects", () => ({ - resolveProjectCwd: async () => Result.ok("/tmp/project"), +mock.module("@cyrus/database/repositories/git", () => ({ + resolveThreadGitCwd: async () => Result.ok("/tmp/project"), })); mock.module("@cyrus/database/repositories/threads", () => ({ @@ -54,6 +54,11 @@ mock.module("@cyrus/database/repositories/threads", () => ({ threadState.sessionId = data.sessionId; return Promise.resolve(Result.ok({ ...threadState })); }, + clearThreadDraftBinding: () => { + threadState.agentName = undefined; + threadState.sessionId = undefined; + return Promise.resolve(Result.ok({ ...threadState })); + }, })); function createCoordinator() { @@ -81,7 +86,7 @@ describe("draft session lifecycle", () => { threadState.agentLocked = undefined; }); - test("bind then catalog then prompt reuses the same session id", async () => { + test("bind keeps session in memory until persistBoundSession", async () => { const coordinator = createCoordinator(); const bound = await coordinator.bindAgent( @@ -93,12 +98,22 @@ describe("draft session lifecycle", () => { if (bound.isErr()) throw new Error("expected bind to succeed"); expect(bound.value.sessionId).toBe("session-1"); expect(bound.value.capabilities).toEqual({ loadSession: true }); + expect(threadState.sessionId).toBeUndefined(); + expect(threadState.agentName).toBeUndefined(); const models = await coordinator.getModels("thread-1"); expect(models.isOk()).toBe(true); if (models.isErr()) throw new Error("expected models to succeed"); expect(models.value[0]?.id).toBe("model-1"); + const persisted = await coordinator.persistBoundSession( + "thread-1", + "project-1" + ); + expect(persisted.isOk()).toBe(true); + expect(threadState.sessionId).toBe("session-1"); + expect(threadState.agentName).toBe("mock-agent"); + const prompt = await coordinator.prompt( "mock-agent", "thread-1", @@ -153,4 +168,22 @@ describe("draft session lifecycle", () => { expect(sessions[0]?.close).toHaveBeenCalled(); }); + + test("clears stale draft db binding without resuming it", async () => { + threadState.agentName = "mock-agent"; + threadState.sessionId = "stale-session"; + threadState.agentLocked = undefined; + + const coordinator = createCoordinator(); + const bound = await coordinator.bindAgent( + "thread-1", + "project-1", + "mock-agent" + ); + expect(bound.isOk()).toBe(true); + if (bound.isErr()) throw new Error("expected bind to succeed"); + expect(bound.value.sessionId).toBe("session-1"); + expect(threadState.sessionId).toBeUndefined(); + expect(threadState.agentName).toBeUndefined(); + }); }); diff --git a/apps/cli/src/core/agents/catalog.ts b/apps/cli/src/core/agents/catalog.ts index 9377c10..312475f 100644 --- a/apps/cli/src/core/agents/catalog.ts +++ b/apps/cli/src/core/agents/catalog.ts @@ -132,3 +132,27 @@ function flattenSelectOptions( } return flattened; } + +function selectOptionValues( + options: SessionConfigSelectOptions +): SessionConfigSelectOption["value"][] { + return flattenSelectOptions(options).map((option) => option.value); +} + +export function reconcileInvalidSelectConfigOptions( + options: SessionConfigOption[] +): Array<{ configId: string; value: string }> { + const resets: Array<{ configId: string; value: string }> = []; + + for (const option of options) { + if (option.type !== "select") continue; + const validValues = new Set(selectOptionValues(option.options)); + if (validValues.size === 0) continue; + if (validValues.has(option.currentValue)) continue; + const fallback = [...validValues][0]; + if (!fallback) continue; + resets.push({ configId: option.id, value: fallback }); + } + + return resets; +} diff --git a/apps/cli/src/core/agents/runtime.ts b/apps/cli/src/core/agents/runtime.ts index 2bf7a8b..4220af1 100644 --- a/apps/cli/src/core/agents/runtime.ts +++ b/apps/cli/src/core/agents/runtime.ts @@ -13,6 +13,7 @@ import { modelsFromSession, modesFromSession, personasFromSession, + reconcileInvalidSelectConfigOptions, } from "./catalog"; import { mapPromptBlocksToAcp } from "./prompt"; @@ -178,6 +179,38 @@ export class AgentRuntime { sessionId ); await session.setModel(modelId); + // Model is already applied — reconcile is best-effort so a dependent + // reset failure does not report setModel as failed (client refreshes on ok). + await Result.tryPromise(() => + this.reconcileDependentConfigOptions( + threadId, + projectId, + cwd, + sessionId, + session + ) + ); + } + + private async reconcileDependentConfigOptions( + threadId: string, + projectId: string, + cwd: string, + sessionId: string, + session: RuntimeSession + ): Promise { + const resets = reconcileInvalidSelectConfigOptions( + session.transcript.session.configOptions + ); + for (const reset of resets) + await this.setConfigOption( + threadId, + projectId, + cwd, + sessionId, + reset.configId, + reset.value + ); } async setMode( @@ -326,6 +359,20 @@ export class AgentRuntime { }); } + getLiveSession(threadId: string): { + sessionId: string; + projectId: string; + cwd: string; + } | null { + const entry = this.sessions.get(threadId); + if (!entry) return null; + return { + sessionId: entry.session.sessionId, + projectId: entry.projectId, + cwd: entry.cwd, + }; + } + async cancel(threadId: string): Promise { const session = this.sessions.get(threadId)?.session; if (!session) return; diff --git a/apps/cli/src/core/threads/coordinator.ts b/apps/cli/src/core/threads/coordinator.ts index b4d11f5..ecc2832 100644 --- a/apps/cli/src/core/threads/coordinator.ts +++ b/apps/cli/src/core/threads/coordinator.ts @@ -1,6 +1,7 @@ import { resolveThreadGitCwd } from "@cyrus/database/repositories/git"; import { bindThreadAgent, + clearThreadDraftBinding, getThread, } from "@cyrus/database/repositories/threads"; import { @@ -33,11 +34,34 @@ type BoundThread = { export class ThreadCoordinator { private readonly agents = new Map(); private readonly pool: AgentPool; + private readonly threadLocks = new Map>(); constructor(pool: AgentPool) { this.pool = pool; } + private async withThreadLock( + threadId: string, + fn: () => Promise + ): Promise { + const previous = this.threadLocks.get(threadId) ?? Promise.resolve(); + let release!: () => void; + const gate = new Promise((resolve) => { + release = resolve; + }); + const held = previous.catch(() => undefined).then(() => gate); + this.threadLocks.set(threadId, held); + await previous.catch(() => undefined); + try { + return await fn(); + } finally { + release(); + if (this.threadLocks.get(threadId) === held) { + this.threadLocks.delete(threadId); + } + } + } + private getAgent(agentName: string): AgentRuntime { let runtime = this.agents.get(agentName); if (!runtime) { @@ -47,6 +71,21 @@ export class ThreadCoordinator { return runtime; } + private findLiveBinding(threadId: string): BoundThread | null { + for (const [agentName, runtime] of this.agents) { + const live = runtime.getLiveSession(threadId); + if (!live) continue; + return { + threadId, + agentName, + sessionId: live.sessionId, + projectId: live.projectId, + cwd: live.cwd, + }; + } + return null; + } + private async withRuntime( fn: () => Promise ): Promise> { @@ -57,10 +96,42 @@ export class ThreadCoordinator { ); } + private async catalogForSession( + runtime: AgentRuntime, + threadId: string, + projectId: string, + cwd: string, + sessionId: string + ): Promise<{ + capabilities: Record; + models: ModelOption[]; + modes: SelectOption[]; + efforts: SelectOption[]; + personas: SelectOption[]; + }> { + return { + capabilities: await runtime.getAgentCapabilities(), + models: await runtime.getModels(threadId, projectId, cwd, sessionId), + modes: await runtime.getModes(threadId, projectId, cwd, sessionId), + efforts: await runtime.getEfforts(threadId, projectId, cwd, sessionId), + personas: await runtime.getPersonas(threadId, projectId, cwd, sessionId), + }; + } + async bindAgent( threadId: string, projectId: string, agentName: string + ): Promise> { + return await this.withThreadLock(threadId, () => + this.bindAgentLocked(threadId, projectId, agentName) + ); + } + + private async bindAgentLocked( + threadId: string, + projectId: string, + agentName: string ): Promise> { const thread = await getThread(threadId); if (thread.isErr()) @@ -77,96 +148,195 @@ export class ThreadCoordinator { if (cwd.isErr()) return Result.err(cwd.error); const runtime = this.getAgent(agentName); + const live = this.findLiveBinding(threadId); + + if (thread.value.agentLocked) { + return this.bindLockedThread({ + threadId, + projectId, + agentName, + cwd: cwd.value, + runtime, + live, + sessionId: thread.value.sessionId, + agentNameOnThread: thread.value.agentName, + }); + } - if (thread.value.agentName === agentName && thread.value.sessionId) { - const sessionId = thread.value.sessionId; - const catalog = await this.withRuntime(async () => ({ - capabilities: await runtime.getAgentCapabilities(), - models: await runtime.getModels( - threadId, - projectId, - cwd.value, - sessionId - ), - modes: await runtime.getModes( - threadId, - projectId, - cwd.value, - sessionId - ), - efforts: await runtime.getEfforts( - threadId, - projectId, - cwd.value, - sessionId - ), - personas: await runtime.getPersonas( - threadId, - projectId, - cwd.value, - sessionId - ), - })); - if (catalog.isOk()) + return this.bindDraftThread({ + threadId, + projectId, + agentName, + cwd: cwd.value, + runtime, + live, + staleAgentName: thread.value.agentName, + staleSessionId: thread.value.sessionId, + }); + } + + private async bindLockedThread(params: { + threadId: string; + projectId: string; + agentName: string; + cwd: string; + runtime: AgentRuntime; + live: BoundThread | null; + sessionId: string | undefined; + agentNameOnThread: string | undefined; + }): Promise> { + const sessionId = params.live?.sessionId ?? params.sessionId; + if (!(sessionId && params.agentNameOnThread)) { + return Result.err(coordinatorAgentNotBound()); + } + + const catalog = await this.withRuntime(() => + this.catalogForSession( + params.runtime, + params.threadId, + params.projectId, + params.cwd, + sessionId + ) + ); + if (catalog.isErr()) return Result.err(catalog.error); + + return Result.ok({ + sessionId, + agentName: params.agentName, + agentLocked: true, + ...catalog.value, + commands: params.runtime.getAvailableCommands(params.threadId), + }); + } + + private async bindDraftThread(params: { + threadId: string; + projectId: string; + agentName: string; + cwd: string; + runtime: AgentRuntime; + live: BoundThread | null; + staleAgentName: string | undefined; + staleSessionId: string | undefined; + }): Promise> { + const { live } = params; + + if (live?.agentName === params.agentName) { + const catalog = await this.withRuntime(() => + this.catalogForSession( + params.runtime, + params.threadId, + params.projectId, + params.cwd, + live.sessionId + ) + ); + if (catalog.isOk()) { return Result.ok({ - sessionId, - agentName, - agentLocked: thread.value.agentLocked, + sessionId: live.sessionId, + agentName: params.agentName, + agentLocked: undefined, ...catalog.value, - commands: runtime.getAvailableCommands(threadId), + commands: params.runtime.getAvailableCommands(params.threadId), }); - - if (thread.value.agentLocked) return Result.err(catalog.error); + } } - const previousAgentName = thread.value.agentName; - const previousSessionId = thread.value.sessionId; - if (previousSessionId && previousAgentName) { + if (live) { const closed = await this.withRuntime(() => - previousAgentName === agentName - ? runtime.closeSession(previousSessionId, threadId) - : this.getAgent(previousAgentName).closeSession( - previousSessionId, - threadId - ) + this.getAgent(live.agentName).closeSession( + live.sessionId, + params.threadId + ) ); if (closed.isErr()) return Result.err(closed.error); } + if (params.staleAgentName || params.staleSessionId) { + const cleared = await clearThreadDraftBinding( + params.threadId, + params.projectId + ); + if (cleared.isErr()) + return Result.err(coordinatorRepositoryError(cleared.error)); + } + const bound = await this.withRuntime(async () => { - const session = await runtime.createBoundSession( - threadId, - projectId, - cwd.value + const session = await params.runtime.createBoundSession( + params.threadId, + params.projectId, + params.cwd ); return { session, - capabilities: await runtime.getAgentCapabilities(), + capabilities: await params.runtime.getAgentCapabilities(), ...catalogSnapshotFromSession(session), }; }); if (bound.isErr()) return Result.err(bound.error); - const persisted = await bindThreadAgent(threadId, projectId, { - agentName, - sessionId: bound.value.session.sessionId, - }); - if (persisted.isErr()) { - await runtime.closeSession(bound.value.session.sessionId, threadId); - return Result.err(coordinatorRepositoryError(persisted.error)); - } - return Result.ok({ sessionId: bound.value.session.sessionId, - agentName, - agentLocked: persisted.value.agentLocked, + agentName: params.agentName, + agentLocked: undefined, capabilities: bound.value.capabilities, models: bound.value.models, modes: bound.value.modes, efforts: bound.value.efforts, personas: bound.value.personas, - commands: runtime.commandsFromSession(bound.value.session), + commands: params.runtime.commandsFromSession(bound.value.session), + }); + } + + /** Persist live draft binding on first user message. No-op if already stored. */ + async persistBoundSession( + threadId: string, + projectId: string, + expectedAgentName?: string + ): Promise> { + return await this.withThreadLock(threadId, () => + this.persistBoundSessionLocked(threadId, projectId, expectedAgentName) + ); + } + + private async persistBoundSessionLocked( + threadId: string, + projectId: string, + expectedAgentName?: string + ): Promise> { + const bound = await this.resolveBoundThread(threadId, projectId); + if (bound.isErr()) return Result.err(bound.error); + + if (expectedAgentName && bound.value.agentName !== expectedAgentName) { + return Result.err( + coordinatorAgentMismatch(bound.value.agentName, expectedAgentName) + ); + } + + const thread = await getThread(threadId); + if (thread.isErr()) + return Result.err(coordinatorRepositoryError(thread.error)); + if (!thread.value || thread.value.projectId !== projectId) { + return Result.err(coordinatorNotFound("thread", threadId)); + } + + if ( + thread.value.agentName === bound.value.agentName && + thread.value.sessionId === bound.value.sessionId + ) { + return Result.ok(bound.value); + } + + const persisted = await bindThreadAgent(threadId, projectId, { + agentName: bound.value.agentName, + sessionId: bound.value.sessionId, }); + if (persisted.isErr()) { + return Result.err(coordinatorRepositoryError(persisted.error)); + } + + return Result.ok(bound.value); } async getModels( @@ -346,7 +516,30 @@ export class ThreadCoordinator { sessionId: string, agentName: string ): Promise { - await this.getAgent(agentName).closeSession(sessionId, threadId); + await this.withThreadLock(threadId, () => + this.getAgent(agentName).closeSession(sessionId, threadId) + ); + } + + async closeAnyThreadSession(threadId: string): Promise { + await this.withThreadLock(threadId, async () => { + const live = this.findLiveBinding(threadId); + if (live) { + await this.getAgent(live.agentName).closeSession( + live.sessionId, + threadId + ); + return; + } + + const thread = await getThread(threadId); + if (thread.isErr() || !thread.value) return; + if (!(thread.value.sessionId && thread.value.agentName)) return; + await this.getAgent(thread.value.agentName).closeSession( + thread.value.sessionId, + threadId + ); + }); } private async resolveCwd( @@ -362,6 +555,14 @@ export class ThreadCoordinator { threadId: string, projectId?: string ): Promise> { + const live = this.findLiveBinding(threadId); + if (live) { + if (projectId && live.projectId !== projectId) { + return Result.err(coordinatorNotFound("thread", threadId)); + } + return Result.ok(live); + } + const thread = await getThread(threadId); if (thread.isErr()) return Result.err(coordinatorRepositoryError(thread.error)); @@ -371,7 +572,15 @@ export class ThreadCoordinator { if (projectId && thread.value.projectId !== projectId) { return Result.err(coordinatorNotFound("thread", threadId)); } - if (!(thread.value.sessionId && thread.value.agentName)) { + + // Only resume sessions that were committed by a first user message. + if ( + !( + thread.value.agentLocked && + thread.value.sessionId && + thread.value.agentName + ) + ) { return Result.err(coordinatorAgentNotBound()); } diff --git a/apps/cli/src/handlers/controller/agents.ts b/apps/cli/src/handlers/controller/agents.ts index 9a5be4c..00ea195 100644 --- a/apps/cli/src/handlers/controller/agents.ts +++ b/apps/cli/src/handlers/controller/agents.ts @@ -1,24 +1,42 @@ +import { + CoordinatorAgentLockedError, + CoordinatorAgentMismatchError, + CoordinatorAgentNotBoundError, + CoordinatorNotFoundError, +} from "@cyrus/errors/coordinator"; import { throwOrpc } from "@cyrus/errors/orpc"; import { listHealthyAgents } from "@/core/agents/health"; +import { persistCoordinatorThreadError } from "@/utils/thread-errors"; import type { ControllerDeps } from "./deps"; +function shouldPersistBindError(error: unknown): boolean { + return !( + CoordinatorAgentLockedError.is(error) || + CoordinatorAgentMismatchError.is(error) || + CoordinatorAgentNotBoundError.is(error) || + CoordinatorNotFoundError.is(error) + ); +} + export function agentsHandlers({ os, runtime }: ControllerDeps) { return { listAgents: os.listAgents.handler(async () => ({ agents: await listHealthyAgents(), })), - bindAgent: os.bindAgent.handler(async ({ input }) => - ( - await runtime.threadCoordinator.bindAgent( - input.threadId, - input.projectId, - input.agentName - ) - ).match({ - ok: (output) => output, - err: throwOrpc, - }) - ), + bindAgent: os.bindAgent.handler(async ({ input }) => { + const result = await runtime.threadCoordinator.bindAgent( + input.threadId, + input.projectId, + input.agentName + ); + if (result.isErr()) { + if (shouldPersistBindError(result.error)) { + await persistCoordinatorThreadError(input.threadId, result.error); + } + throwOrpc(result.error); + } + return result.value; + }), }; } diff --git a/apps/cli/src/handlers/controller/chat.ts b/apps/cli/src/handlers/controller/chat.ts index cc41cc7..55593ab 100644 --- a/apps/cli/src/handlers/controller/chat.ts +++ b/apps/cli/src/handlers/controller/chat.ts @@ -1,9 +1,11 @@ import { appendConversation } from "@cyrus/database/repositories/conversations"; import { + applyAgentThreadTitle, ensureThread, getThread, setAgentLocked, } 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"; @@ -16,6 +18,7 @@ import { resolvePersistEvent, trackDelta, } from "@/utils/streams"; +import { maybeApplyAutoThreadTitle } from "@/utils/thread-title"; import type { ControllerDeps } from "./deps"; export function chatHandlers({ os, runtime }: ControllerDeps) { @@ -31,15 +34,24 @@ export function chatHandlers({ os, runtime }: ControllerDeps) { const existing = await getThread(threadId); if (existing.isErr()) throwOrpc(existing.error); - if (!(existing.value?.sessionId && existing.value.agentName)) - throw new ORPCError("BAD_REQUEST", { - message: "agent must be bound before chat; call bindAgent first", + if (!existing.value) + throw new ORPCError("NOT_FOUND", { + message: `thread ${threadId} not found`, }); - if (existing.value.agentName !== agentName) - throw new ORPCError("BAD_REQUEST", { - message: "agentName does not match the bound thread agent", - }); + const persisted = await runtime.threadCoordinator.persistBoundSession( + threadId, + projectId, + agentName + ); + if (persisted.isErr()) { + if (CoordinatorAgentNotBoundError.is(persisted.error)) { + throw new ORPCError("BAD_REQUEST", { + message: "agent must be bound before chat; call bindAgent first", + }); + } + throwOrpc(persisted.error); + } const thread = await ensureThread(threadId, projectId, { firstMessage: formatPromptBlocks(message), @@ -55,7 +67,25 @@ export function chatHandlers({ os, runtime }: ControllerDeps) { 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)) @@ -100,6 +130,9 @@ export function chatHandlers({ os, runtime }: ControllerDeps) { if (entry.isOk()) { publishChunk(entry.value.chunk); + if (event.type === "turn_completed") { + await maybeApplyAutoThreadTitle(threadId, turnId); + } return; } diff --git a/apps/cli/src/handlers/controller/projects.ts b/apps/cli/src/handlers/controller/projects.ts index e1e9ed4..343bc7e 100644 --- a/apps/cli/src/handlers/controller/projects.ts +++ b/apps/cli/src/handlers/controller/projects.ts @@ -40,13 +40,7 @@ export function projectsHandlers({ os, runtime }: ControllerDeps) { if (listed.isErr()) throwOrpc(listed.error); for (const thread of listed.value) { - if (thread.sessionId && thread.agentName) { - await runtime.threadCoordinator.closeThreadSession( - thread.id, - thread.sessionId, - thread.agentName - ); - } + await runtime.threadCoordinator.closeAnyThreadSession(thread.id); } const deleted = await deleteStoredProject(input.projectId); diff --git a/apps/cli/src/handlers/controller/threads.ts b/apps/cli/src/handlers/controller/threads.ts index 7f0c342..20758b8 100644 --- a/apps/cli/src/handlers/controller/threads.ts +++ b/apps/cli/src/handlers/controller/threads.ts @@ -70,13 +70,7 @@ export function threadsHandlers({ os, runtime }: ControllerDeps) { if (thread.isErr()) throwOrpc(thread.error); if (!thread.value) throwOrpc(notFound("thread", input.threadId)); - if (thread.value.sessionId && thread.value.agentName) { - await runtime.threadCoordinator.closeThreadSession( - input.threadId, - thread.value.sessionId, - thread.value.agentName - ); - } + await runtime.threadCoordinator.closeAnyThreadSession(input.threadId); if (thread.value.worktreePath) { const projectCwd = await resolveProjectCwd(thread.value.projectId); diff --git a/apps/cli/src/utils/run-turn.test.ts b/apps/cli/src/utils/run-turn.test.ts index b62080f..d10f1b9 100644 --- a/apps/cli/src/utils/run-turn.test.ts +++ b/apps/cli/src/utils/run-turn.test.ts @@ -1,4 +1,5 @@ import { describe, expect, test } from "bun:test"; +import { coordinatorRuntimeError } from "@cyrus/errors/coordinator"; import type { ChatChunk } from "@cyrus/schemas/rtc/chat"; import { Result } from "better-result"; import { runTurn } from "./run-turn"; @@ -49,13 +50,20 @@ describe("runTurn", () => { test("emits interrupted terminal event when initial emit fails", async () => { const terminal: ChatChunk["event"][] = []; + const emitted: ChatChunk["event"][] = []; const result = await runTurn({ agentName: "claude", threadId: "thread-1", projectId: "project-1", message: textMessage("hello"), - emit: () => Promise.reject(new Error("emit failed")), + emit: (event) => { + if (event.type === "user_message") { + return Promise.reject(new Error("emit failed")); + } + emitted.push(event); + return Promise.resolve(); + }, emitTerminal: (event) => { terminal.push(event); return Promise.resolve(); @@ -73,18 +81,61 @@ describe("runTurn", () => { }); expect(result.isErr()).toBe(true); + expect(emitted).toContainEqual({ + type: "thread_error", + message: "emit failed", + code: "turn.emit_failed", + }); + expect(terminal).toEqual([{ type: "turn_interrupted" }]); + }); + + test("emits thread error and interrupted terminal event when prompt fails", async () => { + const terminal: ChatChunk["event"][] = []; + const emitted: ChatChunk["event"][] = []; + + const result = await runTurn({ + agentName: "claude", + threadId: "thread-1", + projectId: "project-1", + message: textMessage("hello"), + emit: (event) => { + emitted.push(event); + return Promise.resolve(); + }, + emitTerminal: (event) => { + terminal.push(event); + return Promise.resolve(); + }, + runtime: { + threadCoordinator: { + prompt: async () => + Result.err(coordinatorRuntimeError("prompt failed")), + }, + } as never, + }); + + expect(result.isErr()).toBe(true); + expect(emitted).toContainEqual({ + type: "thread_error", + message: "prompt failed", + code: "coordinator.runtime", + }); expect(terminal).toEqual([{ type: "turn_interrupted" }]); }); - test("emits interrupted terminal event when prompt fails", async () => { + test("emits interrupted terminal event when streamed prompt fails", async () => { const terminal: ChatChunk["event"][] = []; + const emitted: ChatChunk["event"][] = []; const result = await runTurn({ agentName: "claude", threadId: "thread-1", projectId: "project-1", message: textMessage("hello"), - emit: () => Promise.resolve(), + emit: (event) => { + emitted.push(event); + return Promise.resolve(); + }, emitTerminal: (event) => { terminal.push(event); return Promise.resolve(); @@ -104,6 +155,11 @@ describe("runTurn", () => { }); expect(result.isErr()).toBe(true); + expect(emitted).toContainEqual({ + type: "thread_error", + message: "agent failed", + code: "turn.stream_failed", + }); expect(terminal).toEqual([{ type: "turn_interrupted" }]); }); }); diff --git a/apps/cli/src/utils/run-turn.ts b/apps/cli/src/utils/run-turn.ts index 4d80d19..87dddc8 100644 --- a/apps/cli/src/utils/run-turn.ts +++ b/apps/cli/src/utils/run-turn.ts @@ -1,7 +1,20 @@ +import type { CoordinatorError } from "@cyrus/errors/coordinator"; +import { + type TurnError, + turnEmitFailed, + turnErrorMessageFromUnknown, + turnStreamFailed, +} from "@cyrus/errors/turn"; import type { ChatChunk, ChatMessage } from "@cyrus/schemas/rtc/chat"; import { formatPromptBlocks } from "@cyrus/schemas/rtc/chat"; import { Result } from "better-result"; import type { WorkerRuntime } from "@/core"; +import { + coordinatorErrorCode, + coordinatorErrorMessage, +} from "@/utils/thread-errors"; + +type RunTurnError = TurnError | CoordinatorError; type RunTurnOptions = { agentName: string; @@ -26,7 +39,7 @@ export async function runTurn({ emit, emitTerminal, runtime, -}: RunTurnOptions): Promise> { +}: RunTurnOptions): Promise> { const started = await Result.tryPromise(async () => { await emit({ type: "user_message", @@ -36,8 +49,14 @@ export async function runTurn({ await emit({ type: "thread_started", threadId }); }); if (started.isErr()) { + const error = turnEmitFailed(turnErrorMessageFromUnknown(started.error)); + await emit({ + type: "thread_error", + message: error.message, + code: error._tag, + }); await emitTerminal({ type: "turn_interrupted" }); - return started; + return Result.err(error); } const promptResult = await runtime.threadCoordinator.prompt( @@ -47,6 +66,11 @@ export async function runTurn({ message ); if (promptResult.isErr()) { + await emit({ + type: "thread_error", + message: coordinatorErrorMessage(promptResult.error), + code: coordinatorErrorCode(promptResult.error), + }); await emitTerminal({ type: "turn_interrupted" }); return Result.err(promptResult.error); } @@ -56,8 +80,14 @@ export async function runTurn({ }); if (streamed.isErr()) { + const error = turnStreamFailed(turnErrorMessageFromUnknown(streamed.error)); + await emit({ + type: "thread_error", + message: error.message, + code: error._tag, + }); await emitTerminal({ type: "turn_interrupted" }); - return streamed; + return Result.err(error); } await emitTerminal({ type: "turn_completed" }); diff --git a/apps/cli/src/utils/thread-errors.ts b/apps/cli/src/utils/thread-errors.ts new file mode 100644 index 0000000..97a45b3 --- /dev/null +++ b/apps/cli/src/utils/thread-errors.ts @@ -0,0 +1,61 @@ +import { appendConversation } from "@cyrus/database/repositories/conversations"; +import { getThread } from "@cyrus/database/repositories/threads"; +import { isCoordinatorError } from "@cyrus/errors/coordinator"; +import { isTurnError } from "@cyrus/errors/turn"; +import type { ChatChunk } from "@cyrus/schemas/rtc/chat"; +import { randomId } from "@cyrus/utils/identity"; + +export function coordinatorErrorCode(error: unknown): string | undefined { + if (isCoordinatorError(error) || isTurnError(error)) return error._tag; +} + +export function coordinatorErrorMessage(error: unknown): string { + if (isCoordinatorError(error) || isTurnError(error)) return error.message; + if ( + error instanceof Error && + "cause" in error && + error.cause instanceof Error + ) { + return error.cause.message; + } + if (error instanceof Error) return error.message; + return String(error); +} + +export async function persistThreadError( + threadId: string, + turnId: string, + message: string, + code?: string +): Promise { + const thread = await getThread(threadId); + if (thread.isErr() || !thread.value) return; + + const event: ChatChunk["event"] = { + type: "thread_error", + message, + ...(code ? { code } : {}), + }; + + const entry = await appendConversation(threadId, { + threadId, + turnId, + event, + }); + if (entry.isErr()) return; + + return entry.value.chunk.event; +} + +export async function persistCoordinatorThreadError( + threadId: string, + error: unknown, + turnId = randomId() +): Promise { + return await persistThreadError( + threadId, + turnId, + coordinatorErrorMessage(error), + coordinatorErrorCode(error) + ); +} diff --git a/apps/cli/src/utils/thread-title.ts b/apps/cli/src/utils/thread-title.ts new file mode 100644 index 0000000..fc7aa02 --- /dev/null +++ b/apps/cli/src/utils/thread-title.ts @@ -0,0 +1,32 @@ +import { getConversations } from "@cyrus/database/repositories/conversations"; +import { applyAutoThreadTitle } from "@cyrus/database/repositories/threads"; +import { fold } from "@cyrus/utils/fold"; + +export async function maybeApplyAutoThreadTitle( + threadId: string, + turnId: string +): Promise { + const conversations = await getConversations(threadId); + if (conversations.isErr()) return; + + const completedTurns = conversations.value.filter( + (entry) => entry.chunk.event.type === "turn_completed" + ); + if (completedTurns.length !== 1) return; + + const folded = fold(conversations.value); + if (folded.isErr()) return; + + const userMessage = folded.value.messages.find( + (message) => message.role === "user" && message.turnId === turnId + ); + const assistantMessage = folded.value.messages.find( + (message) => message.role === "assistant" && message.turnId === turnId + ); + + await applyAutoThreadTitle( + threadId, + userMessage?.content ?? "", + assistantMessage?.content + ); +} diff --git a/apps/mobile/app/(drawer)/(tabs)/index.tsx b/apps/mobile/app/(drawer)/(tabs)/index.tsx index ebe0209..2cc76d8 100644 --- a/apps/mobile/app/(drawer)/(tabs)/index.tsx +++ b/apps/mobile/app/(drawer)/(tabs)/index.tsx @@ -1,2 +1,9 @@ -// biome-ignore lint/performance/noBarrelFile: Expo Router route entry point -export { MobileChatApp as default } from "@/components/mobile-chat-app"; +import { Text, View } from "react-native"; + +export default function ThreadsTab() { + return ( + + Threads (stub) + + ); +} diff --git a/apps/web/src/components/chat/composer/composer-prompt-editor.tsx b/apps/web/src/components/chat/composer/composer-prompt-editor.tsx index 9bd904f..32fbe3f 100644 --- a/apps/web/src/components/chat/composer/composer-prompt-editor.tsx +++ b/apps/web/src/components/chat/composer/composer-prompt-editor.tsx @@ -124,9 +124,27 @@ function $deleteBackwardChars(count: number) { } } +function $setMessage(message: ChatMessage): void { + const root = $getRoot(); + root.clear(); + const paragraph = $createParagraphNode(); + for (const block of message) { + if (block.type === "text") { + if (block.text) paragraph.append($createTextNode(block.text)); + continue; + } + paragraph.append($createComposerResourceNode(block.uri, block.name)); + paragraph.append($createTextNode(" ")); + } + root.append(paragraph); + if (message.length > 0) paragraph.selectEnd(); +} + export type ComposerPromptEditorHandle = { focus: () => void; clear: () => void; + setPlainText: (text: string) => void; + setMessage: (message: ChatMessage) => void; getPlainText: () => string; getMessage: () => ChatMessage; hasContent: () => boolean; @@ -255,6 +273,16 @@ function EditorHandlePlugin({ root.append($createParagraphNode()); }); }, + setPlainText: (text) => { + editor.update(() => { + $setMessage(text ? [{ type: "text", text }] : []); + }); + }, + setMessage: (message) => { + editor.update(() => { + $setMessage(message); + }); + }, getPlainText: () => { let text = ""; editor.getEditorState().read(() => { @@ -356,6 +384,12 @@ const ComposerPromptEditorInner = forwardRef< useImperativeHandle(ref, () => ({ focus: () => handleRef.current?.focus(), clear: () => handleRef.current?.clear(), + setPlainText: (text) => { + handleRef.current?.setPlainText(text); + }, + setMessage: (message) => { + handleRef.current?.setMessage(message); + }, getPlainText: () => handleRef.current?.getPlainText() ?? "", getMessage: () => handleRef.current?.getMessage() ?? [], hasContent: () => handleRef.current?.hasContent() ?? false, diff --git a/apps/web/src/components/chat/composer/composer-skeleton.tsx b/apps/web/src/components/chat/composer/composer-skeleton.tsx index 320bb2c..7dd0851 100644 --- a/apps/web/src/components/chat/composer/composer-skeleton.tsx +++ b/apps/web/src/components/chat/composer/composer-skeleton.tsx @@ -1,17 +1,15 @@ import { Skeleton } from "@/components/ui/skeleton"; +/** + * Full composer rectangle — height mirrors the empty glass shell: + * pt-3.5 + editor min-h-17.5 + pb-2 + footer h-8 + pb-2.5 (= 8.375rem). + * sm: pt-4 + … + pb-3 (= 8.625rem). + */ export function ComposerSkeleton() { return ( -
-
-
- -
-
- - -
-
+
+ Loading composer +
); } diff --git a/apps/web/src/components/chat/composer/index.tsx b/apps/web/src/components/chat/composer/index.tsx index 5b3fb2c..7fb6e86 100644 --- a/apps/web/src/components/chat/composer/index.tsx +++ b/apps/web/src/components/chat/composer/index.tsx @@ -3,8 +3,13 @@ import { useGitStatus } from "@cyrus/hooks/connection/use-git"; import { useListAgents } from "@cyrus/hooks/connection/use-list-agents"; import { useProjects } from "@cyrus/hooks/connection/use-projects"; import { useSearchEntries } from "@cyrus/hooks/connection/use-search-entries"; +import { + useComposerDraft, + useComposerDraftStore, +} from "@cyrus/hooks/stores/composer-draft"; import type { ChatMessage } from "@cyrus/schemas/rtc/chat"; import type { Thread } from "@cyrus/schemas/rtc/threads"; +import type { ErrorView } from "@cyrus/schemas/view"; import { cn } from "cnfast"; import { type ClipboardEvent, @@ -69,6 +74,7 @@ export function Composer({ onStop, busy = false, stopping = false, + threadError = null, }: { projectId: string; threadId: string; @@ -77,6 +83,7 @@ export function Composer({ onStop?: () => void; busy?: boolean; stopping?: boolean; + threadError?: ErrorView | null; }) { const agentsQuery = useListAgents(); const agents = agentsQuery.data?.agents ?? []; @@ -95,9 +102,11 @@ export function Composer({ catalog.promptCapabilities.embeddedContext !== false; const canAttachFiles = Boolean(threadCwd) && supportsEmbeddedContext; const canPasteUrls = supportsEmbeddedContext; + const composerBlocked = Boolean(threadError ?? catalog.bindError); const gitStatus = useGitStatus(threadId); const isGitRepo = gitStatus.data?.isRepo === true; + const { setValue: setDraft, clear: clearDraft } = useComposerDraft(threadId); const [plainText, setPlainText] = useState(""); const [hasContent, setHasContent] = useState(false); const [sending, setSending] = useState(false); @@ -106,8 +115,40 @@ export function Composer({ const [mentionDismissed, setMentionDismissed] = useState(false); const [slashDismissed, setSlashDismissed] = useState(false); const editorRef = useRef(null); - const submitStateRef = useRef({ busy, stopping, sending, hasAgents }); - submitStateRef.current = { busy, stopping, sending, hasAgents }; + const restoredForThreadRef = useRef(null); + const submitStateRef = useRef({ + busy, + stopping, + sending, + hasAgents, + composerBlocked, + }); + submitStateRef.current = { + busy, + stopping, + sending, + hasAgents, + composerBlocked, + }; + + // restore persisted draft once per thread (do not depend on draftMessage — + // setMessage triggers onChange → setDraft and would loop) + useEffect(() => { + if (restoredForThreadRef.current === threadId) return; + restoredForThreadRef.current = threadId; + const draft = useComposerDraftStore.getState().draftsByThread[threadId]; + const frame = requestAnimationFrame(() => { + if (draft && draft.length > 0) { + editorRef.current?.setMessage(draft); + setHasContent(true); + } else { + editorRef.current?.clear(); + setPlainText(""); + setHasContent(false); + } + }); + return () => cancelAnimationFrame(frame); + }, [threadId]); const triggerText = textForTriggers(plainText); @@ -171,7 +212,14 @@ export function Composer({ const submit = useCallback(async () => { const state = submitStateRef.current; - if (state.stopping || state.sending || !state.hasAgents) return; + if ( + state.stopping || + state.sending || + !state.hasAgents || + state.composerBlocked + ) { + return; + } const message = editorRef.current?.getMessage() ?? []; if (message.length === 0) return; @@ -181,10 +229,11 @@ export function Composer({ editorRef.current?.clear(); setPlainText(""); setHasContent(false); + clearDraft(); } finally { setSending(false); } - }, [onSend]); + }, [clearDraft, onSend]); const handleMentionKeys = useCallback( (key: ComposerCommandKey): boolean => { @@ -262,6 +311,7 @@ export function Composer({ function handlePlainTextChange(next: string) { setPlainText(next); setHasContent(editorRef.current?.hasContent() ?? false); + setDraft(editorRef.current?.getMessage() ?? []); const trigger = textForTriggers(next); if (canPasteUrls && TRAILING_URL_PATTERN.test(trigger)) { @@ -360,7 +410,7 @@ export function Composer({ > +
+ +
+

+ {error.message} +

+

+ Try re-selecting the agent or starting a new message once the issue + is resolved. +

+
+
+
+ ); +} diff --git a/apps/web/src/components/chat/feed/feed-entry-view.tsx b/apps/web/src/components/chat/feed/feed-entry-view.tsx index 09e706d..3826b83 100644 --- a/apps/web/src/components/chat/feed/feed-entry-view.tsx +++ b/apps/web/src/components/chat/feed/feed-entry-view.tsx @@ -1,4 +1,5 @@ import type { FeedEntry } from "@cyrus/utils/conversations/thread-feed"; +import { ErrorRow } from "@/components/chat/feed/error-row"; import { AssistantMessage } from "@/components/chat/messages/assistant-message"; import { AssistantThinking } from "@/components/chat/messages/assistant-thinking"; import { UserMessage } from "@/components/chat/messages/user-message"; @@ -19,6 +20,8 @@ export function FeedEntryView({ entry }: { entry: FeedEntry }) { return ; case "diff": return ; + case "error": + return ; default: { const _exhaustive: never = entry; return _exhaustive; diff --git a/apps/web/src/components/chat/main/thread-workspace.tsx b/apps/web/src/components/chat/main/thread-workspace.tsx index b6e0e8e..3f1f095 100644 --- a/apps/web/src/components/chat/main/thread-workspace.tsx +++ b/apps/web/src/components/chat/main/thread-workspace.tsx @@ -47,6 +47,16 @@ export function ThreadWorkspace({ const lastTurn = conversation.turns.at(-1); const lastTurnStateRef = useRef(lastTurn?.state); + // Only orphan tip errors (e.g. bind failures) block send. Turn errors stay in + // the feed so a failed turn does not permanently prevent the next message. + const lastError = conversation.errors.at(-1) ?? null; + const lastMessageAt = conversation.messages.at(-1)?.createdAt; + const composerBlockingError = + lastError && + !conversation.turns.some((turn) => turn.id === lastError.turnId) && + (lastMessageAt == null || lastError.createdAt >= lastMessageAt) + ? lastError + : null; useEffect(() => { if (!(diffOpen && lastTurn)) return; @@ -97,6 +107,7 @@ export function ThreadWorkspace({ projectId={projectId} stopping={stopping} thread={thread} + threadError={composerBlockingError} threadId={thread.id} /> diff --git a/openspec/changes/external-agent-conversation-ux/.openspec.yaml b/openspec/changes/archive/2026-07-15-external-agent-conversation-ux/.openspec.yaml similarity index 100% rename from openspec/changes/external-agent-conversation-ux/.openspec.yaml rename to openspec/changes/archive/2026-07-15-external-agent-conversation-ux/.openspec.yaml diff --git a/openspec/changes/external-agent-conversation-ux/design.md b/openspec/changes/archive/2026-07-15-external-agent-conversation-ux/design.md similarity index 100% rename from openspec/changes/external-agent-conversation-ux/design.md rename to openspec/changes/archive/2026-07-15-external-agent-conversation-ux/design.md diff --git a/openspec/changes/external-agent-conversation-ux/proposal.md b/openspec/changes/archive/2026-07-15-external-agent-conversation-ux/proposal.md similarity index 100% rename from openspec/changes/external-agent-conversation-ux/proposal.md rename to openspec/changes/archive/2026-07-15-external-agent-conversation-ux/proposal.md diff --git a/openspec/changes/external-agent-conversation-ux/specs/chat-timeline-ui/spec.md b/openspec/changes/archive/2026-07-15-external-agent-conversation-ux/specs/chat-timeline-ui/spec.md similarity index 100% rename from openspec/changes/external-agent-conversation-ux/specs/chat-timeline-ui/spec.md rename to openspec/changes/archive/2026-07-15-external-agent-conversation-ux/specs/chat-timeline-ui/spec.md diff --git a/openspec/changes/external-agent-conversation-ux/specs/conversation-view/spec.md b/openspec/changes/archive/2026-07-15-external-agent-conversation-ux/specs/conversation-view/spec.md similarity index 100% rename from openspec/changes/external-agent-conversation-ux/specs/conversation-view/spec.md rename to openspec/changes/archive/2026-07-15-external-agent-conversation-ux/specs/conversation-view/spec.md diff --git a/openspec/changes/external-agent-conversation-ux/specs/thread-error-surfacing/spec.md b/openspec/changes/archive/2026-07-15-external-agent-conversation-ux/specs/thread-error-surfacing/spec.md similarity index 100% rename from openspec/changes/external-agent-conversation-ux/specs/thread-error-surfacing/spec.md rename to openspec/changes/archive/2026-07-15-external-agent-conversation-ux/specs/thread-error-surfacing/spec.md diff --git a/openspec/changes/external-agent-conversation-ux/specs/thread-title/spec.md b/openspec/changes/archive/2026-07-15-external-agent-conversation-ux/specs/thread-title/spec.md similarity index 100% rename from openspec/changes/external-agent-conversation-ux/specs/thread-title/spec.md rename to openspec/changes/archive/2026-07-15-external-agent-conversation-ux/specs/thread-title/spec.md diff --git a/openspec/changes/external-agent-conversation-ux/specs/wire-schemas/spec.md b/openspec/changes/archive/2026-07-15-external-agent-conversation-ux/specs/wire-schemas/spec.md similarity index 100% rename from openspec/changes/external-agent-conversation-ux/specs/wire-schemas/spec.md rename to openspec/changes/archive/2026-07-15-external-agent-conversation-ux/specs/wire-schemas/spec.md diff --git a/openspec/changes/archive/2026-07-15-external-agent-conversation-ux/tasks.md b/openspec/changes/archive/2026-07-15-external-agent-conversation-ux/tasks.md new file mode 100644 index 0000000..ed783d3 --- /dev/null +++ b/openspec/changes/archive/2026-07-15-external-agent-conversation-ux/tasks.md @@ -0,0 +1,38 @@ +## 1. Wire events and feed types + +- [x] 1.1 Add `ThreadErrorEventSchema` to `@cyrus/schemas/rtc/chat` and `AgentEvent` union +- [x] 1.2 Extend `fold()` and view schemas for error entries +- [x] 1.3 Add `error` variant to `FeedEntry` in `shared/utils/src/conversations/thread-feed.ts` + +## 2. Worker: errors and titles + +- [x] 2.1 Emit and persist `thread_error` on bind/resume/prompt failures +- [x] 2.2 Implement auto-title update after first `turn_completed` (default name only) +- [x] 2.3 Map ACP session title notifications to `threads.name` when not user-renamed +- [x] 2.4 Track title source (auto vs user) or infer from rename RPC + +## 3. Worker: catalog refresh + +- [x] 3.1 After `setModel`, refresh session config options before returning +- [x] 3.2 Reset invalid effort/persona selection server-side when options change + +## 4. Web UI (PR #49/#52 feed + composer) + +- [x] 4.1 Add `ErrorRow` (or inline error entry) in `feed-entry-view.tsx` +- [x] 4.2 Add composer warning/disabled state in `composer/index.tsx` on bind/resume/chat error (beyond agent-list load errors) +- [x] 4.3 Invalidate effort/persona queries after model change in `use-agent-catalog.ts` / `compact-composer-controls.tsx` +- [x] 4.4 Add `useComposerDraftStore` with zustand persist; replace composer `useState` draft +- [x] 4.5 Reflect auto-generated titles in `thread-header.tsx` breadcrumb (alongside git actions) + +## 5. Mobile UI + +- [x] 5.1 Port error feed rendering and composer draft persist + +## 6. Specs + +- [x] 6.1 Delta-update `chat-timeline-ui` spec for error feed entries + +## 7. Tests + +- [x] 7.1 Event mapping tests for `thread_error` and `deriveFeed` error entry +- [x] 7.2 Auto-title repository test (default name replaced, manual preserved) diff --git a/openspec/changes/external-agent-conversation-ux/tasks.md b/openspec/changes/external-agent-conversation-ux/tasks.md deleted file mode 100644 index 0e94579..0000000 --- a/openspec/changes/external-agent-conversation-ux/tasks.md +++ /dev/null @@ -1,38 +0,0 @@ -## 1. Wire events and feed types - -- [ ] 1.1 Add `ThreadErrorEventSchema` to `@cyrus/schemas/rtc/chat` and `AgentEvent` union -- [ ] 1.2 Extend `fold()` and view schemas for error entries -- [ ] 1.3 Add `error` variant to `FeedEntry` in `shared/utils/src/conversations/thread-feed.ts` - -## 2. Worker: errors and titles - -- [ ] 2.1 Emit and persist `thread_error` on bind/resume/prompt failures -- [ ] 2.2 Implement auto-title update after first `turn_completed` (default name only) -- [ ] 2.3 Map ACP session title notifications to `threads.name` when not user-renamed -- [ ] 2.4 Track title source (auto vs user) or infer from rename RPC - -## 3. Worker: catalog refresh - -- [ ] 3.1 After `setModel`, refresh session config options before returning -- [ ] 3.2 Reset invalid effort/persona selection server-side when options change - -## 4. Web UI (PR #49/#52 feed + composer) - -- [ ] 4.1 Add `ErrorRow` (or inline error entry) in `feed-entry-view.tsx` -- [ ] 4.2 Add composer warning/disabled state in `composer/index.tsx` on bind/resume/chat error (beyond agent-list load errors) -- [ ] 4.3 Invalidate effort/persona queries after model change in `use-agent-catalog.ts` / `compact-composer-controls.tsx` -- [ ] 4.4 Add `useComposerDraftStore` with zustand persist; replace composer `useState` draft -- [ ] 4.5 Reflect auto-generated titles in `thread-header.tsx` breadcrumb (alongside git actions) - -## 5. Mobile UI - -- [ ] 5.1 Port error feed rendering and composer draft persist - -## 6. Specs - -- [ ] 6.1 Delta-update `chat-timeline-ui` spec for error feed entries - -## 7. Tests - -- [ ] 7.1 Event mapping tests for `thread_error` and `deriveFeed` error entry -- [ ] 7.2 Auto-title repository test (default name replaced, manual preserved) diff --git a/openspec/specs/acp-draft-session/spec.md b/openspec/specs/acp-draft-session/spec.md index eb38353..dab769a 100644 --- a/openspec/specs/acp-draft-session/spec.md +++ b/openspec/specs/acp-draft-session/spec.md @@ -8,40 +8,46 @@ Thread-scoped ACP draft session lifecycle: bind agent on draft threads, switch a ### Requirement: Bind agent creates draft session -The worker SHALL provide a `bindAgent` controller operation that, for a given `threadId`, `projectId`, and `agentName`, ensures the agent subprocess is initialized, calls ACP `session/new` with the project's working directory, persists the returned `sessionId`, and returns a catalog snapshot (models, modes, efforts, personas, capabilities). +The worker SHALL provide a `bindAgent` controller operation that, for a given `threadId`, `projectId`, and `agentName`, ensures the agent subprocess is initialized, calls ACP `session/new` with the project's working directory, keeps the returned `sessionId` in memory only, and returns a catalog snapshot (models, modes, efforts, personas, capabilities). Draft binds SHALL NOT persist `agentName` or `sessionId` to Turso. #### Scenario: First bind on draft thread -- **WHEN** `bindAgent` is called for a thread with no `sessionId` and a valid project +- **WHEN** `bindAgent` is called for an unlocked thread and a valid project - **THEN** the worker creates an ACP session with `cwd` equal to the project path -- **AND** persists `agentName` and `sessionId` on the thread row +- **AND** stores the session in the worker's in-memory map only - **AND** returns catalog options from that session +- **AND** leaves `agentName` and `sessionId` null on the thread row #### Scenario: Re-bind same agent is idempotent -- **WHEN** `bindAgent` is called for a thread already bound to the same `agentName` with a valid in-memory or persisted session +- **WHEN** `bindAgent` is called for a thread already bound to the same `agentName` with a valid in-memory session - **THEN** the worker reuses the existing session without calling `session/new` again - **AND** returns the current catalog snapshot ### Requirement: Agent switch on draft thread -The worker SHALL allow changing `agentName` on a draft thread (no conversation entries and `agentLocked` false) by closing the previous ACP session and creating a new session for the newly selected agent. +The worker SHALL allow changing `agentName` on a draft thread (`agentLocked` false) by closing the previous ACP session and creating a new session for the newly selected agent, without writing either session to Turso. #### Scenario: Switch agent before first message - **WHEN** `bindAgent` is called with a different `agentName` on an unlocked draft thread -- **THEN** the worker calls `closeSession` on the previous `sessionId` if present -- **AND** creates a new session for the new agent -- **AND** updates persisted `agentName` and `sessionId` +- **THEN** the worker calls `closeSession` on the previous in-memory session if present +- **AND** creates a new session for the new agent in memory +- **AND** does not persist `agentName` or `sessionId` #### Scenario: Switch agent blocked after first turn - **WHEN** `bindAgent` is called with a different `agentName` on a thread where `agentLocked` is true - **THEN** the worker rejects the request with an error indicating the agent is locked -### Requirement: Agent lock after first turn +### Requirement: Persist and lock on first message -The worker SHALL set `agentLocked` to true when the first conversation entry (user message) is persisted for a thread. +The worker SHALL persist `agentName` and `sessionId` when the first chat turn begins, and SHALL set `agentLocked` to true when the first conversation entry (user message) is persisted for a thread. + +#### Scenario: Persist on first chat + +- **WHEN** `chat` is called for a draft thread with a live in-memory session +- **THEN** `agentName` and `sessionId` are written to the thread row before the prompt runs #### Scenario: Lock on first message @@ -51,22 +57,22 @@ The worker SHALL set `agentLocked` to true when the first conversation entry (us ### Requirement: Close session on thread delete -The worker SHALL call ACP `closeSession` for a thread's persisted `sessionId` when the thread is deleted, before removing the thread row. +The worker SHALL call ACP `closeSession` for a thread's in-memory or persisted session when the thread is deleted, before removing the thread row. #### Scenario: Delete bound thread -- **WHEN** `deleteThread` is called for a thread with a non-null `sessionId` +- **WHEN** `deleteThread` is called for a thread with a live or persisted session - **THEN** the worker attempts `closeSession` on that session - **AND** deletes the thread and its conversation entries from Turso #### Scenario: Delete unbound thread -- **WHEN** `deleteThread` is called for a thread with no `sessionId` +- **WHEN** `deleteThread` is called for a thread with no session - **THEN** the worker deletes the thread row without ACP calls ### Requirement: Thread-scoped catalog reads -Catalog operations (`getModels`, `getModes`, `getEfforts`, `getPersonas`) SHALL require a `threadId` and SHALL read catalog data from that thread's bound ACP session. The worker SHALL NOT use a shared probe session in a unrelated working directory. +Catalog operations (`getModels`, `getModes`, `getEfforts`, `getPersonas`) SHALL require a `threadId` and SHALL read catalog data from that thread's bound ACP session (in-memory draft or resumed committed session). The worker SHALL NOT use a shared probe session in a unrelated working directory. #### Scenario: Catalog requires a bound session diff --git a/openspec/specs/acp-session-router/spec.md b/openspec/specs/acp-session-router/spec.md index b5ec722..67311c8 100644 --- a/openspec/specs/acp-session-router/spec.md +++ b/openspec/specs/acp-session-router/spec.md @@ -2,27 +2,29 @@ ## Purpose -Map Cyrus threads to ACP sessions on shared agent subprocesses, route prompts, stream events, and recover sessions after respawn. +Map Cyrus threads to ACP sessions on shared agent subprocesses, route prompts, stream events, and recover committed sessions after respawn. ## Requirements ### Requirement: Thread to session mapping -The worker SHALL maintain a mapping from Cyrus thread IDs to ACP `sessionId` values per agent, persisted in the local `threads` table and hydrated into memory on use. Sessions SHALL be created at agent bind time, not on first prompt. +The worker SHALL maintain a mapping from Cyrus thread IDs to ACP `sessionId` values per agent in memory. Draft (unlocked) sessions SHALL remain memory-only until the first user message persists them to the local `threads` table. Sessions SHALL be created at agent bind time, not on first prompt. -#### Scenario: Bind creates session +#### Scenario: Bind creates draft session in memory - **WHEN** `bindAgent` succeeds for a draft thread -- **THEN** the worker stores the returned `sessionId` in Turso and in the in-memory map +- **THEN** the worker stores the returned `sessionId` in the in-memory map +- **AND** does not write `agentName` or `sessionId` to Turso -#### Scenario: First prompt reuses bound session +#### Scenario: First prompt persists and reuses bound session -- **WHEN** a prompt arrives for a thread with a persisted `sessionId` -- **THEN** the worker calls ACP `session/prompt` on that session without creating a new one +- **WHEN** a prompt arrives for a thread with a live in-memory session and no persisted `sessionId` +- **THEN** the worker persists `agentName` and `sessionId` to Turso before prompting +- **AND** calls ACP `session/prompt` on that session without creating a new one #### Scenario: Prompt without bind fails -- **WHEN** a prompt arrives for a thread with no persisted `sessionId` +- **WHEN** a prompt arrives for a thread with no in-memory session and no committed persisted `sessionId` - **THEN** the worker rejects the prompt with an error indicating the agent must be bound first ### Requirement: Multi-project sessions on one subprocess @@ -36,17 +38,22 @@ The worker SHALL support multiple sessions with different `cwd` values on the sa ### Requirement: Session recovery after respawn -The worker SHALL attempt to recover sessions after an agent subprocess respawn or worker restart, preferring `session/resume` over `session/load`, using the persisted `sessionId` and project `cwd` from the thread row. +The worker SHALL attempt to recover **committed** sessions (`agentLocked` true) after an agent subprocess respawn or worker restart, preferring `session/resume` over `session/load`, using the persisted `sessionId` and project `cwd` from the thread row. Draft sessions SHALL NOT be resumed from Turso; the client SHALL call `bindAgent` again to mint a fresh session. -#### Scenario: Resume on worker restart +#### Scenario: Resume on worker restart for committed thread -- **WHEN** the worker restarts and a prompt or catalog operation needs a thread with persisted `sessionId` +- **WHEN** the worker restarts and a prompt or catalog operation needs a locked thread with persisted `sessionId` - **THEN** the worker re-attaches via `session/resume` when the agent advertised resume capability +#### Scenario: Draft thread after worker restart + +- **WHEN** the worker restarts and a draft thread has no live session +- **THEN** catalog and chat require a new `bindAgent` (no resume of a disposable draft id) + #### Scenario: Load fallback - **WHEN** a subprocess respawns and the agent supports `loadSession` but not `resume` -- **THEN** the worker calls `session/load` for the persisted `sessionId` and waits for history replay +- **THEN** the worker calls `session/load` for the persisted `sessionId` of a locked thread and waits for history replay - **AND** Turso conversation entries remain the client transcript source of truth ### Requirement: ACP event mapping diff --git a/openspec/specs/chat-timeline-ui/spec.md b/openspec/specs/chat-timeline-ui/spec.md index 5277320..f6f37a3 100644 --- a/openspec/specs/chat-timeline-ui/spec.md +++ b/openspec/specs/chat-timeline-ui/spec.md @@ -87,3 +87,19 @@ User and assistant messages SHALL render using shadcn `Message` and `Bubble` com - **WHEN** an assistant message is rendered - **THEN** it uses `Message` with start alignment and `Bubble variant="ghost"` for full-width prose + +### Requirement: Error feed entry in flat timeline + +The web chat timeline SHALL support a flat `error` feed entry type rendered alongside messages, tools, and diffs. Error entries SHALL be produced by `deriveFeed` from persisted thread error events. + +#### Scenario: Error renders in feed + +- **WHEN** a thread contains a persisted `thread_error` event for a turn +- **THEN** `deriveFeed` emits a `FeedEntry` with `type: "error"` +- **AND** `FeedEntryView` renders an inline error card for that entry + +#### Scenario: Error does not use WorkLog bundling + +- **WHEN** an error occurs during a turn that also has tool activity +- **THEN** the error appears as its own flat feed entry +- **AND** is not nested inside a collapsible work-log group diff --git a/openspec/specs/conversation-persistence/spec.md b/openspec/specs/conversation-persistence/spec.md index c494b5b..adcc762 100644 --- a/openspec/specs/conversation-persistence/spec.md +++ b/openspec/specs/conversation-persistence/spec.md @@ -94,21 +94,32 @@ The `threads` table SHALL store optional `sessionId` (ACP session id) and `agent - **WHEN** a new thread is created - **THEN** `agentName`, `sessionId` are null and `agentLocked` is false -#### Scenario: Bind persists session fields +#### Scenario: Bind does not persist draft session fields -- **WHEN** `bindAgent` succeeds -- **THEN** `agentName` and `sessionId` are persisted on the thread row +- **WHEN** `bindAgent` succeeds for an unlocked draft thread +- **THEN** `agentName` and `sessionId` remain null on the thread row +- **AND** the live session exists only in the worker memory map -#### Scenario: Agent name persisted before first message +#### Scenario: First chat persists session fields -- **WHEN** the user selects an agent and `bindAgent` completes -- **THEN** `agentName` is available in `listThreads` without waiting for the first chat message +- **WHEN** the first `chat` call runs for a draft thread with a live session +- **THEN** `agentName` and `sessionId` are persisted on the thread row before the prompt + +#### Scenario: Agent name available after first message + +- **WHEN** the first user message has been accepted for a thread +- **THEN** `agentName` and `sessionId` are available in `listThreads` ### Requirement: Thread schema wire sync `ThreadSchema` SHALL expose optional `sessionId` and `agentLocked` fields consistent with the database columns. -#### Scenario: Thread list includes bind state +#### Scenario: Thread list includes bind state when committed -- **WHEN** `listThreads` returns a bound draft thread +- **WHEN** `listThreads` returns a thread after the first user message - **THEN** each thread includes `agentName`, `sessionId`, and `agentLocked` when set + +#### Scenario: Draft thread omits session until first message + +- **WHEN** `listThreads` returns an unlocked draft thread that was bound only in memory +- **THEN** `agentName` and `sessionId` are unset on the thread object diff --git a/openspec/specs/conversation-view/spec.md b/openspec/specs/conversation-view/spec.md index 19ab38e..d56d9da 100644 --- a/openspec/specs/conversation-view/spec.md +++ b/openspec/specs/conversation-view/spec.md @@ -6,7 +6,7 @@ Client-derived conversation view schemas and the platform-agnostic `fold()` pipe ### Requirement: View schemas for folded conversation data -The system SHALL define Zod view schemas in `@cyrus/schemas/view` for the client-derived conversation shape: `MessageViewSchema`, `ToolCallViewSchema`, `DiffViewSchema`, `TurnViewSchema`, and `ThreadConversationSchema`. TypeScript types SHALL be derived via `z.infer` with no parallel hand-rolled definitions. +The system SHALL define Zod view schemas in `@cyrus/schemas/view` for the client-derived conversation shape: `MessageViewSchema`, `ToolCallViewSchema`, `DiffViewSchema`, `TurnViewSchema`, and `ThreadConversationSchema`. TypeScript types SHALL be derived via `z.infer` with no parallel hand-rolled definitions. View schemas SHALL include representations for thread error entries and diff review state where applicable. #### Scenario: View schemas use wire and ACP field names @@ -14,6 +14,11 @@ The system SHALL define Zod view schemas in `@cyrus/schemas/view` for the client - **THEN** tool call fields use `toolCallId` and `title` (not `id` and `name`) - **AND** project paths use `cwd` from `ProjectSchema` (not a separate `path` alias) +#### Scenario: Error entries fold into conversation view + +- **WHEN** `fold()` processes a thread error event +- **THEN** the result includes a renderable error entry associated with the correct turn + ### Requirement: Platform-agnostic conversation folding The system SHALL provide a `fold()` function in `@cyrus/utils` that accepts `ConversationEntry[]` and returns a `ThreadConversation` object validated against `ThreadConversationSchema`. @@ -112,3 +117,27 @@ The system SHALL remove the following from shared client types: `branch`, `lates - **WHEN** merged entries are passed to `fold()` - **THEN** the existing `fold()` implementation in `@cyrus/utils` produces the view without modification to its logic + +### Requirement: Catalog refresh on model change + +When the user changes model for a bound thread, the client SHALL invalidate and refetch effort and persona catalog queries for that thread. The worker SHALL return updated config options from the bound session after `setModel` completes. + +#### Scenario: Effort options refresh + +- **WHEN** `setModel` succeeds for a thread +- **THEN** the client refetches `getEfforts` and `getPersonas` for that thread +- **AND** invalid prior selections reset to a valid default + +### Requirement: Draft composer client persistence + +The web and mobile clients SHALL persist the unsent structured composer `ChatMessage` per thread (text and resource/attachment blocks) using client-side storage (Zustand persist). Drafts SHALL survive navigation and app reload on the same device. + +#### Scenario: Draft restored on return + +- **WHEN** the user navigates away from a thread with an unsent structured composer message and returns +- **THEN** the composer is restored to that exact `ChatMessage` (including resource blocks) + +#### Scenario: Draft cleared on send + +- **WHEN** the user successfully sends a message +- **THEN** the persisted structured draft for that thread is cleared diff --git a/openspec/specs/thread-error-surfacing/spec.md b/openspec/specs/thread-error-surfacing/spec.md new file mode 100644 index 0000000..4571cd2 --- /dev/null +++ b/openspec/specs/thread-error-surfacing/spec.md @@ -0,0 +1,47 @@ +## Purpose + +Persist and display ACP bind/resume/prompt failures as thread conversation events, without a separate client health dashboard. + +## Requirements + +### Requirement: Persisted thread error events + +The worker SHALL emit and persist conversation events when ACP bind, resume, prompt, or subprocess operations fail for a thread. Clients SHALL render these errors in the thread feed. + +#### Scenario: Bind failure + +- **WHEN** `bindAgent` fails due to spawn or initialize error +- **THEN** the worker returns an error to the client +- **AND** if a partial thread state exists, an error event is persisted on the thread when applicable + +#### Scenario: Prompt failure mid-turn + +- **WHEN** a chat turn fails due to agent crash or ACP error +- **THEN** the worker persists a `thread_error` event (or normalized session error event) for that turn +- **AND** emits `turn_interrupted` + +#### Scenario: Resume failure on worker restart + +- **WHEN** session resume fails for a persisted `sessionId` +- **THEN** the worker persists an error event on the thread +- **AND** subsequent prompt attempts return a descriptive error until re-bind + +### Requirement: Inline error display + +The web and mobile clients SHALL display thread error events as inline feed entries with a human-readable message and optional retry guidance (e.g. re-select agent). + +#### Scenario: Error visible in feed + +- **WHEN** a persisted error event exists for a thread +- **THEN** `fold()` produces a view entry rendered in the conversation feed +- **AND** the composer shows a disabled or warning state when the thread cannot accept input + +### Requirement: No client health dashboard + +The clients SHALL NOT display per-agent connection health indicators. Unhealthy agents SHALL be excluded server-side from `listAgents`. + +#### Scenario: Missing agent in dropdown + +- **WHEN** an enabled agent fails health check +- **THEN** it does not appear in the agent selector +- **AND** no separate health UI is shown diff --git a/openspec/specs/thread-title/spec.md b/openspec/specs/thread-title/spec.md new file mode 100644 index 0000000..dcbf045 --- /dev/null +++ b/openspec/specs/thread-title/spec.md @@ -0,0 +1,49 @@ +## Purpose + +Thread display names from Turso: auto-title after first turn, optional agent-provided titles, and user rename precedence. Titles are never discovered via ACP `listSessions`. + +## Requirements + +### Requirement: Auto title after first turn + +The worker SHALL update a thread's `name` after the first completed turn when the name is still the default `"New thread"`, using a Cyrus-owned title generator derived from the first user message and/or first assistant response text (sanitized, max 50 characters). + +#### Scenario: Default title replaced + +- **WHEN** the first `turn_completed` event is persisted for a thread named `"New thread"` +- **THEN** the worker updates `threads.name` to a generated title +- **AND** `listThreads` reflects the new name + +#### Scenario: Custom title preserved + +- **WHEN** the user renamed the thread before the first turn completed +- **THEN** auto-title does not overwrite the user-set name + +#### Scenario: Manual rename wins + +- **WHEN** the user calls `renameThread` after an auto title was applied +- **THEN** subsequent auto-title logic does not overwrite the manual name + +### Requirement: Agent-provided title updates + +When an ACP session notification includes a session title update mapped by the worker, the worker MAY update `threads.name` only if the current name is still default or previously agent-generated (not user-renamed). + +#### Scenario: Agent pushes title + +- **WHEN** the bound session emits a title update and the thread name is `"New thread"` +- **THEN** the worker updates `threads.name` to the agent-provided title + +#### Scenario: User rename blocks agent title + +- **WHEN** the user manually renamed the thread +- **THEN** agent-provided title updates are ignored + +### Requirement: Not sourced from listSessions + +Thread listing and titles SHALL come from Turso `threads` rows. The worker SHALL NOT call ACP `listSessions` for thread discovery. + +#### Scenario: Thread list from database + +- **WHEN** the client calls `listThreads` +- **THEN** results come from Turso ordered by `updatedAt` +- **AND** no agent session list API is invoked diff --git a/openspec/specs/wire-schemas/spec.md b/openspec/specs/wire-schemas/spec.md index 8ba2757..c82427f 100644 --- a/openspec/specs/wire-schemas/spec.md +++ b/openspec/specs/wire-schemas/spec.md @@ -177,3 +177,12 @@ The controller SHALL provide optional `getContextUsage({ threadId })` returning - **WHEN** the agent session reports token usage metadata - **THEN** `getContextUsage` returns used and limit values + +### Requirement: Thread error wire event + +The system SHALL define a `thread_error` chat event schema in `@cyrus/schemas/rtc/chat` with a human-readable message and optional error code. The event SHALL be persistable as a `ConversationEntry`. + +#### Scenario: Schema exported + +- **WHEN** a consumer imports chat event types from `@cyrus/schemas/rtc/chat` +- **THEN** `ThreadErrorEventSchema` is available and included in the `AgentEvent` union diff --git a/shared/constants/src/operation-keys.ts b/shared/constants/src/operation-keys.ts index 12e6d10..e4280ba 100644 --- a/shared/constants/src/operation-keys.ts +++ b/shared/constants/src/operation-keys.ts @@ -22,14 +22,14 @@ export const RTC_OPERATION_KEYS = { ["controller", "get-conversations", threadId] as const, listAgents: ["controller", "list-agents"], bindAgent: ["controller", "bind-agent"], - getModels: (threadId: string) => - ["controller", "get-models", threadId] as const, - getModes: (threadId: string) => - ["controller", "get-modes", threadId] as const, - getEfforts: (threadId: string) => - ["controller", "get-efforts", threadId] as const, - getPersona: (threadId: string) => - ["controller", "get-persona", threadId] as const, + getModels: (threadId: string, agentName = "") => + ["controller", "get-models", threadId, agentName] as const, + getModes: (threadId: string, agentName = "") => + ["controller", "get-modes", threadId, agentName] as const, + getEfforts: (threadId: string, agentName = "") => + ["controller", "get-efforts", threadId, agentName] as const, + getPersona: (threadId: string, agentName = "") => + ["controller", "get-persona", threadId, agentName] as const, getContextUsage: (threadId: string) => ["controller", "get-context-usage", threadId] as const, setModel: ["controller", "set-model"], diff --git a/shared/database/__tests__/integration/repositories.test.ts b/shared/database/__tests__/integration/repositories.test.ts index 2666726..0410d3d 100644 --- a/shared/database/__tests__/integration/repositories.test.ts +++ b/shared/database/__tests__/integration/repositories.test.ts @@ -8,8 +8,10 @@ import { listProjects, } from "@cyrus/database/repositories/projects"; import { + applyAutoThreadTitle, createThread, ensureThread, + getThread, listThreads, renameThread, threadNameFromPrompt, @@ -37,7 +39,7 @@ describe("database repositories", () => { }); }); - test("creates threads and derives names from first message", async () => { + test("creates threads without renaming from first message before turn completion", async () => { await withTempDatabase(async () => { const project = await createProject("Repo"); expect(project.isOk()).toBe(true); @@ -50,11 +52,57 @@ describe("database repositories", () => { expect(thread.isOk()).toBe(true); if (!thread.isOk()) return; - expect(thread.value.name).toBe("Fix the failing tests"); + expect(thread.value.name).toBe("New thread"); expect(threadNameFromPrompt(" hello ")).toBe("hello"); }); }); + test("auto-title replaces default name and preserves manual renames", async () => { + await withTempDatabase(async () => { + const project = await createProject("Repo"); + expect(project.isOk()).toBe(true); + if (!project.isOk()) return; + + const thread = await createThread(project.value.id); + expect(thread.isOk()).toBe(true); + if (!thread.isOk()) return; + + const renamed = await renameThread(thread.value.id, "Manual title"); + expect(renamed.isOk()).toBe(true); + if (!renamed.isOk()) return; + + const preserved = await applyAutoThreadTitle( + thread.value.id, + "First user message", + "Assistant summary sentence." + ); + expect(preserved.isOk()).toBe(true); + if (!preserved.isOk()) return; + expect(preserved.value).toBeUndefined(); + + const refreshed = await getThread(thread.value.id); + expect(refreshed.isOk()).toBe(true); + if (!refreshed.isOk()) return; + expect(refreshed.value?.name).toBe("Manual title"); + + const defaultThread = await createThread(project.value.id); + expect(defaultThread.isOk()).toBe(true); + if (!defaultThread.isOk()) return; + + const autoTitled = await applyAutoThreadTitle( + defaultThread.value.id, + "Fix the failing tests in auth module", + "Updated the auth tests and they pass now." + ); + expect(autoTitled.isOk()).toBe(true); + if (!autoTitled.isOk()) return; + expect(autoTitled.value?.name).toBe( + "Fix the failing tests in auth module" + ); + expect(autoTitled.value?.titleSource).toBe("auto"); + }); + }); + test("renames threads and persists conversation entries", async () => { await withTempDatabase(async () => { const project = await createProject("Repo"); diff --git a/shared/database/src/models/threads.ts b/shared/database/src/models/threads.ts index 654cdfe..9339067 100644 --- a/shared/database/src/models/threads.ts +++ b/shared/database/src/models/threads.ts @@ -7,6 +7,7 @@ export const threads = sqliteTable("threads", { .notNull() .references(() => projects.id, { onDelete: "cascade" }), name: text("name").notNull(), + titleSource: text("title_source"), agentName: text("agent_name"), sessionId: text("session_id"), agentLocked: integer("agent_locked").notNull().default(0), diff --git a/shared/database/src/repositories/threads.ts b/shared/database/src/repositories/threads.ts index e33c7bb..aec30e5 100644 --- a/shared/database/src/repositories/threads.ts +++ b/shared/database/src/repositories/threads.ts @@ -1,20 +1,37 @@ import type { RepositoryError } from "@cyrus/errors/repository"; import { notFound } from "@cyrus/errors/repository"; -import type { Thread } from "@cyrus/schemas/rtc/threads"; +import type { Thread, TitleSource } from "@cyrus/schemas/rtc/threads"; import { ThreadSchema } from "@cyrus/schemas/rtc/threads"; import { randomId } from "@cyrus/utils/identity"; import { nowISO } from "@cyrus/utils/time"; import { Result } from "better-result"; -import { and, desc, eq } from "drizzle-orm"; +import { and, desc, eq, isNull, ne, or } from "drizzle-orm"; import { connection } from "../connection"; import { threads } from "../models/threads"; import { repoArgs } from "../utils/repo"; import { getProject } from "./projects"; +export const DEFAULT_THREAD_NAME = "New thread"; + export function threadNameFromPrompt(message: string): string { const trimmed = message.trim(); - if (!trimmed) return "New thread"; - return trimmed.slice(0, 50); + if (!trimmed) return DEFAULT_THREAD_NAME; + return Array.from(trimmed).slice(0, 50).join(""); +} + +export function generateThreadTitle( + userMessage: string, + _assistantMessage?: string +): string { + return threadNameFromPrompt(userMessage); +} + +function canApplyAutoTitle(thread: Thread): boolean { + return thread.titleSource !== "user" && thread.name === DEFAULT_THREAD_NAME; +} + +function canApplyAgentTitle(thread: Thread): boolean { + return thread.titleSource !== "user"; } type ThreadCreateOptions = { @@ -41,23 +58,17 @@ const upsertThread = repoArgs( .limit(1); if (existing) { - const name = - options?.firstMessage && existing.name === "New thread" - ? threadNameFromPrompt(options.firstMessage) - : existing.name; const updatedAt = nowISO(); const agentName = options?.agentName ?? existing.agentName ?? undefined; await connection.db .update(threads) .set({ - name, agentName: agentName ?? null, updatedAt, }) .where(and(eq(threads.id, id), eq(threads.projectId, projectId))); return parseThreadRow({ ...existing, - name, agentName: agentName ?? null, updatedAt, }); @@ -67,10 +78,9 @@ const upsertThread = repoArgs( const thread = ThreadSchema.parse({ id, projectId, - name: options?.firstMessage - ? threadNameFromPrompt(options.firstMessage) - : (options?.branch ?? "New thread"), + name: options?.branch ?? DEFAULT_THREAD_NAME, agentName: options?.agentName, + titleSource: null, branch: options?.branch ?? null, worktreePath: options?.worktreePath ?? null, createdAt, @@ -80,6 +90,7 @@ const upsertThread = repoArgs( id: thread.id, projectId: thread.projectId, name: thread.name, + titleSource: null, agentName: thread.agentName ?? null, sessionId: null, agentLocked: 0, @@ -155,13 +166,37 @@ export const getThreadSession = repoArgs(async (threadId: string) => { }); const writeThreadName = repoArgs( - async (threadId: string, name: string, current: Thread) => { + async ( + threadId: string, + name: string, + current: Thread, + titleSource: TitleSource, + options?: { preserveUserTitle?: boolean } + ) => { const updatedAt = nowISO(); - await connection.db + const conditions = [eq(threads.id, threadId)]; + if (options?.preserveUserTitle) { + const titleGuard = or( + isNull(threads.titleSource), + ne(threads.titleSource, "user") + ); + if (titleGuard) conditions.push(titleGuard); + } + + const updated = await connection.db .update(threads) - .set({ name, updatedAt }) - .where(eq(threads.id, threadId)); - return ThreadSchema.parse({ ...current, name, updatedAt }); + .set({ name, titleSource, updatedAt }) + .where(and(...conditions)) + .returning(); + + if (updated.length === 0) { + const latest = await getThread(threadId); + if (latest.isErr()) throw latest.error; + if (!latest.value) throw notFound("thread", threadId); + return latest.value; + } + + return ThreadSchema.parse({ ...current, name, titleSource, updatedAt }); } ); @@ -173,7 +208,44 @@ export async function renameThread( if (thread.isErr()) return Result.err(thread.error); if (!thread.value) return Result.err(notFound("thread", threadId)); - return writeThreadName(threadId, name, thread.value); + return writeThreadName(threadId, name, thread.value, "user"); +} + +export async function applyAutoThreadTitle( + threadId: string, + userMessage: string, + assistantMessage?: string +): Promise> { + const thread = await getThread(threadId); + if (thread.isErr()) return Result.err(thread.error); + if (!thread.value) return Result.err(notFound("thread", threadId)); + if (!canApplyAutoTitle(thread.value)) return Result.ok(undefined); + + const title = generateThreadTitle(userMessage, assistantMessage); + return writeThreadName(threadId, title, thread.value, "auto", { + preserveUserTitle: true, + }); +} + +export async function applyAgentThreadTitle( + threadId: string, + title: string +): Promise> { + const trimmed = title.trim(); + if (!trimmed) return Result.ok(undefined); + + const thread = await getThread(threadId); + if (thread.isErr()) return Result.err(thread.error); + if (!thread.value) return Result.err(notFound("thread", threadId)); + if (!canApplyAgentTitle(thread.value)) return Result.ok(undefined); + + return writeThreadName( + threadId, + Array.from(trimmed).slice(0, 50).join(""), + thread.value, + "agent", + { preserveUserTitle: true } + ); } const writeThreadWorktreePath = repoArgs( @@ -239,6 +311,60 @@ export async function bindThreadAgent( return writeThreadAgent(threadId, projectId, thread.value, data); } +const clearThreadAgentBinding = repoArgs( + async (threadId: string, projectId: string, current: Thread) => { + if (!(current.agentName || current.sessionId)) { + return current; + } + const updatedAt = nowISO(); + const updated = await connection.db + .update(threads) + .set({ + agentName: null, + sessionId: null, + updatedAt, + }) + .where( + and( + eq(threads.id, threadId), + eq(threads.projectId, projectId), + eq(threads.agentLocked, 0) + ) + ) + .returning(); + + if (updated.length === 0) { + const latest = await getThread(threadId); + if (latest.isErr()) throw latest.error; + if (!latest.value) throw notFound("thread", threadId); + return latest.value; + } + + return ThreadSchema.parse({ + ...current, + agentName: null, + sessionId: null, + updatedAt, + }); + } +); + +/** Clears leftover draft agent/session fields (unlocked threads only). */ +export async function clearThreadDraftBinding( + threadId: string, + projectId: string +): Promise> { + const thread = await getThread(threadId); + if (thread.isErr()) return Result.err(thread.error); + if (!thread.value) return Result.err(notFound("thread", threadId)); + if (thread.value.projectId !== projectId) { + return Result.err(notFound("thread", threadId)); + } + if (thread.value.agentLocked) return Result.ok(thread.value); + + return clearThreadAgentBinding(threadId, projectId, thread.value); +} + const lockThreadAgent = repoArgs(async (threadId: string, current: Thread) => { const updatedAt = nowISO(); await connection.db diff --git a/shared/errors/src/common.ts b/shared/errors/src/common.ts index a3b79fd..2c2d8d6 100644 --- a/shared/errors/src/common.ts +++ b/shared/errors/src/common.ts @@ -4,6 +4,7 @@ export const errorModules = { repository: "repository", coordinator: "coordinator", git: "git", + turn: "turn", } as const; export type ErrorModule = (typeof errorModules)[keyof typeof errorModules]; diff --git a/shared/errors/src/turn.ts b/shared/errors/src/turn.ts new file mode 100644 index 0000000..1372c29 --- /dev/null +++ b/shared/errors/src/turn.ts @@ -0,0 +1,65 @@ +import { TaggedError } from "better-result"; +import { errorModules, errorTag, isModuleError } from "./common"; + +const tags = { + emitFailed: errorTag(errorModules.turn, "emit_failed"), + streamFailed: errorTag(errorModules.turn, "stream_failed"), +} as const; + +export class TurnEmitFailedError extends TaggedError(tags.emitFailed)<{ + message: string; + detail?: string; +}>() { + get orpcCode() { + return "INTERNAL_SERVER_ERROR" as const; + } +} + +export class TurnStreamFailedError extends TaggedError(tags.streamFailed)<{ + message: string; + detail?: string; +}>() { + get orpcCode() { + return "INTERNAL_SERVER_ERROR" as const; + } +} + +export type TurnError = TurnEmitFailedError | TurnStreamFailedError; + +export function isTurnError(cause: unknown): cause is TurnError { + return isModuleError(cause, errorModules.turn); +} + +export function turnEmitFailed( + message: string, + detail?: string +): TurnEmitFailedError { + return new TurnEmitFailedError({ message, detail }); +} + +export function turnStreamFailed( + message: string, + detail?: string +): TurnStreamFailedError { + return new TurnStreamFailedError({ message, detail }); +} + +export function turnErrorMessageFromUnknown(error: unknown): string { + if ( + error instanceof Error && + "cause" in error && + error.cause instanceof Error + ) { + return error.cause.message; + } + if (error instanceof Error) return error.message; + if ( + typeof error === "object" && + error !== null && + "message" in error && + typeof error.message === "string" + ) { + return error.message; + } + return String(error); +} diff --git a/shared/hooks/src/connection/use-agent-catalog.ts b/shared/hooks/src/connection/use-agent-catalog.ts index 8b3bfb9..e7afc3b 100644 --- a/shared/hooks/src/connection/use-agent-catalog.ts +++ b/shared/hooks/src/connection/use-agent-catalog.ts @@ -1,12 +1,7 @@ import { RTC_OPERATION_KEYS } from "@cyrus/constants/operation-keys"; import type { AvailableCommand } from "@cyrus/schemas/rtc/catalog"; import type { ListThreadsOutput } from "@cyrus/schemas/rtc/threads"; -import { - keepPreviousData, - useMutation, - useQuery, - useQueryClient, -} from "@tanstack/react-query"; +import { useMutation, useQuery, useQueryClient } from "@tanstack/react-query"; import { useEffect } from "react"; import { useRtc } from "../contexts/rtc"; import { @@ -18,6 +13,8 @@ const EMPTY_COMMANDS: AvailableCommand[] = []; type CatalogOption = { id: string; name: string }; +const EMPTY_OPTIONS: CatalogOption[] = []; + function pickExplicitOption( id: string | undefined, options: CatalogOption[] @@ -63,6 +60,9 @@ export function useAgentCatalog({ const pendingAgent = useAgentCatalogStore( (state) => state.pendingAgentByThread[threadId] ); + const liveBinding = useAgentCatalogStore( + (state) => state.liveBindingByThread[threadId] + ); const setModelSelection = useAgentCatalogStore((state) => state.setModel); const setModeSelection = useAgentCatalogStore((state) => state.setMode); const setEffortSelection = useAgentCatalogStore((state) => state.setEffort); @@ -80,15 +80,16 @@ export function useAgentCatalog({ const clearPendingAgent = useAgentCatalogStore( (state) => state.clearPendingAgent ); + const setLiveBinding = useAgentCatalogStore((state) => state.setLiveBinding); + const clearLiveBinding = useAgentCatalogStore( + (state) => state.clearLiveBinding + ); const markResumeBindRequested = useAgentCatalogStore( (state) => state.markResumeBindRequested ); const clearResumeBindRequested = useAgentCatalogStore( (state) => state.clearResumeBindRequested ); - const resumeBindRequested = useAgentCatalogStore( - (state) => state.resumeBindRequestedByThread[threadId] - ); const threadsQueryKey = RTC_OPERATION_KEYS.listThreads(projectId); const threadsQuery = useQuery({ @@ -102,21 +103,20 @@ export function useAgentCatalog({ (item) => item.id === threadId ); const agentLocked = Boolean(thread?.agentLocked); - const boundSessionId = thread?.sessionId; + const persistedSessionId = agentLocked ? thread?.sessionId : undefined; + const preferredAgent = + pendingAgent ?? + liveBinding?.agentName ?? + (agentLocked ? thread?.agentName : undefined); - const boundAgent = pickExplicitOption( - pendingAgent ?? thread?.agentName, - agents - ); - const displayAgent = pickDisplayOption( - pendingAgent ?? thread?.agentName, - agents - ); + const boundAgent = pickExplicitOption(preferredAgent, agents); + const displayAgent = pickDisplayOption(preferredAgent, agents); + const catalogAgent = preferredAgent ?? displayAgent; - const modelsQueryKey = RTC_OPERATION_KEYS.getModels(threadId); - const modesQueryKey = RTC_OPERATION_KEYS.getModes(threadId); - const effortsQueryKey = RTC_OPERATION_KEYS.getEfforts(threadId); - const personaQueryKey = RTC_OPERATION_KEYS.getPersona(threadId); + const modelsQueryKey = RTC_OPERATION_KEYS.getModels(threadId, catalogAgent); + const modesQueryKey = RTC_OPERATION_KEYS.getModes(threadId, catalogAgent); + const effortsQueryKey = RTC_OPERATION_KEYS.getEfforts(threadId, catalogAgent); + const personaQueryKey = RTC_OPERATION_KEYS.getPersona(threadId, catalogAgent); const contextUsageQueryKey = RTC_OPERATION_KEYS.getContextUsage(threadId); const bindAgentMutation = useMutation({ @@ -124,76 +124,89 @@ export function useAgentCatalog({ mutationKey: RTC_OPERATION_KEYS.bindAgent, }), onMutate: async (variables) => { + const nextModelsKey = RTC_OPERATION_KEYS.getModels( + threadId, + variables.agentName + ); + const nextModesKey = RTC_OPERATION_KEYS.getModes( + threadId, + variables.agentName + ); + const nextEffortsKey = RTC_OPERATION_KEYS.getEfforts( + threadId, + variables.agentName + ); + const nextPersonaKey = RTC_OPERATION_KEYS.getPersona( + threadId, + variables.agentName + ); + setPendingAgent(threadId, variables.agentName); + const currentLive = + useAgentCatalogStore.getState().liveBindingByThread[threadId]; + if (currentLive && currentLive.agentName !== variables.agentName) { + clearLiveBinding(threadId); + } await queryClient.cancelQueries({ queryKey: threadsQueryKey }); const previousThreads = queryClient.getQueryData(threadsQueryKey); - const previousAgent = previousThreads?.threads.find( - (item) => item.id === threadId - )?.agentName; - if (previousThreads) { - queryClient.setQueryData(threadsQueryKey, { - ...previousThreads, - threads: previousThreads.threads.map((item) => - item.id === threadId - ? { ...item, agentName: variables.agentName } - : item - ), - }); - } - const previousModels = queryClient.getQueryData(modelsQueryKey); - const previousModes = queryClient.getQueryData(modesQueryKey); - const previousEfforts = queryClient.getQueryData(effortsQueryKey); - const previousPersonas = queryClient.getQueryData(personaQueryKey); + const previousAgent = + currentLive?.agentName ?? + previousThreads?.threads.find((item) => item.id === threadId) + ?.agentName; const previousCapabilities = capabilities; const previousCommands = commands; const previousUsage = contextUsage; if (previousAgent && previousAgent !== variables.agentName) { - queryClient.setQueryData(modelsQueryKey, { models: [] }); - queryClient.setQueryData(modesQueryKey, { modes: [] }); - queryClient.setQueryData(effortsQueryKey, { efforts: [] }); - queryClient.setQueryData(personaQueryKey, { personas: [] }); setCapabilities(threadId, {}); setCommands(threadId, []); setContextUsage(threadId, null); } return { previousThreads, - previousModels, - previousModes, - previousEfforts, - previousPersonas, previousCapabilities, previousCommands, previousUsage, + nextModelsKey, + nextModesKey, + nextEffortsKey, + nextPersonaKey, }; }, - onSuccess: (data) => { - queryClient.setQueryData(modelsQueryKey, { models: data.models }); - queryClient.setQueryData(modesQueryKey, { modes: data.modes }); - queryClient.setQueryData(effortsQueryKey, { efforts: data.efforts }); - queryClient.setQueryData(personaQueryKey, { personas: data.personas }); + onSuccess: (data, variables) => { + const agentName = variables.agentName; + setLiveBinding(threadId, { + agentName, + sessionId: data.sessionId, + }); + queryClient.setQueryData( + RTC_OPERATION_KEYS.getModels(threadId, agentName), + { models: data.models } + ); + queryClient.setQueryData( + RTC_OPERATION_KEYS.getModes(threadId, agentName), + { modes: data.modes } + ); + queryClient.setQueryData( + RTC_OPERATION_KEYS.getEfforts(threadId, agentName), + { efforts: data.efforts } + ); + queryClient.setQueryData( + RTC_OPERATION_KEYS.getPersona(threadId, agentName), + { personas: data.personas } + ); setCapabilities(threadId, data.capabilities); setCommands(threadId, data.commands ?? []); - queryClient.invalidateQueries({ queryKey: threadsQueryKey }); + if (data.agentLocked) { + queryClient.invalidateQueries({ queryKey: threadsQueryKey }); + } queryClient.invalidateQueries({ queryKey: contextUsageQueryKey }); }, onError: (_error, _variables, context) => { + clearLiveBinding(threadId); if (context?.previousThreads) { queryClient.setQueryData(threadsQueryKey, context.previousThreads); } - if (context?.previousModels) { - queryClient.setQueryData(modelsQueryKey, context.previousModels); - } - if (context?.previousModes) { - queryClient.setQueryData(modesQueryKey, context.previousModes); - } - if (context?.previousEfforts) { - queryClient.setQueryData(effortsQueryKey, context.previousEfforts); - } - if (context?.previousPersonas) { - queryClient.setQueryData(personaQueryKey, context.previousPersonas); - } if (context?.previousCapabilities) { setCapabilities(threadId, context.previousCapabilities); } @@ -212,7 +225,11 @@ export function useAgentCatalog({ const { mutate: bindAgent, isPending: bindAgentPending } = bindAgentMutation; - const catalogEnabled = Boolean(boundSessionId) && !bindAgentPending; + const hasLiveOrPersistedSession = Boolean( + liveBinding?.sessionId || persistedSessionId + ); + const catalogEnabled = + Boolean(hasLiveOrPersistedSession && catalogAgent) && !bindAgentPending; const modelsQuery = useQuery({ ...orpcController.getModels.queryOptions({ @@ -221,9 +238,10 @@ export function useAgentCatalog({ }), queryKey: modelsQueryKey, enabled: catalogEnabled, - placeholderData: keepPreviousData, }); - const models = modelsQuery.data?.models ?? []; + const models = catalogAgent + ? (modelsQuery.data?.models ?? EMPTY_OPTIONS) + : EMPTY_OPTIONS; const modesQuery = useQuery({ ...orpcController.getModes.queryOptions({ @@ -232,9 +250,10 @@ export function useAgentCatalog({ }), queryKey: modesQueryKey, enabled: catalogEnabled, - placeholderData: keepPreviousData, }); - const modes = modesQuery.data?.modes ?? []; + const modes = catalogAgent + ? (modesQuery.data?.modes ?? EMPTY_OPTIONS) + : EMPTY_OPTIONS; const effortsQuery = useQuery({ ...orpcController.getEfforts.queryOptions({ @@ -243,9 +262,10 @@ export function useAgentCatalog({ }), queryKey: effortsQueryKey, enabled: catalogEnabled, - placeholderData: keepPreviousData, }); - const efforts = effortsQuery.data?.efforts ?? []; + const efforts = catalogAgent + ? (effortsQuery.data?.efforts ?? EMPTY_OPTIONS) + : EMPTY_OPTIONS; const personaQuery = useQuery({ ...orpcController.getPersona.queryOptions({ @@ -254,9 +274,10 @@ export function useAgentCatalog({ }), queryKey: personaQueryKey, enabled: catalogEnabled, - placeholderData: keepPreviousData, }); - const personas = personaQuery.data?.personas ?? []; + const personas = catalogAgent + ? (personaQuery.data?.personas ?? EMPTY_OPTIONS) + : EMPTY_OPTIONS; const contextUsageQuery = useQuery({ ...orpcController.getContextUsage.queryOptions({ @@ -287,6 +308,11 @@ export function useAgentCatalog({ ...orpcController.setModel.mutationOptions({ mutationKey: RTC_OPERATION_KEYS.setModel, }), + onSuccess: () => { + queryClient.invalidateQueries({ queryKey: effortsQueryKey }); + queryClient.invalidateQueries({ queryKey: personaQueryKey }); + queryClient.invalidateQueries({ queryKey: modelsQueryKey }); + }, }); const setModeMutation = useMutation({ ...orpcController.setMode.mutationOptions({ @@ -305,41 +331,83 @@ export function useAgentCatalog({ }); useEffect(() => { - if (!(thread?.agentName && boundSessionId) || agentLocked) return; - if (bindAgentPending || pendingAgent || resumeBindRequested) return; + bindAgentMutation.reset(); + // Reset scoped to thread switches only — mutation identity changes each render. + // eslint-disable-next-line react-hooks/exhaustive-deps -- threadId is the intentional dependency + }, [threadId]); + + useEffect(() => { + const store = useAgentCatalogStore.getState(); + if ( + bindAgentPending || + store.pendingAgentByThread[threadId] || + store.resumeBindRequestedByThread[threadId] + ) { + return; + } if (bindAgentMutation.isError) return; - const cachedModels = queryClient.getQueryData<{ models: unknown[] }>( - modelsQueryKey - ); - if (cachedModels?.models?.length) return; + const currentLive = store.liveBindingByThread[threadId]; + if (currentLive) return; + + // Committed threads: rehydrate the persisted session after worker restart. + if (agentLocked && thread?.agentName && persistedSessionId) { + const cachedModels = queryClient.getQueryData<{ models: unknown[] }>( + modelsQueryKey + ); + if (cachedModels?.models?.length) return; + + markResumeBindRequested(threadId); + bindAgent({ + threadId, + projectId, + agentName: thread.agentName, + }); + return; + } + + // Drafts: bind the first agent; session stays worker-local until first message. + if (agentLocked) return; + const defaultAgent = agents[0]?.id; + if (!defaultAgent) return; markResumeBindRequested(threadId); bindAgent({ threadId, projectId, - agentName: thread.agentName, + agentName: defaultAgent, }); }, [ agentLocked, + agents, bindAgent, bindAgentMutation.isError, bindAgentPending, - boundSessionId, markResumeBindRequested, modelsQueryKey, - pendingAgent, + persistedSessionId, projectId, queryClient, - resumeBindRequested, thread?.agentName, threadId, ]); + useEffect(() => { + if (!catalogAgent || models.length === 0) return; + const currentModelId = + useAgentCatalogStore.getState().selectionByThread[threadId]?.modelId; + if (currentModelId && models.some((model) => model.id === currentModelId)) { + return; + } + const firstModel = models[0]; + if (!firstModel) return; + setModelSelection(threadId, firstModel.id); + }, [catalogAgent, models, setModelSelection, threadId]); + const modelsLoading = + Boolean(catalogAgent) && models.length === 0 && - Boolean(boundAgent || displayAgent) && - (bindAgentPending || (catalogEnabled && modelsQuery.isLoading)); + (bindAgentPending || (catalogEnabled && modelsQuery.isFetching)); function selectAgent(agentName: string) { if (agentLocked || (boundAgent && agentName === boundAgent)) return; @@ -396,6 +464,7 @@ export function useAgentCatalog({ return { agentLocked, + bindError: bindAgentMutation.error, capabilities, commands, contextUsage, diff --git a/shared/hooks/src/connection/use-thread-conversation.ts b/shared/hooks/src/connection/use-thread-conversation.ts index 565f981..7ddc0ab 100644 --- a/shared/hooks/src/connection/use-thread-conversation.ts +++ b/shared/hooks/src/connection/use-thread-conversation.ts @@ -14,6 +14,7 @@ import { useRtc } from "../contexts/rtc"; const EMPTY: ThreadConversation = { diffs: [], + errors: [], messages: [], thoughts: [], toolCalls: [], diff --git a/shared/hooks/src/connection/use-worker-conversation-sync.ts b/shared/hooks/src/connection/use-worker-conversation-sync.ts index 2f3c380..2970a50 100644 --- a/shared/hooks/src/connection/use-worker-conversation-sync.ts +++ b/shared/hooks/src/connection/use-worker-conversation-sync.ts @@ -60,6 +60,14 @@ export function useWorkerConversationSync(): void { if (!isTerminalChunk(chunk)) return; settleTurnWaiter(chunk.threadId, chunk.turnId, chunk.event); + if (chunk.event.type === "turn_completed") { + queryClient.invalidateQueries({ + predicate: (query) => + Array.isArray(query.queryKey) && + query.queryKey[0] === "controller" && + query.queryKey[1] === "list-threads", + }); + } // Persisted events already arrived via subscribe; prune handled in // applyChunkToCache. Mark stale without refetching — an immediate refetch // swaps local entry IDs for server IDs and remounts the feed. diff --git a/shared/hooks/src/stores/agent-catalog.ts b/shared/hooks/src/stores/agent-catalog.ts index 3666676..4981c70 100644 --- a/shared/hooks/src/stores/agent-catalog.ts +++ b/shared/hooks/src/stores/agent-catalog.ts @@ -11,12 +11,18 @@ type ThreadCatalogSelection = { personaId?: string; }; +type LiveThreadBinding = { + agentName: string; + sessionId: string; +}; + type AgentCatalogState = { selectionByThread: Record; capabilitiesByThread: Record>; commandsByThread: Record; contextUsageByThread: Record; pendingAgentByThread: Record; + liveBindingByThread: Record; resumeBindRequestedByThread: Record; setModel: (threadId: string, modelId: string) => void; setMode: (threadId: string, modeId: string) => void; @@ -30,6 +36,8 @@ type AgentCatalogState = { setContextUsage: (threadId: string, usage: ContextUsage | null) => void; setPendingAgent: (threadId: string, agentName: string) => void; clearPendingAgent: (threadId: string) => void; + setLiveBinding: (threadId: string, binding: LiveThreadBinding) => void; + clearLiveBinding: (threadId: string) => void; markResumeBindRequested: (threadId: string) => void; clearResumeBindRequested: (threadId: string) => void; }; @@ -53,15 +61,36 @@ export const useAgentCatalogStore = create((set) => ({ commandsByThread: {}, contextUsageByThread: {}, pendingAgentByThread: {}, + liveBindingByThread: {}, resumeBindRequestedByThread: {}, setModel: (threadId, modelId) => - set((state) => patchSelection(state, threadId, { modelId })), + set((state) => { + if (state.selectionByThread[threadId]?.modelId === modelId) { + return state; + } + return patchSelection(state, threadId, { modelId }); + }), setMode: (threadId, modeId) => - set((state) => patchSelection(state, threadId, { modeId })), + set((state) => { + if (state.selectionByThread[threadId]?.modeId === modeId) { + return state; + } + return patchSelection(state, threadId, { modeId }); + }), setEffort: (threadId, effortId) => - set((state) => patchSelection(state, threadId, { effortId })), + set((state) => { + if (state.selectionByThread[threadId]?.effortId === effortId) { + return state; + } + return patchSelection(state, threadId, { effortId }); + }), setPersona: (threadId, personaId) => - set((state) => patchSelection(state, threadId, { personaId })), + set((state) => { + if (state.selectionByThread[threadId]?.personaId === personaId) { + return state; + } + return patchSelection(state, threadId, { personaId }); + }), setCapabilities: (threadId, capabilities) => set((state) => ({ capabilitiesByThread: { @@ -109,6 +138,29 @@ export const useAgentCatalogStore = create((set) => ({ state.pendingAgentByThread; return { pendingAgentByThread }; }), + setLiveBinding: (threadId, binding) => + set((state) => { + const current = state.liveBindingByThread[threadId]; + if ( + current?.agentName === binding.agentName && + current.sessionId === binding.sessionId + ) { + return state; + } + return { + liveBindingByThread: { + ...state.liveBindingByThread, + [threadId]: binding, + }, + }; + }), + clearLiveBinding: (threadId) => + set((state) => { + if (!(threadId in state.liveBindingByThread)) return state; + const { [threadId]: _removed, ...liveBindingByThread } = + state.liveBindingByThread; + return { liveBindingByThread }; + }), markResumeBindRequested: (threadId) => set((state) => ({ resumeBindRequestedByThread: { diff --git a/shared/hooks/src/stores/composer-draft.ts b/shared/hooks/src/stores/composer-draft.ts new file mode 100644 index 0000000..fc29a3a --- /dev/null +++ b/shared/hooks/src/stores/composer-draft.ts @@ -0,0 +1,102 @@ +import type { ChatMessage } from "@cyrus/schemas/rtc/chat"; +import { create } from "zustand"; +import { persist } from "zustand/middleware"; + +const EMPTY_DRAFT: ChatMessage = []; + +type ComposerDraftState = { + draftsByThread: Record; + setDraft: (threadId: string, message: ChatMessage) => void; + clearDraft: (threadId: string) => void; +}; + +function sameDraft(left: ChatMessage | undefined, right: ChatMessage): boolean { + if (!left) return right.length === 0; + if (left.length !== right.length) return false; + return left.every((block, index) => { + const other = right[index]; + if (!other || block.type !== other.type) return false; + if (block.type === "text" && other.type === "text") { + return block.text === other.text; + } + if (block.type === "resource" && other.type === "resource") { + return block.uri === other.uri && block.name === other.name; + } + return false; + }); +} + +export const useComposerDraftStore = create()( + persist( + (set) => ({ + draftsByThread: {}, + setDraft: (threadId, message) => + set((state) => { + if (sameDraft(state.draftsByThread[threadId], message)) { + return state; + } + if (message.length === 0) { + if (!(threadId in state.draftsByThread)) return state; + const { [threadId]: _removed, ...draftsByThread } = + state.draftsByThread; + return { draftsByThread }; + } + return { + draftsByThread: { + ...state.draftsByThread, + [threadId]: message, + }, + }; + }), + clearDraft: (threadId) => + set((state) => { + if (!(threadId in state.draftsByThread)) return state; + const { [threadId]: _removed, ...draftsByThread } = + state.draftsByThread; + return { draftsByThread }; + }), + }), + { + name: "cyrus-composer-drafts", + partialize: (state) => ({ draftsByThread: state.draftsByThread }), + version: 1, + migrate: (persisted) => { + const state = persisted as { + draftsByThread?: Record; + }; + const draftsByThread: Record = {}; + for (const [threadId, value] of Object.entries( + state.draftsByThread ?? {} + )) { + if (typeof value === "string") { + const text = value.trim(); + if (text) draftsByThread[threadId] = [{ type: "text", text }]; + continue; + } + if (Array.isArray(value) && value.length > 0) { + draftsByThread[threadId] = value as ChatMessage; + } + } + return { draftsByThread }; + }, + } + ) +); + +export function useComposerDraft(threadId: string): { + value: ChatMessage; + setValue: (message: ChatMessage) => void; + clear: () => void; +} { + const value = useComposerDraftStore( + (state) => state.draftsByThread[threadId] ?? EMPTY_DRAFT + ); + const setDraft = useComposerDraftStore((state) => state.setDraft); + const clearDraft = useComposerDraftStore((state) => state.clearDraft); + + return { + value, + setValue: (message) => setDraft(threadId, message), + clear: () => clearDraft(threadId), + }; +} diff --git a/shared/schemas/src/rtc/chat.test.ts b/shared/schemas/src/rtc/chat.test.ts index bbc7174..e0edc1c 100644 --- a/shared/schemas/src/rtc/chat.test.ts +++ b/shared/schemas/src/rtc/chat.test.ts @@ -99,6 +99,17 @@ describe("chat schemas", () => { }, }) ).toMatchObject({ type: "plan_update" }); + + expect( + AgentEventSchema.parse({ + type: "thread_error", + message: "Session resume failed", + code: "coordinator.runtime", + }) + ).toMatchObject({ + type: "thread_error", + message: "Session resume failed", + }); }); test("parses chat chunks with nested events", () => { diff --git a/shared/schemas/src/rtc/chat.ts b/shared/schemas/src/rtc/chat.ts index 0928c96..5ee0353 100644 --- a/shared/schemas/src/rtc/chat.ts +++ b/shared/schemas/src/rtc/chat.ts @@ -247,11 +247,18 @@ export const TurnInterruptedEventSchema = z.object({ type: z.literal("turn_interrupted"), }); +export const ThreadErrorEventSchema = z.object({ + type: z.literal("thread_error"), + message: z.string(), + code: z.string().optional(), +}); + export const AgentEventSchema = z.discriminatedUnion("type", [ ThreadStartedEventSchema, UserMessageEventSchema, TurnCompletedEventSchema, TurnInterruptedEventSchema, + ThreadErrorEventSchema, TokenEventSchema, ThoughtEventSchema, MessageCompletedEventSchema, diff --git a/shared/schemas/src/rtc/threads.test.ts b/shared/schemas/src/rtc/threads.test.ts index 1b7d00a..9a86480 100644 --- a/shared/schemas/src/rtc/threads.test.ts +++ b/shared/schemas/src/rtc/threads.test.ts @@ -22,6 +22,7 @@ describe("thread schemas", () => { projectId: "project-1", name: "Main", agentName: undefined, + titleSource: null, createdAt: "2026-07-11T00:00:00.000Z", updatedAt: "2026-07-11T00:00:00.000Z", }); @@ -73,6 +74,7 @@ describe("thread schemas", () => { agentName: "claude-acp", sessionId: "session-1", agentLocked: true, + titleSource: null, createdAt: "2026-07-11T00:00:00.000Z", updatedAt: "2026-07-11T00:00:00.000Z", }); diff --git a/shared/schemas/src/rtc/threads.ts b/shared/schemas/src/rtc/threads.ts index 9b77d46..db0e2c0 100644 --- a/shared/schemas/src/rtc/threads.ts +++ b/shared/schemas/src/rtc/threads.ts @@ -7,6 +7,8 @@ const nullableString = z .nullish() .transform((value) => value ?? null); +export const TitleSourceSchema = z.enum(["auto", "agent", "user"]); + export const ThreadSchema = z.object({ id: z.string(), projectId: z.string(), @@ -14,6 +16,7 @@ export const ThreadSchema = z.object({ agentName: optionalString, sessionId: optionalString, agentLocked: optionalBoolean, + titleSource: TitleSourceSchema.nullish().transform((value) => value ?? null), branch: nullableString.optional(), worktreePath: nullableString.optional(), createdAt: z.string(), @@ -72,6 +75,7 @@ export const GetConversationsOutputSchema = z.object({ conversations: z.array(ConversationEntrySchema), }); +export type TitleSource = z.infer; export type Thread = z.infer; export type ConversationEntry = z.infer; export type ProjectQueryInput = z.infer; diff --git a/shared/schemas/src/view/index.ts b/shared/schemas/src/view/index.ts index 3debf17..3c07b41 100644 --- a/shared/schemas/src/view/index.ts +++ b/shared/schemas/src/view/index.ts @@ -48,11 +48,20 @@ export const TurnViewSchema = z.object({ completedAt: z.string().nullable(), }); +export const ErrorViewSchema = z.object({ + id: z.string(), + message: z.string(), + code: z.string().optional(), + createdAt: z.string(), + turnId: z.string(), +}); + export const ThreadConversationSchema = z.object({ messages: z.array(MessageViewSchema), thoughts: z.array(ThoughtViewSchema), toolCalls: z.array(ToolCallViewSchema), diffs: z.array(DiffViewSchema), + errors: z.array(ErrorViewSchema), turns: z.array(TurnViewSchema), }); @@ -61,4 +70,5 @@ export type ThoughtView = z.infer; export type ToolCallView = z.infer; export type DiffView = z.infer; export type TurnView = z.infer; +export type ErrorView = z.infer; export type ThreadConversation = z.infer; diff --git a/shared/utils/src/conversations/thread-feed.test.ts b/shared/utils/src/conversations/thread-feed.test.ts new file mode 100644 index 0000000..dc59db9 --- /dev/null +++ b/shared/utils/src/conversations/thread-feed.test.ts @@ -0,0 +1,87 @@ +import { describe, expect, test } from "bun:test"; +import { deriveFeed } from "./thread-feed"; + +describe("deriveFeed", () => { + test("emits flat error feed entries from thread errors", () => { + const feed = deriveFeed({ + diffs: [], + errors: [ + { + code: "coordinator.runtime", + createdAt: "2026-07-11T00:00:02.000Z", + id: "error-1", + message: "Bind failed", + turnId: "turn-1", + }, + ], + messages: [ + { + content: "Hello", + createdAt: "2026-07-11T00:00:01.000Z", + id: "user-turn-1", + role: "user", + turnId: "turn-1", + }, + ], + thoughts: [], + toolCalls: [], + turns: [ + { + completedAt: "2026-07-11T00:00:03.000Z", + id: "turn-1", + index: 0, + state: "interrupted", + threadId: "thread-1", + }, + ], + }); + + expect(feed.some((entry) => entry.type === "error")).toBe(true); + expect(feed.find((entry) => entry.type === "error")).toMatchObject({ + type: "error", + error: { + message: "Bind failed", + turnId: "turn-1", + }, + }); + }); + + test("interleaves orphaned errors by createdAt", () => { + const feed = deriveFeed({ + diffs: [], + errors: [ + { + createdAt: "2026-07-11T00:00:01.500Z", + id: "error-orphan", + message: "Early bind failed", + turnId: "orphan-turn", + }, + ], + messages: [ + { + content: "Later", + createdAt: "2026-07-11T00:00:02.000Z", + id: "user-turn-2", + role: "user", + turnId: "turn-2", + }, + ], + thoughts: [], + toolCalls: [], + turns: [ + { + completedAt: "2026-07-11T00:00:03.000Z", + id: "turn-2", + index: 0, + state: "complete", + threadId: "thread-1", + }, + ], + }); + + expect(feed.map((entry) => entry.id)).toEqual([ + "error-orphan", + "user-turn-2", + ]); + }); +}); diff --git a/shared/utils/src/conversations/thread-feed.ts b/shared/utils/src/conversations/thread-feed.ts index 9543524..ff14ba1 100644 --- a/shared/utils/src/conversations/thread-feed.ts +++ b/shared/utils/src/conversations/thread-feed.ts @@ -1,5 +1,6 @@ import type { DiffView, + ErrorView, MessageView, ThoughtView, ThreadConversation, @@ -32,11 +33,18 @@ export type DiffFeedEntry = FeedEntryBase & { turnId: string; }; +export type ErrorFeedEntry = FeedEntryBase & { + type: "error"; + error: ErrorView; + turnId: string; +}; + export type FeedEntry = | MessageFeedEntry | ThoughtFeedEntry | ToolFeedEntry - | DiffFeedEntry; + | DiffFeedEntry + | ErrorFeedEntry; type TimelineItem = { createdAt: string; @@ -54,7 +62,8 @@ function buildTurnTimeline( messages: MessageView[], thoughts: ThoughtView[], toolCalls: ToolCallView[], - diffs: DiffView[] + diffs: DiffView[], + errors: ErrorView[] ): FeedEntry[] { const timeline: TimelineItem[] = []; @@ -117,6 +126,20 @@ function buildTurnTimeline( }); } + for (const error of errors) { + if (error.turnId !== turnId) continue; + timeline.push({ + createdAt: error.createdAt, + kind: 5, + entry: { + type: "error", + id: error.id, + error, + turnId, + }, + }); + } + timeline.sort((left, right) => { const leftIsUser = left.kind === 0; const rightIsUser = right.kind === 0; @@ -146,7 +169,8 @@ export function deriveFeed( conversation.messages, conversation.thoughts, conversation.toolCalls, - conversation.diffs + conversation.diffs, + conversation.errors ) ); } @@ -156,9 +180,54 @@ export function deriveFeed( entries.push({ type: "message", id: message.id, message }); } + const orphanedErrors = conversation.errors.filter( + (error) => !knownTurnIds.has(error.turnId) + ); + for (const error of orphanedErrors) { + insertFeedEntryByCreatedAt(entries, { + type: "error", + id: error.id, + error, + turnId: error.turnId, + }); + } + return entries; } +function feedEntryCreatedAt(entry: FeedEntry): string | null { + switch (entry.type) { + case "message": + return entry.message.createdAt; + case "thought": + return entry.thought.createdAt; + case "tool": + return entry.tool.createdAt; + case "diff": + return null; + case "error": + return entry.error.createdAt; + default: { + const _exhaustive: never = entry; + return _exhaustive; + } + } +} + +function insertFeedEntryByCreatedAt( + entries: FeedEntry[], + entry: ErrorFeedEntry +): void { + const createdAt = entry.error.createdAt; + let index = 0; + for (const existing of entries) { + const existingAt = feedEntryCreatedAt(existing); + if (existingAt !== null && existingAt > createdAt) break; + index += 1; + } + entries.splice(index, 0, entry); +} + export function getRunningTurn( conversation: ThreadConversation | null ): ThreadConversation["turns"][number] | null { diff --git a/shared/utils/src/fold.test.ts b/shared/utils/src/fold.test.ts index ca3678d..6187aef 100644 --- a/shared/utils/src/fold.test.ts +++ b/shared/utils/src/fold.test.ts @@ -153,4 +153,27 @@ describe("fold", () => { expect(conversation.turns[0]?.state).toBe("interrupted"); }); + + test("folds thread error events into error views", () => { + const conversation = folded([ + entry(1, "turn-1", { type: "user_message", content: "Hello" }), + entry(2, "turn-1", { + type: "thread_error", + message: "Agent crashed", + code: "coordinator.runtime", + }), + entry(3, "turn-1", { type: "turn_interrupted" }), + ]); + + expect(conversation.errors).toEqual([ + { + code: "coordinator.runtime", + createdAt: "2026-07-11T00:00:02.000Z", + id: "error-entry-2", + message: "Agent crashed", + turnId: "turn-1", + }, + ]); + expect(conversation.turns[0]?.state).toBe("interrupted"); + }); }); diff --git a/shared/utils/src/fold.ts b/shared/utils/src/fold.ts index 8b915f1..c46db29 100644 --- a/shared/utils/src/fold.ts +++ b/shared/utils/src/fold.ts @@ -2,6 +2,7 @@ import type { AgentEvent, ToolCallContent } from "@cyrus/schemas/rtc/chat"; import type { ConversationEntry } from "@cyrus/schemas/rtc/threads"; import { type DiffView, + type ErrorView, type MessageView, type ThoughtView, type ThreadConversation, @@ -17,6 +18,7 @@ type MutableState = { thoughts: Map; toolCalls: Map; diffs: Map; + errors: Map; }; function touchTurn( @@ -230,6 +232,22 @@ function applyMessageCompleted( }); } +function applyThreadError( + state: MutableState, + entry: ConversationEntry, + event: Extract, + turnId: string +): void { + const key = `error-${entry.id}`; + state.errors.set(key, { + code: event.code, + createdAt: entry.createdAt, + id: key, + message: event.message, + turnId, + }); +} + function inferTurnState( turnEntries: ConversationEntry[], isLatest: boolean @@ -239,6 +257,9 @@ function inferTurnState( if (events.some((event) => event.type === "turn_interrupted")) { return "interrupted"; } + if (events.some((event) => event.type === "thread_error")) { + return "interrupted"; + } if (events.some((event) => event.type === "turn_completed")) { return "complete"; } @@ -336,6 +357,9 @@ function applyEvent( case "tool_call_update": applyToolCallUpdate(state, entry, event, turnId); return; + case "thread_error": + applyThreadError(state, entry, event, turnId); + return; default: return; } @@ -346,6 +370,7 @@ export function fold( ): Result { const state: MutableState = { diffs: new Map(), + errors: new Map(), messages: new Map(), thoughts: new Map(), toolCalls: new Map(), @@ -386,6 +411,12 @@ export function fold( const parsed = ThreadConversationSchema.safeParse({ diffs: [...state.diffs.values()], + errors: [...state.errors.values()].sort((left, right) => { + const leftTurn = turnOrder.get(left.turnId) ?? Number.MAX_SAFE_INTEGER; + const rightTurn = turnOrder.get(right.turnId) ?? Number.MAX_SAFE_INTEGER; + if (leftTurn !== rightTurn) return leftTurn - rightTurn; + return left.createdAt.localeCompare(right.createdAt); + }), thoughts: [...state.thoughts.values()] .filter((thought) => thought.content.trim().length > 0) .sort((left, right) => { diff --git a/tests/e2e/manual/verify-draft-session.ts b/tests/e2e/manual/verify-draft-session.ts index 8fadb15..52247ff 100644 --- a/tests/e2e/manual/verify-draft-session.ts +++ b/tests/e2e/manual/verify-draft-session.ts @@ -83,7 +83,7 @@ try { }); const thread = await client.createThread({ projectId: project.project.id }); - console.log("3. bindAgent"); + console.log("3. bindAgent (memory-only)"); const bound = await client.bindAgent({ threadId: thread.thread.id, projectId: project.project.id, @@ -91,12 +91,19 @@ try { }); console.log(" sessionId:", bound.sessionId); console.log(" models:", bound.models.length); + const afterBind = await client.listThreads({ + projectId: project.project.id, + }); + const draftRow = afterBind.threads.find((t) => t.id === thread.thread.id); + if (draftRow?.sessionId) { + throw new Error("bindAgent should not persist sessionId on draft threads"); + } console.log("4. getModels"); const models = await client.getModels({ threadId: thread.thread.id }); console.log(" models:", models.models.length); - console.log("5. chat (locks agent)"); + console.log("5. chat (persists + locks agent)"); await client.chat({ threadId: thread.thread.id, projectId: project.project.id, @@ -108,9 +115,10 @@ try { for (let i = 0; i < 40; i++) { const listed = await client.listThreads({ projectId: project.project.id }); const row = listed.threads.find((t) => t.id === thread.thread.id); - if (row?.agentLocked) { + if (row?.agentLocked && row.sessionId) { locked = true; console.log(" agentLocked: true"); + console.log(" sessionId persisted:", row.sessionId); break; } await Bun.sleep(500);