diff --git a/apps/desktop/packages/devo-ai-sdk/package.json b/apps/desktop/packages/devo-ai-sdk/package.json index 1d7372e8..64adb248 100644 --- a/apps/desktop/packages/devo-ai-sdk/package.json +++ b/apps/desktop/packages/devo-ai-sdk/package.json @@ -6,6 +6,7 @@ "type": "module", "exports": { "./v2/client": "./src/v2/client.ts", + "./v2/protocol-validation": "./src/v2/protocol-validation.ts", "./v2/native-client-support": "./src/v2/native-client-support.ts" }, "dependencies": { diff --git a/apps/desktop/packages/devo-ai-sdk/src/v2/client-native-interactions.test.ts b/apps/desktop/packages/devo-ai-sdk/src/v2/client-native-interactions.test.ts index 72c9a357..f17844a6 100644 --- a/apps/desktop/packages/devo-ai-sdk/src/v2/client-native-interactions.test.ts +++ b/apps/desktop/packages/devo-ai-sdk/src/v2/client-native-interactions.test.ts @@ -14,6 +14,7 @@ class FakeNativeTransport implements DevoNativeTransport { pendingControlRequests: unknown[] = [] subscriptionCursors: Array<{ streamId: string; seq: number }> = [] sessionItems: unknown[] = [] + resumeSession?: unknown async request(method: string, params?: unknown, directory?: string): Promise { this.requests.push({ method, params, directory }) @@ -47,9 +48,11 @@ class FakeNativeTransport implements DevoNativeTransport { return { views: [nativeWorkspaceView] } case "turn/start": return { turn: nativeTurnInProgress } + case "session/message/edit": + return editedMessageResult case "session/resume": return { - session: nativeSession, + session: this.resumeSession ?? nativeSession, lastContextOccupancy: nativeOccupancy, } case "session/items/list": @@ -155,6 +158,26 @@ const nativeTurnCompleted = { completedAt: "2026-08-24T00:00:08Z", } +const editedMessageResult = { + editState: "accepted", + replacementTurnId: "turn-2", + item: { + id: "item-user-edited", + sessionId: nativeSession.id, + turnId: "turn-2", + revision: 2, + seq: 1, + state: "completed", + createdAt: "2026-08-24T00:00:10.000Z", + updatedAt: "2026-08-24T00:00:10.000Z", + item: { + type: "userMessage", + content: [{ type: "text", text: "edited" }], + entry: "turnStart", + }, + }, +} + const nativeOccupancy = { totalTokens: 100_000, contextWindowTokens: 200_000, @@ -1197,4 +1220,122 @@ describe("Native desktop SDK interactions", () => { expect(after?.time.lastActivity).toBe(Date.parse("2026-08-24T00:00:08Z")) expect(after?.time.updated).toBe(Date.parse("2026-08-24T00:00:08Z")) }) + + test("loading session history emits the resume-enriched snapshot as session.updated", async () => { + const transport = new FakeNativeTransport() + transport.resumeSession = { + ...nativeSession, + model: { provider: "test", model: "alt-model" }, + settings: { ...nativeSession.settings, reasoningEffort: "enabled", mode: "plan" }, + } + const client = createDevoClient({ directory: "/repo", transport }) + const stream = (await client.global.event()).stream[Symbol.asyncIterator]() + await client.session.messages({ sessionID: nativeSession.id }) + + // The cold session/list snapshot carries the base model; resume is + // authoritative for persisted per-session selections, so its enriched + // snapshot must reach renderer session stores (they re-seed the + // composer from session.updated) instead of staying client-internal. + const update = await nextPayloadOfType(stream, "session.updated") + expect(update.properties.session.model?.model).toBe("alt-model") + expect(update.properties.session.settings?.reasoningEffort).toBe("enabled") + expect(update.properties.session.settings?.mode).toBe("plan") + }) + + test("session/message/edit sends canonical params", async () => { + const transport = new FakeNativeTransport() + const client = createDevoClient({ directory: "/repo", transport }) + await client.session.create() + await client.session.editMessage({ + sessionID: nativeSession.id, + itemID: "item-user-1", + text: "edited", + }) + const request = transport.requests.find((entry) => entry.method === "session/message/edit") + const params = (request?.params ?? {}) as { + sessionId?: string + itemId?: string + expectedRevision?: number + content?: unknown + idempotencyKey?: string + } + expect({ + method: request?.method, + sessionId: params.sessionId, + itemId: params.itemId, + expectedRevision: params.expectedRevision, + content: params.content, + hasIdempotencyKey: typeof params.idempotencyKey === "string" && params.idempotencyKey.length > 0, + }).toEqual({ + method: "session/message/edit", + sessionId: nativeSession.id, + itemId: "item-user-1", + expectedRevision: 0, + content: [{ type: "text", text: "edited" }], + hasIdempotencyKey: true, + }) + }) + + test("turn/superseded removes messages from the replaced turn", async () => { + const transport = new FakeNativeTransport() + transport.sessionItems = [ + { + id: "item-user-1", + sessionId: nativeSession.id, + turnId: "turn-1", + seq: 1, + revision: 1, + createdAt: "2026-08-24T00:00:00.000Z", + updatedAt: "2026-08-24T00:00:00.000Z", + state: "completed", + item: { + type: "userMessage", + content: [{ type: "text", text: "hello" }], + entry: "turnStart", + }, + }, + { + id: "item-assistant-1", + sessionId: nativeSession.id, + turnId: "turn-1", + seq: 2, + revision: 1, + createdAt: "2026-08-24T00:00:02.000Z", + updatedAt: "2026-08-24T00:00:14.000Z", + state: "completed", + item: { + type: "assistantMessage", + text: "world", + }, + }, + ] + const client = createDevoClient({ directory: "/repo", transport }) + const loaded = await client.session.messages({ sessionID: nativeSession.id }) + expect(loaded.data.map((entry) => entry.info.id).sort()).toEqual([ + "item-assistant-1", + "item-user-1", + ]) + const stream = (await client.global.event()).stream[Symbol.asyncIterator]() + transport.emit({ + type: "notification", + method: "turn/superseded", + params: { + sessionId: nativeSession.id, + supersededTurnId: "turn-1", + replacementTurnId: "turn-2", + editId: "edit-1", + reason: "message_edit_previous", + }, + }) + expect(await nextPayloadOfType(stream, "message.removed")).toEqual({ + type: "message.removed", + properties: { sessionID: nativeSession.id, messageID: "item-user-1" }, + }) + expect(await nextPayloadOfType(stream, "message.removed")).toEqual({ + type: "message.removed", + properties: { sessionID: nativeSession.id, messageID: "item-assistant-1" }, + }) + const remaining = await client.session.messages({ sessionID: nativeSession.id }) + expect(remaining.data).toEqual([]) + }) }) diff --git a/apps/desktop/packages/devo-ai-sdk/src/v2/client-session-fork.test.ts b/apps/desktop/packages/devo-ai-sdk/src/v2/client-session-fork.test.ts new file mode 100644 index 00000000..6e1dcedb --- /dev/null +++ b/apps/desktop/packages/devo-ai-sdk/src/v2/client-session-fork.test.ts @@ -0,0 +1,111 @@ +import { describe, expect, test } from "bun:test" +import { createDevoClient, type DevoNativeTransport, type DevoNativeTransportEvent } from "./client" + +class FakeTransport implements DevoNativeTransport { + readonly requests: Array<{ method: string; params: unknown }> = [] + private listeners: Array<(event: DevoNativeTransportEvent) => void> = [] + + constructor(private readonly handler: (method: string, params: unknown) => unknown) {} + + async request(method: string, params?: unknown): Promise { + this.requests.push({ method, params }) + return this.handler(method, params) + } + + async respond(): Promise {} + + subscribe(listener: (event: DevoNativeTransportEvent) => void): () => void { + this.listeners.push(listener) + return () => { + this.listeners = this.listeners.filter((item) => item !== listener) + } + } + + connected(): boolean { + return true + } +} + +const forkedSession = { + id: "child-session", + version: 1, + cwd: "/repo", + title: "Forked", + parent: null, + forkFromId: "parent-session", + atTurnId: "turn-2", + createdAt: "2026-01-01T00:00:00.000Z", + lastActivityAt: "2026-01-01T00:00:00.000Z", + status: "idle", + flags: [], + archived: false, + ephemeral: false, + model: { provider: "test", model: "test-model" }, + settings: { permissionProfile: "default" }, + preview: "", + queuedCount: 0, + usage: { + total: { + inputTokens: 0, + outputTokens: 0, + cacheCreationInputTokens: 0, + cacheReadInputTokens: 0, + reasoningTokens: 0, + totalTokens: 0, + callCount: 0, + meteredCallCount: 0, + failedCallCount: 0, + cancelledCallCount: 0, + }, + byPurpose: [], + updatedAt: "2026-01-01T00:00:00.000Z", + }, +} + +describe("session.fork", () => { + test("sends canonical session/fork params and remembers the child", async () => { + const transport = new FakeTransport((method, params) => { + if (method === "initialize") { + return { + protocolVersion: 1, + agentCapabilities: {}, + authMethods: [], + } + } + if (method === "session/fork") { + expect(params).toEqual({ + sessionId: "parent-session", + atTurnId: "turn-2", + cut: "before", + }) + return { session: forkedSession } + } + if (method === "session/resume") { + return { session: forkedSession } + } + if (method === "session/items/list") { + return { data: [], nextCursor: null } + } + if (method === "subscription/create") { + return { subscriptionId: "sub-1", cursors: [] } + } + throw new Error(`unexpected method ${method}`) + }) + + const client = createDevoClient({ + directory: "/repo", + transport, + }) + + const result = await client.session.fork({ + sessionID: "parent-session", + atTurnId: "turn-2", + cut: "before", + }) + + expect(result.data.id).toBe("child-session") + expect(result.data.forkFromId).toBe("parent-session") + expect(result.data.parentID).toBeUndefined() + expect(transport.requests.some((request) => request.method === "session/fork")).toBe(true) + }) +}) diff --git a/apps/desktop/packages/devo-ai-sdk/src/v2/client-subscription-replay.test.ts b/apps/desktop/packages/devo-ai-sdk/src/v2/client-subscription-replay.test.ts new file mode 100644 index 00000000..d361d045 --- /dev/null +++ b/apps/desktop/packages/devo-ai-sdk/src/v2/client-subscription-replay.test.ts @@ -0,0 +1,263 @@ +import { describe, expect, test } from "bun:test" +import { createDevoClient, type DevoNativeTransport, type DevoNativeTransportEvent } from "./client" + +class FakeTransport implements DevoNativeTransport { + readonly requests: Array<{ method: string; params: unknown }> = [] + + constructor(private readonly handler: (method: string, params: unknown) => unknown) {} + + async request(method: string, params?: unknown): Promise { + this.requests.push({ method, params }) + return this.handler(method, params) + } + + async respond(): Promise {} + + subscribe(listener: (event: DevoNativeTransportEvent) => void): () => void { + void listener + return () => {} + } + + connected(): boolean { + return true + } +} + +const nativeSession = { + id: "session-1", + version: 1, + cwd: "/repo", + title: "Repro", + parent: null, + forkFromId: null, + atTurnId: null, + createdAt: "2026-08-30T07:00:00.000Z", + lastActivityAt: "2026-08-30T07:00:00.000Z", + status: "idle", + flags: [], + archived: false, + ephemeral: false, + model: { provider: "unknown", model: "test-model", reasoningEffort: "high" }, + settings: { permissionProfile: "default", mode: "plan", reasoningEffort: "high" }, + preview: "", + queuedCount: 0, + usage: { + total: { + inputTokens: 0, + outputTokens: 0, + cacheCreationInputTokens: 0, + cacheReadInputTokens: 0, + reasoningTokens: 0, + totalTokens: 0, + callCount: 0, + meteredCallCount: 0, + failedCallCount: 0, + cancelledCallCount: 0, + }, + byPurpose: [], + updatedAt: "2026-08-30T07:00:00.000Z", + }, +} + +function envelope(method: string, eventId: string): Record { + return { + event: { + eventId, + streamId: "session:session-1", + seq: 1, + emittedAt: "2026-08-30T07:00:00.000Z", + persisted: true, + schemaVersion: 1, + }, + notification: { method, params: { session: nativeSession } }, + } +} + +/** + * Regression for the Desktop "subscription/create against + * SubscriptionCreateResult: /replay/N/notification/method must be equal to + * one of the allowed values" loop: one replay envelope whose notification + * method this schema build does not know must be skipped (replay processing + * ignores unknown methods anyway), never fail the whole subscription. + */ +describe("subscription replay forward compatibility", () => { + test("event.subscribe survives an unknown replay notification method", async () => { + const transport = new FakeTransport((method) => { + if (method === "initialize") { + return { protocolVersion: 1, agentCapabilities: {}, authMethods: [] } + } + if (method === "session/list") { + return { data: [nativeSession], nextCursor: null } + } + if (method === "subscription/create") { + return { + subscriptionId: "sub_01a051ccae687f12891cc843a70224d2", + cursors: [{ streamId: "session:session-1", seq: 3 }], + snapshots: [ + { + streamId: "session:session-1", + barrierSeq: 3, + data: { kind: "sessionsList", sessions: [nativeSession] }, + }, + ], + replay: [ + envelope("session/created", "e1"), + // The offending shape: a persisted event whose method this + // build's ServerNotification schema does not know. + envelope("session/title/updated", "e2"), + envelope("session/metadataUpdated", "e3"), + ], + } + } + if (method === "subscription/ack") { + return { serverTimeMs: 0 } + } + throw new Error(`unexpected method ${method}`) + }) + + const client = createDevoClient({ directory: "/repo", transport }) + const result = await client.event.subscribe() + + expect(typeof result.stream[Symbol.asyncIterator] === "function" || result.stream !== undefined).toBe(true) + const created = transport.requests.find((request) => request.method === "subscription/create") + expect(created).toBeDefined() + }) + + test("normalized sessions keep persisted model and settings for composer re-seeding", async () => { + const transport = new FakeTransport((method) => { + if (method === "initialize") { + return { protocolVersion: 1, agentCapabilities: {}, authMethods: [] } + } + if (method === "session/list") { + return { data: [nativeSession], nextCursor: null } + } + throw new Error(`unexpected method ${method}`) + }) + + const client = createDevoClient({ directory: "/repo", transport }) + const result = await client.session.list() + const session = (result.data as Array>)[0] + expect(session.model).toEqual({ + provider: "unknown", + model: "test-model", + reasoningEffort: "high", + }) + expect(session.settings).toEqual({ + mode: "plan", + reasoningEffort: "high", + permissionProfile: "default", + }) + }) + + test("updateSettings persists a combined selection patch in one metadata/update", async () => { + const transport = new FakeTransport((method, params) => { + if (method === "initialize") { + return { protocolVersion: 1, agentCapabilities: {}, authMethods: [] } + } + if (method === "session/metadata/update") { + expect(params).toEqual({ + sessionId: "session-1", + expectedVersion: 0, + model: { provider: "", model: "test-model" }, + settings: { reasoningEffort: "high", mode: "plan" }, + }) + return { session: nativeSession } + } + throw new Error(`unexpected method ${method}`) + }) + + const client = createDevoClient({ directory: "/repo", transport }) + await client.session.updateSettings({ + sessionID: "session-1", + modelID: "test-model", + reasoningEffort: "high", + mode: "plan", + }) + expect( + transport.requests.some((request) => request.method === "session/metadata/update"), + ).toBe(true) + + // An empty patch must not hit the wire at all. + const before = transport.requests.length + await client.session.updateSettings({ sessionID: "session-1" }) + expect(transport.requests.length).toBe(before) + }) + + test("updateSettings writes a cold session without resuming it", async () => { + const transport = new FakeTransport((method, params) => { + if (method === "initialize") { + return { protocolVersion: 1, agentCapabilities: {}, authMethods: [] } + } + if (method === "session/metadata/update") { + expect(params).toMatchObject({ sessionId: "session-1" }) + return { session: nativeSession } + } + throw new Error(`unexpected method ${method}`) + }) + + const client = createDevoClient({ directory: "/repo", transport }) + await client.session.updateSettings({ sessionID: "session-1", modelID: "test-model" }) + const calls = transport.requests.map((request) => request.method) + expect(calls).not.toContain("session/resume") + expect(calls.filter((method) => method === "session/metadata/update").length).toBe(1) + }) + + test("updateSettings serializes in-flight changes and preserves the latest patch", async () => { + let releaseFirst: (() => void) | undefined + const firstRequest = new Promise((resolve) => { + releaseFirst = resolve + }) + const transport = new FakeTransport(async (method, params) => { + if (method === "initialize") { + return { protocolVersion: 1, agentCapabilities: {}, authMethods: [] } + } + if (method === "session/metadata/update") { + const request = params as { settings?: Record } + if (!request.settings?.mode) await firstRequest + return { + session: { + ...nativeSession, + settings: { ...nativeSession.settings, ...(request.settings ?? {}) }, + }, + } + } + throw new Error(`unexpected method ${method}`) + }) + + const client = createDevoClient({ directory: "/repo", transport }) + const first = client.session.updateSettings({ sessionID: "session-1", modelID: "model-a" }) + await Promise.resolve() + const second = client.session.updateSettings({ sessionID: "session-1", mode: "plan" }) + releaseFirst?.() + await Promise.all([first, second]) + + const updates = transport.requests.filter((request) => request.method === "session/metadata/update") + expect(updates).toHaveLength(2) + expect(updates[1]?.params).toMatchObject({ settings: { mode: "plan" } }) + expect("model" in (updates[1]?.params ?? {})).toBe(false) + }) + + test("retrySettings replays a failed patch retained by the queue", async () => { + let failed = true + const transport = new FakeTransport((method) => { + if (method === "initialize") { + return { protocolVersion: 1, agentCapabilities: {}, authMethods: [] } + } + if (method === "session/metadata/update") { + if (failed) throw new Error('{"code":"InvalidParams","message":"rejected"}') + return { session: nativeSession } + } + throw new Error(`unexpected method ${method}`) + }) + + const client = createDevoClient({ directory: "/repo", transport }) + await expect( + client.session.updateSettings({ sessionID: "session-1", modelID: "test-model" }), + ).rejects.toThrow("rejected") + failed = false + await client.session.retrySettings({ sessionID: "session-1" }) + expect( + transport.requests.filter((request) => request.method === "session/metadata/update"), + ).toHaveLength(2) + }) +}) diff --git a/apps/desktop/packages/devo-ai-sdk/src/v2/client.ts b/apps/desktop/packages/devo-ai-sdk/src/v2/client.ts index acfead80..907f4f0a 100644 --- a/apps/desktop/packages/devo-ai-sdk/src/v2/client.ts +++ b/apps/desktop/packages/devo-ai-sdk/src/v2/client.ts @@ -10,10 +10,8 @@ import { partTime, providerDataFromConfigOptions, questionInfoFromNative, - requestUserInputFromOriginalEvent, sessionErrorEvent, stableId, - statusFromDevo, textFromUpdate, toolCallIdFromUpdate, toolPartFromUpdate, @@ -40,6 +38,7 @@ import type { import { ProtocolValidationError, assertValidProtocolPayload, + dropUnknownReplayEnvelopes, } from "./protocol-validation" import { ReferenceSearchSession, @@ -53,19 +52,6 @@ export type { export type JsonRpcId = number | string -type LegacySessionInfo = { - sessionId: string - cwd: string - title?: string - updatedAt?: string - _meta?: Record -} -type LegacySessionNotification = { - sessionId: string - update: Record - _meta?: Record -} - export interface DevoNativeTransportEvent { type: "notification" | "request" | "closed" id?: JsonRpcId @@ -252,10 +238,18 @@ function partCacheKey(sessionId: string, messageId: string): string { return `${sessionId}\u001f${messageId}` } +function renderedNativeItemKey(sessionId: string, itemId: string): string { + return partCacheKey(sessionId, itemId) +} + function objectRecord(value: unknown): Record | undefined { return value && typeof value === "object" ? (value as Record) : undefined } +function stringOrUndefined(value: unknown): string | undefined { + return typeof value === "string" && value.length > 0 ? value : undefined +} + /** Maps a Native FileChangeKind entry into flat tool-input fields the UI can render. */ function mapFileChangeKind(change: Record | undefined): { changeType?: string @@ -477,61 +471,6 @@ function unwrapToolOutputValue(output: unknown): unknown { return output } -function sessionMeta(value: unknown): Record | undefined { - const meta = objectRecord(value) - return objectRecord(meta?.["devo/session"]) -} - -function providerRetryStatusFromOriginalEvent( - original: Record, - originalMethod?: string, -): Record | null { - if (originalMethod !== "turn/provider_retry_status" && !("TurnProviderRetryStatus" in original) && original.kind !== "turn_provider_retry_status") { - return null - } - const payload = objectRecord(original.TurnProviderRetryStatus) ?? original - const sessionID = String(payload.session_id ?? payload.sessionId ?? "") - const turnID = String(payload.turn_id ?? payload.turnId ?? "") - if (!sessionID || !turnID) return null - return { - sessionID, - turnID, - attempt: numberFromProtocol(payload.attempt), - backoffMs: numberFromProtocol(payload.backoff_ms ?? payload.backoffMs), - provider: String(payload.provider ?? ""), - model: String(payload.model ?? ""), - phase: String(payload.phase ?? ""), - message: String(payload.message ?? ""), - } -} - -function turnFailureFromOriginalEvent( - original: Record, - originalMethod?: string, -): { sessionID: string; code: string; message: string } | null { - if (originalMethod !== "turn/failed" && !("TurnFailed" in original) && original.kind !== "turn_failed") { - return null - } - const payload = objectRecord(original.TurnFailed) ?? original - const sessionID = String(payload.session_id ?? payload.sessionId ?? "") - if (!sessionID) return null - const error = objectRecord(payload.error) - if (!error || typeof error.message !== "string" || !error.message.trim()) return null - return { - sessionID, - code: String(error.code ?? "TURN_FAILED"), - message: error.message, - } -} - -function sessionStatusFromMetadata(value: unknown): string | undefined { - const meta = objectRecord(value) - const nestedStatus = objectRecord(meta?.["devo/session"])?.status - if (typeof nestedStatus === "string") return nestedStatus - const directStatus = meta?.["devo/session.status"] - return typeof directStatus === "string" ? directStatus : undefined -} - function numberFromProtocol(value: unknown): number { if (typeof value === "number" && Number.isFinite(value)) return value if (typeof value === "bigint") return Number(value) @@ -581,32 +520,6 @@ function workspaceChangeStats(value: unknown): WorkspaceChangeStats { } } -function workspaceChangesUpdatedFromOriginalEvent( - original: unknown, -): WorkspaceChangesUpdatedPayload | null { - const event = objectRecord(original) - if (!event) return null - const payload = - event.kind === "workspace_changes_updated" - ? event - : objectRecord(event.WorkspaceChangesUpdated) ?? - objectRecord(event.workspace_changes_updated) - if (!payload) return null - return { - session_id: String(payload.session_id ?? payload.sessionId ?? ""), - turn_id: String(payload.turn_id ?? payload.turnId ?? ""), - scope: String(payload.scope ?? "turn") as WorkspaceChangeScope, - status: String(payload.status ?? "ready") as WorkspaceChangeViewStatus, - coverage: String(payload.coverage ?? "none") as WorkspaceChangeCoverage, - change_set_status: String( - payload.change_set_status ?? payload.changeSetStatus ?? "finalized", - ) as WorkspaceChangeSetStatus, - stats: workspaceChangeStats(payload.stats), - version: numberFromProtocol(payload.version), - generated_at: String(payload.generated_at ?? payload.generatedAt ?? ""), - } -} - // ── Canonical provider conversions (ratified #11) ── function canonicalProviderVendorWire(vendor: ProviderVendor): Record { @@ -763,138 +676,6 @@ function legacyWorkspaceChangeBaseFromCanonical( } as WorkspaceChangeBase } -function deletedSessionIdsFromOriginalEvent(original: unknown): string[] { - const event = objectRecord(original) - if (!event) return [] - const payload = - event.kind === "session_deleted" - ? event - : objectRecord(event.SessionDeleted) ?? objectRecord(event.session_deleted) - if (!payload) return [] - const rawIds = payload.deleted_session_ids ?? payload.deletedSessionIds - if (Array.isArray(rawIds)) return rawIds.map(String).filter(Boolean) - const sessionId = payload.session_id ?? payload.sessionId - return sessionId ? [String(sessionId)] : [] -} - -function sessionStatusChangedFromOriginalEvent( - original: unknown, - originalMethod?: string, -): { sessionId: string; status: string } | null { - const event = objectRecord(original) - if (!event) return null - const payload = - originalMethod === "session/status/changed" - ? objectRecord(event.SessionStatusChanged) ?? event - : event.kind === "session_status_changed" || event.kind === "session/status/changed" - ? event - : objectRecord(event.SessionStatusChanged) ?? - objectRecord(event.session_status_changed) ?? - objectRecord(event.sessionStatusChanged) - if (!payload) return null - const sessionId = payload.session_id ?? payload.sessionId - const status = payload.status - return typeof sessionId === "string" && typeof status === "string" ? { sessionId, status } : null -} - -function sessionIdFromCompactionPayload(payload: Record): string | null { - const direct = payload.session_id ?? payload.sessionId - if (typeof direct === "string" && direct) return direct - const context = objectRecord(payload.context) - const contextual = context?.session_id ?? context?.sessionId - if (typeof contextual === "string" && contextual) return contextual - const session = objectRecord(payload.session) - const nested = session?.session_id ?? session?.sessionId - return typeof nested === "string" && nested ? nested : null -} - -function sessionCompactionFromOriginalEvent( - original: unknown, - originalMethod?: string, -): { - sessionId: string - status: "started" | "completed" | "failed" - message?: string - itemId?: string - turnId?: string -} | null { - const event = objectRecord(original) - if (!event) return null - - let status: "started" | "completed" | "failed" | null = null - let payload: Record | undefined - let itemId: string | undefined - let turnId: string | undefined - if (originalMethod === "item/started" || originalMethod === "item/completed") { - const item = objectRecord(event.item) - if (item?.item_kind !== "context_compaction" && item?.itemKind !== "context_compaction") { - return null - } - const context = objectRecord(event.context) - const itemPayload = objectRecord(item.payload) - status = originalMethod === "item/started" - ? "started" - : itemPayload?.status === "failed" - ? "failed" - : "completed" - payload = event - const rawItemId = item.item_id ?? item.itemId - const rawTurnId = context?.turn_id ?? context?.turnId - itemId = typeof rawItemId === "string" && rawItemId ? rawItemId : undefined - turnId = typeof rawTurnId === "string" && rawTurnId ? rawTurnId : undefined - } else if (originalMethod === "session/compaction/started") { - status = "started" - payload = objectRecord(event.SessionCompactionStarted) ?? event - } else if (originalMethod === "session/compaction/completed") { - status = "completed" - payload = objectRecord(event.SessionCompactionCompleted) ?? event - } else if (originalMethod === "session/compaction/failed") { - status = "failed" - payload = objectRecord(event.SessionCompactionFailed) ?? event - } else { - const candidates: Array< - ["started" | "completed" | "failed", Record | undefined] - > = [ - ["started", objectRecord(event.SessionCompactionStarted)], - ["started", objectRecord(event.session_compaction_started)], - ["started", objectRecord(event.sessionCompactionStarted)], - ["completed", objectRecord(event.SessionCompactionCompleted)], - ["completed", objectRecord(event.session_compaction_completed)], - ["completed", objectRecord(event.sessionCompactionCompleted)], - ["failed", objectRecord(event.SessionCompactionFailed)], - ["failed", objectRecord(event.session_compaction_failed)], - ["failed", objectRecord(event.sessionCompactionFailed)], - ] - const found = candidates.find(([, value]) => value) - if (found) { - status = found[0] - payload = found[1] - } else if (event.kind === "session_compaction_started") { - status = "started" - payload = event - } else if (event.kind === "session_compaction_completed") { - status = "completed" - payload = event - } else if (event.kind === "session_compaction_failed") { - status = "failed" - payload = event - } - } - - if (!status || !payload) return null - const sessionId = sessionIdFromCompactionPayload(payload) - if (!sessionId) return null - const itemPayload = objectRecord(objectRecord(payload.item)?.payload) - const message = payload.message ?? itemPayload?.message - return { - sessionId, - status, - ...(typeof message === "string" && message ? { message } : {}), - ...(itemId ? { itemId } : {}), - ...(turnId ? { turnId } : {}), - } -} - function workspaceChangesUpdatedEventProperties( payload: WorkspaceChangesUpdatedPayload, ): WorkspaceChangesUpdatedEventProperties { @@ -924,6 +705,15 @@ function parseTimestampMs(value: unknown): number | undefined { return Number.isFinite(parsed) ? parsed : undefined } +function parseTitleState(value: unknown): string | undefined { + if (typeof value === "string") return value + if (value && typeof value === "object") { + const keys = Object.keys(value as Record) + if (keys.length === 1) return keys[0] + } + return undefined +} + /** Wire timestamps from a native ItemEnvelope into appendText/appendTool updates. */ function nativeItemTimingFields( envelope: Record, @@ -949,12 +739,29 @@ function updateEventTimeMs(update: Record, fallback: number): n } type LoadedSessionLimit = number | null +type SessionSettingsPatch = { + modelID?: string + reasoningEffort?: string + mode?: string +} + +type SessionSettingsWaiter = { + resolve: (session: Session | undefined) => void + reject: (error: unknown) => void +} + +type SessionSettingsQueue = { + pending: SessionSettingsPatch | null + waiters: SessionSettingsWaiter[] + running: Promise | null + paused: boolean +} + +const SESSION_SETTINGS_RETRY_DELAYS_MS = [250, 1_000, 2_000] as const const HISTORY_MESSAGE_ID_RE = /^(?:tool-)?history-(\d+)$/ const DEVO_TURN_ID_META = "devo/turnId" -const DEVO_ACTIVITY_AT_META = "devo/activityAt" const DEVO_HISTORY_INDEX_META = "devo/historyIndex" const DEVO_PARENT_MESSAGE_ID_META = "devo/parentMessageId" -const DEVO_TURN_DURATION_MS_META = "devo/turnDurationMs" const DEVO_ITEM_KIND_META = "devo/itemKind" const DEVO_RESEARCH_ARTIFACT_TYPE_META = "devo/researchArtifactType" const DEVO_RESEARCH_ARTIFACT_TITLE_META = "devo/researchArtifactTitle" @@ -1022,6 +829,58 @@ function loadedLimitCovers(loaded: LoadedSessionLimit | undefined, requested: nu return requested !== undefined && loaded >= requested } +function mergeSessionSettingsPatch( + base: SessionSettingsPatch | null, + patch: SessionSettingsPatch, +): SessionSettingsPatch { + return { ...(base ?? {}), ...patch } +} + +function errorRecord(error: unknown): Record | undefined { + if (error && typeof error === "object") return error as Record + if (typeof error !== "string") return undefined + try { + return objectRecord(JSON.parse(error)) + } catch { + return undefined + } +} + +function settingsErrorCode(error: unknown): string | undefined { + const record = errorRecord(error) + if (typeof record?.code === "string") return record.code + if (error instanceof Error) { + try { + const messageRecord = objectRecord(JSON.parse(error.message)) + return typeof messageRecord?.code === "string" ? messageRecord.code : undefined + } catch { + return undefined + } + } + return undefined +} + +function isTransientSessionSettingsError(error: unknown): boolean { + const record = errorRecord(error) + const code = settingsErrorCode(error) + const normalizedCode = code?.replace(/([a-z0-9])([A-Z])/g, "$1_$2").toLowerCase() + if ( + normalizedCode === "service_unavailable" || + normalizedCode === "temporary_unavailable" || + normalizedCode === "timeout" + ) { + return true + } + const status = record?.status ?? record?.statusCode + if (typeof status === "number" && status >= 500 && status <= 599) return true + const message = error instanceof Error ? error.message : String(error) + return /(?:^|\D)5\d{2}(?:\D|$)|timeout|timed out|network|fetch failed|connection|temporar(?:y|ily)|unavailable/i.test(message) +} + +function waitForSessionSettingsRetry(delayMs: number): Promise { + return new Promise((resolve) => setTimeout(resolve, delayMs)) +} + function historyMessageCreatedAt(messageId: string): number | undefined { const match = HISTORY_MESSAGE_ID_RE.exec(messageId) if (!match) return undefined @@ -1150,6 +1009,8 @@ class NativeClient { private turnSessions = new Map() private nativeItemCallIds = new Map() private sessionDiscovery = new Map>() + private sessionLoads = new Map>() + private sessionSettingsQueues = new Map() private lastEventTime = 0 private referenceSearchSession: ReferenceSearchSession | null = null @@ -1207,12 +1068,50 @@ class NativeClient { }) } }, + editMessage: async (params: { sessionID: string; itemID: string; text: string }) => { + const directory = this.sessionDirectories.get(params.sessionID) ?? this.options.directory ?? defaultCwd() + const promptStartedAt = Math.max(Date.now(), this.lastEventTime + 1) + this.promptStartedAtBySession.set(params.sessionID, promptStartedAt) + this.touchNativeSessionActivity(params.sessionID, promptStartedAt) + try { + const result = await this.requestCanonical("session/message/edit", { + sessionId: params.sessionID, + itemId: params.itemID, + expectedRevision: 0, + content: [{ type: "text", text: params.text }], + idempotencyKey: crypto.randomUUID(), + }) + if (this.sessionStatuses.get(params.sessionID)?.type !== "busy") { + const busyStatus = { type: "busy" } + this.sessionStatuses.set(params.sessionID, busyStatus) + this.emit(directory, { + type: "session.status", + properties: { sessionID: params.sessionID, status: busyStatus }, + }) + } + return { data: result } + } catch (error) { + this.promptStartedAtBySession.delete(params.sessionID) + throw error + } + }, abort: async (params: { sessionID: string }) => { const interruptParams: SessionInterruptParams = { scope: { scope: "session", sessionId: params.sessionID }, } await this.request("session/interrupt", interruptParams) }, + /** + * Persists composer selections through one per-session queue. Durable + * metadata updates do not require a prior resume, so this path is safe + * while history is still loading and returns only after the server ACKs. + */ + updateSettings: async (params: SessionSettingsPatch & { sessionID: string }) => { + return { data: await this.enqueueSessionSettings(params.sessionID, params) } + }, + retrySettings: async (params: { sessionID: string }) => { + return { data: await this.retrySessionSettings(params.sessionID) } + }, update: async (params: { sessionID: string; title: string }) => { // Canonical session/metadata/update (L2-DES-APP-008): the title // patch on the persist-first path; result is the canonical Session. @@ -1238,6 +1137,9 @@ class NativeClient { const { directory } = this.forgetSession(params.sessionID) this.emitSessionDeleted(params.sessionID, directory) }, + // `session.get` is a durable snapshot read. It intentionally does not + // resume the server actor; callers that need history use `messages`, + // while metadata updates can write the snapshot directly. get: async (params: { sessionID: string }) => ({ data: await this.getSessionById(params.sessionID), }), @@ -1277,9 +1179,30 @@ class NativeClient { messages: async (params: { sessionID: string; limit?: number }) => ({ data: await this.sessionMessages(params.sessionID, normalizedHistoryLimit(params.limit)), }), - fork: async (params: { sessionID: string }) => ({ - data: this.sessions.get(params.sessionID), - }), + fork: async (params: { + sessionID: string + atTurnId?: string + cut?: "through" | "before" + }) => { + await this.ensureInitialized() + const result = (await this.requestCanonical("session/fork", { + sessionId: params.sessionID, + ...(params.atTurnId ? { atTurnId: params.atTurnId } : {}), + ...(params.cut ? { cut: params.cut } : {}), + })) as { session?: Record } + const sessionValue = result.session + if (!sessionValue) { + throw new Error("session/fork returned no session") + } + const session = this.rememberNativeSession(sessionValue) + await this.ensureSessionSubscription(session.id) + await this.loadSession(session.id) + this.emit(session.directory ?? this.options.directory ?? defaultCwd(), { + type: "session.created", + properties: { info: session, session }, + }) + return { data: session } + }, } turn = { @@ -1291,10 +1214,19 @@ class NativeClient { cwd?: string | null collaborationMode?: string }) => { - const model = params.model as { modelID?: string } | undefined - if (model?.modelID) await this.setSessionConfigOption(params.sessionID, "model", model.modelID) - if (params.variant) await this.setSessionConfigOption(params.sessionID, "thought_level", params.variant) - if (params.collaborationMode) await this.setSessionConfigOption(params.sessionID, "mode", params.collaborationMode) + // Model/variant selections are persisted by the composer's + // persist-on-selection path (session.updateSettings) and must NOT + // be re-derived here: callers used to pass fallback-resolved + // models (request slugs, defaults) which then overwrote the + // user's persisted choices on every send. Only the collaboration + // mode rides along — canonical turn/start carries no mode, and + // toggling mode without sending must still apply to the next turn. + if (params.collaborationMode) { + const settingsPatch: SessionSettingsPatch = { + mode: params.collaborationMode, + } + await this.enqueueSessionSettings(params.sessionID, settingsPatch) + } await this.ensureSessionSubscription(params.sessionID) const result = (await this.requestCanonical("turn/start", { sessionId: params.sessionID, @@ -1305,6 +1237,34 @@ class NativeClient { }, } + task = { + startAgent: async (params: { + sessionID: string + prompt: string + forkTurns?: string + maxTurns?: number + toolPolicy?: "inherit" | "deny_all" + ephemeral?: boolean + }) => { + await this.ensureInitialized() + const result = (await this.requestCanonical("task/start", { + kind: "agent", + sessionId: params.sessionID, + input: [{ type: "text", text: params.prompt }], + ...(params.forkTurns ? { forkTurns: params.forkTurns } : {}), + ...(params.maxTurns !== undefined ? { maxTurns: params.maxTurns } : {}), + ...(params.toolPolicy ? { toolPolicy: params.toolPolicy } : {}), + ephemeral: params.ephemeral ?? false, + idempotencyKey: crypto.randomUUID(), + })) as { itemId?: string; item_id?: string } + return { + data: { + itemId: String(result.itemId ?? result.item_id ?? ""), + }, + } + }, + } + question = { reply: async (params: { requestID: string; answers: QuestionAnswer[] }) => { await this.respondToQuestion(params.requestID, params.answers, "question.replied") @@ -1668,9 +1628,21 @@ class NativeClient { parts: this.parts.get(partCacheKey(sessionId, info.id)) ?? [], })) } - private async loadSession(sessionId: string, limit?: number): Promise { + private async loadSession(sessionId: string, limit?: number): Promise { const loadedLimit = this.loadedSessionLimits.get(sessionId) if (loadedLimitCovers(loadedLimit, limit)) return + const pending = this.sessionLoads.get(sessionId) + if (pending) return pending + const load = this.loadSessionOnce(sessionId, limit) + this.sessionLoads.set(sessionId, load) + try { + await load + } finally { + if (this.sessionLoads.get(sessionId) === load) this.sessionLoads.delete(sessionId) + } + } + + private async loadSessionOnce(sessionId: string, limit?: number): Promise { await this.ensureInitialized() const session = await this.getSessionById(sessionId) const cwd = session?.directory ?? this.sessionDirectories.get(sessionId) @@ -1678,7 +1650,16 @@ class NativeClient { const resumed = (await this.requestCanonical("session/resume", { sessionId, })) as { session: Record; lastContextOccupancy?: unknown; last_context_occupancy?: unknown } - this.rememberNativeSession(resumed.session) + const enriched = this.rememberNativeSession(resumed.session) + // The resume response carries the authoritative persisted model / + // settings for the session; cold `session/list` snapshots may lack + // them. Surface the enrichment so renderer session stores re-seed the + // composer — without this, the enriched snapshot stays buried in this + // client's internal cache and restored sessions fall back to defaults. + this.emit(enriched.directory ?? cwd, { + type: "session.updated", + properties: { info: enriched, session: enriched }, + }) this.emitContextUsage( sessionId, resumed.lastContextOccupancy ?? resumed.last_context_occupancy, @@ -1715,44 +1696,6 @@ class NativeClient { return discovery } - private rememberSession(info: LegacySessionInfo): Session { - const existing = this.sessions.get(info.sessionId) - const meta = sessionMeta(info._meta) - const metadataStatus = sessionStatusFromMetadata(info._meta) - const parsedCreated = parseTimestampMs(meta?.created_at ?? info.updatedAt) - const created = parsedCreated ?? existing?.time.created ?? Date.now() - const parsedUpdated = parseTimestampMs(meta?.updated_at ?? info.updatedAt) - const updated = parsedUpdated ?? existing?.time.updated ?? created - const parsedLastActivity = parseTimestampMs( - meta?.last_activity_at ?? (meta ? undefined : info.updatedAt), - ) - const lastActivity = parsedLastActivity ?? existing?.time.lastActivity ?? created - const session: Session = { - id: info.sessionId, - title: info.title ?? existing?.title ?? "New session", - parentID: meta?.parent_session_id ?? existing?.parentID ?? undefined, - time: { created, updated, lastActivity }, - directory: info.cwd, - totalInputTokens: meta?.total_input_tokens ?? existing?.totalInputTokens ?? 0, - totalOutputTokens: meta?.total_output_tokens ?? existing?.totalOutputTokens ?? 0, - totalTokens: meta?.total_tokens ?? existing?.totalTokens ?? 0, - totalCacheCreationTokens: - meta?.total_cache_creation_tokens ?? existing?.totalCacheCreationTokens ?? 0, - totalCacheReadTokens: meta?.total_cache_read_tokens ?? existing?.totalCacheReadTokens ?? 0, - promptTokenEstimate: meta?.prompt_token_estimate ?? existing?.promptTokenEstimate ?? 0, - lastQueryTotalTokens: meta?.last_query_total_tokens ?? existing?.lastQueryTotalTokens ?? 0, - } - this.sessions.set(session.id, session) - this.sessionDirectories.set(session.id, info.cwd) - this.sessionStatuses.set( - session.id, - metadataStatus === undefined - ? this.sessionStatuses.get(session.id) ?? statusFromDevo() - : statusFromDevo(metadataStatus), - ) - return session - } - private rememberNativeSession(info: Record): Session { const id = String(info.id ?? "") if (!id) throw new Error("Native session is missing id") @@ -1762,12 +1705,62 @@ class NativeClient { const created = parseTimestampMs(info.createdAt) ?? existing?.time.created ?? Date.now() const updated = parseTimestampMs(info.lastActivityAt) ?? existing?.time.updated ?? created const parent = objectRecord(info.parent) + const forkFromId = + typeof info.forkFromId === "string" + ? info.forkFromId + : typeof info.fork_from_id === "string" + ? info.fork_from_id + : existing?.forkFromId + const atTurnId = + typeof info.atTurnId === "string" + ? info.atTurnId + : typeof info.fork_at_turn_id === "string" + ? info.fork_at_turn_id + : existing?.atTurnId + const titleState = parseTitleState(info.titleState ?? info.title_state) + // Persisted per-session turn settings (model / reasoning effort / mode). + // The server restores these on resume; without them the Desktop composer + // cannot re-seed per-session selections after a restart and every + // session falls back to the project default. + const wireModel = objectRecord(info.model) + const wireSettings = objectRecord(info.settings) + const sessionModel = wireModel + ? { + provider: + typeof wireModel.provider === "string" + ? wireModel.provider + : existing?.model?.provider, + model: + typeof wireModel.model === "string" + ? wireModel.model + : existing?.model?.model, + reasoningEffort: + stringOrUndefined(wireModel.reasoningEffort ?? wireModel.reasoning_effort) ?? + existing?.model?.reasoningEffort, + } + : existing?.model + const sessionSettings = wireSettings + ? { + mode: stringOrUndefined(wireSettings.mode) ?? existing?.settings?.mode, + reasoningEffort: + stringOrUndefined(wireSettings.reasoningEffort ?? wireSettings.reasoning_effort) ?? + existing?.settings?.reasoningEffort, + permissionProfile: + stringOrUndefined(wireSettings.permissionProfile ?? wireSettings.permission_profile) ?? + existing?.settings?.permissionProfile, + } + : existing?.settings const session: Session = { id, - title: typeof info.title === "string" ? info.title : existing?.title ?? "New session", + title: typeof info.title === "string" ? info.title : existing?.title, + titleState: titleState ?? existing?.titleState ?? "Unset", parentID: typeof parent?.sessionId === "string" ? parent.sessionId : existing?.parentID, + forkFromId, + atTurnId, time: { created, updated, lastActivity: updated }, directory: String(info.cwd ?? existing?.directory ?? this.options.directory ?? defaultCwd()), + model: sessionModel, + settings: sessionSettings, totalInputTokens: Number(total?.inputTokens ?? existing?.totalInputTokens ?? 0), totalOutputTokens: Number(total?.outputTokens ?? existing?.totalOutputTokens ?? 0), totalTokens: Number(total?.totalTokens ?? existing?.totalTokens ?? 0), @@ -1852,6 +1845,9 @@ class NativeClient { payload: params, }) const result = await this.transport.request(method, validParams, this.options.directory) + if (method === "subscription/create" || method === "subscription/update") { + return this.validateSubscriptionResult(method, result) + } return assertValidProtocolPayload({ method, direction: "incomingResult", @@ -1859,6 +1855,24 @@ class NativeClient { }) } + /** + * Subscription results carry persisted event replay. A server build whose + * event generation differs from this client's schema can include a + * notification method this bundle does not recognize; replay processing + * ignores unknown methods anyway, so drop those envelopes instead of + * failing the whole event stream. Dropped envelopes are logged + * (rate-limited) so the generation skew stays observable. + */ + private validateSubscriptionResult(method: string, result: unknown): unknown { + const { payload, dropped } = dropUnknownReplayEnvelopes(result) + if (dropped.length > 0) reportDroppedReplayEnvelopes(method, dropped) + return assertValidProtocolPayload({ + method, + direction: "incomingResult", + payload, + }) + } + /** Native RPC path shared by all first-party Desktop consumers. */ private async requestCanonical(method: string, params: unknown): Promise { await this.ensureInitialized() @@ -2181,6 +2195,12 @@ class NativeClient { this.handleWorkspaceChangesUpdated(payload as WorkspaceChangesUpdatedPayload) return true } + if (method === "turn/superseded") { + const sessionId = String(value.sessionId ?? "") + const supersededTurnId = String(value.supersededTurnId ?? "") + if (sessionId && supersededTurnId) this.removeMessagesForTurn(sessionId, supersededTurnId) + return true + } return false } @@ -2209,15 +2229,16 @@ class NativeClient { status, turnId: String(envelope.turnId ?? ""), }) - if (completed) this.renderedNativeItems.add(id) + if (completed) this.renderedNativeItems.add(renderedNativeItemKey(sessionId, id)) return } if (itemType === "plan") { this.upsertPlan(sessionId, directory, id, item, String(envelope.turnId ?? "")) - if (completed) this.renderedNativeItems.add(id) + if (completed) this.renderedNativeItems.add(renderedNativeItemKey(sessionId, id)) return } - if (completed && this.renderedNativeItems.has(id)) { + const renderedKey = renderedNativeItemKey(sessionId, id) + if (completed && this.renderedNativeItems.has(renderedKey)) { // History dual-writes ToolResult then FileChange under the same item id. // Allow a richer FileChange (or file-shaped ToolResult) to upgrade the // nameless generic tool that won the first pass. @@ -2353,7 +2374,7 @@ class NativeClient { ...nativeItemTimingFields(envelope, { includeCompletedAt: completed }), }) } - if (completed) this.renderedNativeItems.add(id) + if (completed) this.renderedNativeItems.add(renderedNativeItemKey(sessionId, id)) } private finalizeNativeAssistantItem( @@ -2548,245 +2569,6 @@ class NativeClient { }) } - private handleSessionUpdate(notification: LegacySessionNotification): void { - const sessionId = notification.sessionId - const deletedSessionIds = deletedSessionIdsFromOriginalEvent( - notification._meta?.["devo/originalEvent"], - ) - if (deletedSessionIds.length > 0) { - this.handleDeletedSessionIds( - deletedSessionIds, - this.sessionDirectories.get(sessionId) ?? - this.sessions.get(sessionId)?.directory ?? - this.options.directory ?? - defaultCwd(), - ) - return - } - const update = notification.update as Record - const kind = typeof update.sessionUpdate === "string" ? update.sessionUpdate : undefined - let session = this.sessions.get(sessionId) - let directory = this.sessionDirectories.get(sessionId) ?? session?.directory - if (!session || !directory) { - const canApplyWithoutDiscoveredSession = - kind === "user_message_chunk" || - kind === "userMessageChunk" || - kind === "agent_message_chunk" || - kind === "agentMessageChunk" || - kind === "agent_thought_chunk" || - kind === "agentThoughtChunk" || - kind === "tool_call" || - kind === "tool_call_update" || - kind === "toolCall" || - kind === "toolCallUpdate" || - kind?.includes("tool") || - Boolean(update.toolCallId) - void this.discoverSession(sessionId) - .then((discovered) => { - if (discovered) { - this.handleSessionUpdate(notification) - return - } - if (!canApplyWithoutDiscoveredSession) return - const fallbackDirectory = this.options.directory ?? defaultCwd() - this.rememberSession({ sessionId, cwd: fallbackDirectory }) - this.handleSessionUpdate(notification) - }) - .catch((error) => { - if (canApplyWithoutDiscoveredSession) { - const fallbackDirectory = this.options.directory ?? defaultCwd() - this.rememberSession({ sessionId, cwd: fallbackDirectory }) - this.handleSessionUpdate(notification) - } else { - this.emit(this.options.directory ?? defaultCwd(), sessionErrorEvent(sessionId, error)) - } - }) - return - } - if (kind === "session_info_update" || kind === "sessionInfoUpdate") { - if (typeof update.title === "string") session.title = update.title - const meta = sessionMeta(update._meta) - const metadataUpdated = parseTimestampMs(meta?.updated_at ?? update.updatedAt) - if (metadataUpdated !== undefined) session.time.updated = metadataUpdated - - const activity = parseTimestampMs(meta?.last_activity_at) - if (activity !== undefined) session.time.lastActivity = activity - - const metadataStatus = sessionStatusFromMetadata(update._meta) - if (metadataStatus !== undefined) { - this.rememberSessionStatus(sessionId, directory, metadataStatus) - } - } - const activityAt = parseTimestampMs(updateMeta(update)?.[DEVO_ACTIVITY_AT_META]) - if (activityAt !== undefined) session.time.lastActivity = activityAt - this.emit(directory, { type: "session.updated", properties: { info: session, session } }) - this.handleOriginalEvent(sessionId, directory, notification) - - switch (kind) { - case "user_message_chunk": - case "userMessageChunk": - this.appendText(sessionId, directory, "user", "text", update) - break - case "agent_message_chunk": - case "agentMessageChunk": - this.appendText(sessionId, directory, "assistant", "text", update) - break - case "agent_thought_chunk": - case "agentThoughtChunk": - this.applyHistoryTurnDuration(sessionId, directory, update) - this.appendText(sessionId, directory, "assistant", "reasoning", update) - break - case "plan": - this.emitPlan(sessionId, directory, update) - break - case "config_option_update": - case "configOptionUpdate": - if (Array.isArray(update.configOptions) && update.configOptions.length > 0) { - this.rememberConfigOptions(sessionId, directory, update.configOptions as SessionConfigOption[]) - } - this.emit(directory, { - type: "session.config.updated", - properties: { sessionID: sessionId, configOptions: update.configOptions ?? [] }, - }) - break - case "available_commands_update": - case "availableCommandsUpdate": - this.emit(directory, { - type: "session.commands.updated", - properties: { sessionID: sessionId, commands: update.availableCommands ?? [] }, - }) - break - case "current_mode_update": - case "currentModeUpdate": - this.emit(directory, { - type: "session.mode.updated", - properties: { sessionID: sessionId, modeID: update.currentModeId }, - }) - break - case "usage_update": - case "usageUpdate": - this.emit(directory, { - type: "session.usage.updated", - properties: { - sessionID: sessionId, - used: update.used, - size: update.size, - cost: update.cost, - }, - }) - break - case "tool_call": - case "tool_call_update": - case "toolCall": - case "toolCallUpdate": - this.appendTool(sessionId, directory, update) - break - default: - if (kind?.includes("tool") || update.toolCallId) { - this.appendTool(sessionId, directory, update) - } - } - } - - private handleOriginalEvent( - sessionId: string, - directory: string, - notification: LegacySessionNotification, - ): void { - const original = notification._meta?.["devo/originalEvent"] - if (!original || typeof original !== "object") return - const originalMethod = - typeof notification._meta?.["devo/originalMethod"] === "string" - ? notification._meta["devo/originalMethod"] - : undefined - const deletedSessionIds = deletedSessionIdsFromOriginalEvent(original) - if (deletedSessionIds.length > 0) { - this.handleDeletedSessionIds(deletedSessionIds, directory) - return - } - const retryStatus = providerRetryStatusFromOriginalEvent(original as Record, originalMethod) - if (retryStatus) { - this.emit(directory, { - type: "turn.provider_retry_status", - properties: retryStatus, - }) - return - } - const turnFailure = turnFailureFromOriginalEvent(original as Record, originalMethod) - if (turnFailure) { - this.emit(directory, { - type: "session.error", - properties: { - sessionID: turnFailure.sessionID, - error: { - name: turnFailure.code, - data: { message: turnFailure.message }, - }, - }, - }) - return - } - const changedStatus = sessionStatusChangedFromOriginalEvent(original, originalMethod) - if (changedStatus) { - this.rememberSessionStatus(changedStatus.sessionId, directory, changedStatus.status) - return - } - const compaction = sessionCompactionFromOriginalEvent(original, originalMethod) - if (compaction) { - this.emit(directory, { - type: `session.compaction.${compaction.status}`, - properties: { - sessionID: compaction.sessionId, - ...(compaction.message ? { message: compaction.message } : {}), - }, - }) - if (compaction.itemId && compaction.status !== "failed") { - this.upsertCompaction(sessionId, directory, { - itemId: compaction.itemId, - status: compaction.status, - turnId: compaction.turnId, - }) - } - return - } - const payload = requestUserInputFromOriginalEvent(original) - if (payload) { - this.handleRequestUserInput(sessionId, directory, payload) - } - const workspaceChanges = workspaceChangesUpdatedFromOriginalEvent(original) - if (workspaceChanges) { - this.handleWorkspaceChangesUpdated(workspaceChanges, directory) - } - if ("ServerRequestResolved" in original) { - const payload = (original as { ServerRequestResolved: Record }) - .ServerRequestResolved - const requestId = String(payload.request_id ?? payload.requestId ?? "") - const pending = this.pendingQuestions.get(requestId) - if (!pending) return - this.pendingQuestions.delete(requestId) - this.emit(directory, { - type: "question.replied", - properties: { sessionID: pending.sessionId, requestID: requestId }, - }) - } - } - - private rememberSessionStatus(sessionId: string, directory: string, protocolStatus: string): void { - const status = statusFromDevo(protocolStatus) - this.sessionStatuses.set(sessionId, status) - this.emit(directory, { - type: "session.status", - properties: { sessionID: sessionId, status }, - }) - } - - private handleDeletedSessionIds(sessionIds: string[], fallbackDirectory: string): void { - for (const sessionId of sessionIds) { - const { directory, known } = this.forgetSession(sessionId, fallbackDirectory) - if (known) this.emitSessionDeleted(sessionId, directory) - } - } - private forgetSession( sessionId: string, fallbackDirectory = this.options.directory ?? defaultCwd(), @@ -2813,6 +2595,33 @@ class NativeClient { return { directory, known } } + private removeMessagesForTurn(sessionId: string, turnId: string): void { + const directory = this.sessionDirectories.get(sessionId) ?? this.options.directory ?? defaultCwd() + const messages = this.messages.get(sessionId) + if (!messages) return + const remaining: Message[] = [] + for (const message of messages) { + const key = partCacheKey(sessionId, message.id) + const messageTurnId = this.messageTurnIds.get(key) ?? message.turnID + if (messageTurnId !== turnId) { + remaining.push(message) + continue + } + this.parts.delete(key) + this.messageTurnIds.delete(key) + this.renderedNativeItems.delete(renderedNativeItemKey(sessionId, message.id)) + if (this.lastUserMessageBySession.get(sessionId) === message.id) { + this.lastUserMessageBySession.delete(sessionId) + } + this.emit(directory, { + type: "message.removed", + properties: { sessionID: sessionId, messageID: message.id }, + }) + } + this.messages.set(sessionId, remaining) + this.userMessageByTurn.delete(this.turnKey(sessionId, turnId)) + } + private emitSessionDeleted(sessionId: string, directory: string): void { this.emit(directory, { type: "session.deleted", @@ -2957,41 +2766,6 @@ class NativeClient { } } - private applyHistoryTurnDuration( - sessionId: string, - directory: string, - update: Record, - ): void { - const durationMs = Math.floor( - numberFromProtocol(updateMeta(update)?.[DEVO_TURN_DURATION_MS_META]), - ) - if (durationMs <= 0) return - const parentID = - updateMetaString(update, DEVO_PARENT_MESSAGE_ID_META) ?? - this.lastUserMessageBySession.get(sessionId) - if (!parentID) return - const messages = this.messages.get(sessionId) - if (!messages) return - const userMessage = messages.find( - (message) => message.id === parentID && message.role === "user", - ) - const userCreated = userMessage?.time?.created - if (typeof userCreated !== "number" || !Number.isFinite(userCreated)) return - - for (let index = messages.length - 1; index >= 0; index--) { - const message = messages[index] - if (message.role !== "assistant" || message.parentID !== parentID) continue - if (typeof message.time?.completed === "number") return - const updated = { - ...message, - time: { ...(message.time ?? {}), completed: userCreated + durationMs }, - } as Message - messages[index] = updated - this.emit(directory, { type: "message.updated", properties: { info: updated, message: updated } }) - return - } - } - private upsertCompaction( sessionId: string, directory: string, @@ -3197,21 +2971,6 @@ class NativeClient { this.emit(directory, { type: "message.part.updated", properties: { part } }) } - private emitPlan(sessionId: string, directory: string, update: Record): void { - const entries = Array.isArray(update.entries) ? update.entries : [] - const todos = entries.map((entry) => { - const value = entry as Record - return { - content: String(value.content ?? value.title ?? ""), - status: String(value.status ?? "pending"), - } - }) - this.emit(directory, { - type: "todo.updated", - properties: { sessionID: sessionId, todos }, - }) - } - private appendTool(sessionId: string, directory: string, update: Record): void { const now = this.nextEventTime() const toolCallId = toolCallIdFromUpdate(update, now) @@ -3355,25 +3114,118 @@ class NativeClient { this.configOptionsByDirectory.set(directory, configOptions) } - private async setSessionConfigOption( + private async enqueueSessionSettings( sessionId: string, - configId: string, - value: string, + patch: SessionSettingsPatch, + ): Promise { + const normalizedPatch: SessionSettingsPatch = {} + if (typeof patch.modelID === "string" && patch.modelID.length > 0) { + normalizedPatch.modelID = patch.modelID + } + if (typeof patch.reasoningEffort === "string" && patch.reasoningEffort.length > 0) { + normalizedPatch.reasoningEffort = patch.reasoningEffort + } + if (typeof patch.mode === "string" && patch.mode.length > 0) { + normalizedPatch.mode = patch.mode + } + if (Object.keys(normalizedPatch).length === 0) return this.sessions.get(sessionId) + + let queue = this.sessionSettingsQueues.get(sessionId) + if (!queue) { + queue = { pending: null, waiters: [], running: null, paused: false } + this.sessionSettingsQueues.set(sessionId, queue) + } + queue.pending = mergeSessionSettingsPatch(queue.pending, normalizedPatch) + queue.paused = false + const result = new Promise((resolve, reject) => { + queue.waiters.push({ resolve, reject }) + }) + this.startSessionSettingsDrain(sessionId, queue) + return result + } + + private startSessionSettingsDrain(sessionId: string, queue: SessionSettingsQueue): void { + if (queue.running || queue.paused || !queue.pending) return + const running = this.drainSessionSettings(sessionId, queue) + queue.running = running + } + + private async drainSessionSettings( + sessionId: string, + queue: SessionSettingsQueue, ): Promise { - const update: Record = { - sessionId, - expectedVersion: 0, - } - if (configId === "model") { - update.model = { provider: "", model: value } - } else if (configId === "thought_level") { - update.settings = { reasoningEffort: value } - } else if (configId === "mode") { - update.settings = { mode: value } - } else { - throw new Error(`unknown session config option '${configId}'`) + try { + while (!queue.paused && queue.pending) { + const patch = queue.pending + const waiters = queue.waiters + queue.pending = null + queue.waiters = [] + try { + const session = await this.persistSessionSettingsWithRetry(sessionId, patch) + if (!queue.pending) { + const directory = + session?.directory ?? + this.sessionDirectories.get(sessionId) ?? + this.options.directory ?? + defaultCwd() + this.emit(directory, { + type: "session.updated", + properties: { info: session, session }, + }) + } + for (const waiter of waiters) waiter.resolve(session) + } catch (error) { + // Keep the failed patch, merged ahead of any newer selection, so + // a manual retry or the next selection cannot lose a field. + queue.pending = mergeSessionSettingsPatch(patch, queue.pending ?? {}) + for (const waiter of waiters) waiter.reject(error) + queue.paused = true + } + } + } finally { + queue.running = null + if (!queue.paused && queue.pending) this.startSessionSettingsDrain(sessionId, queue) + } + } + + private async retrySessionSettings(sessionId: string): Promise { + const queue = this.sessionSettingsQueues.get(sessionId) + if (!queue?.pending) return this.sessions.get(sessionId) + queue.paused = false + const result = new Promise((resolve, reject) => { + queue.waiters.push({ resolve, reject }) + }) + this.startSessionSettingsDrain(sessionId, queue) + return result + } + + private async persistSessionSettingsWithRetry( + sessionId: string, + patch: SessionSettingsPatch, + ): Promise { + for (let retry = 0; ; retry += 1) { + try { + const update: Record = { + sessionId, + expectedVersion: 0, + } + if (patch.modelID) update.model = { provider: "", model: patch.modelID } + const settings: Record = {} + if (patch.reasoningEffort) settings.reasoningEffort = patch.reasoningEffort + if (patch.mode) settings.mode = patch.mode + if (Object.keys(settings).length > 0) update.settings = settings + + const result = (await this.requestCanonical("session/metadata/update", update)) as { + session?: Record + } + if (!result.session) throw new Error("session/metadata/update returned no session") + return this.rememberNativeSession(result.session) + } catch (error) { + const delay = SESSION_SETTINGS_RETRY_DELAYS_MS[retry] + if (delay === undefined || !isTransientSessionSettingsError(error)) throw error + await waitForSessionSettingsRetry(delay) + } } - await this.requestCanonical("session/metadata/update", update) } private async setDefaultConfigOption( @@ -3467,6 +3319,36 @@ export function createDevoClient(options: CreateDevoClientOptions = {}): DevoCli return new NativeClient(options) } +const DROPPED_REPLAY_LOG_INTERVAL_MS = 60_000 +let lastDroppedReplayLogAt = 0 + +/** + * Root-cause capture for forward-compatible replay handling: one line per + * minute max, carrying the offending envelope itself (Electron's log + * formatter renders nested objects as `[Object]`, so it must be stringified + * here). An empty method string means the envelope had no parseable method. + */ +function reportDroppedReplayEnvelopes(method: string, dropped: Array): void { + const now = Date.now() + if (now - lastDroppedReplayLogAt < DROPPED_REPLAY_LOG_INTERVAL_MS) return + lastDroppedReplayLogAt = now + const details = dropped + .slice(0, 3) + .map((envelope) => { + let text: string + try { + text = JSON.stringify(envelope) ?? String(envelope) + } catch { + text = String(envelope) + } + return text.slice(0, 800) + }) + .join(" | ") + console.warn( + `[devo-sdk] dropped ${dropped.length} replay envelope(s) with unknown notification method from ${method}: ${details}`, + ) +} + function sessionIdFromPayload(payload: unknown): string | null { if (!payload || typeof payload !== "object") return null const value = payload as Record diff --git a/apps/desktop/packages/devo-ai-sdk/src/v2/native-client-support.ts b/apps/desktop/packages/devo-ai-sdk/src/v2/native-client-support.ts index 660d4b75..e73ebd7b 100644 --- a/apps/desktop/packages/devo-ai-sdk/src/v2/native-client-support.ts +++ b/apps/desktop/packages/devo-ai-sdk/src/v2/native-client-support.ts @@ -151,18 +151,6 @@ export function questionInfoFromNative(question: unknown): any { } } -export function requestUserInputFromOriginalEvent( - original: unknown, -): Record | undefined { - if (!original || typeof original !== "object") return undefined - const event = original as Record - if (event.kind === "request_user_input") return event - const legacy = event.RequestUserInput - return legacy && typeof legacy === "object" - ? (legacy as Record) - : undefined -} - export function partTime( existingPart: any, now: number, diff --git a/apps/desktop/packages/devo-ai-sdk/src/v2/protocol-validation.test.ts b/apps/desktop/packages/devo-ai-sdk/src/v2/protocol-validation.test.ts index d66b9f2f..30a7525f 100644 --- a/apps/desktop/packages/devo-ai-sdk/src/v2/protocol-validation.test.ts +++ b/apps/desktop/packages/devo-ai-sdk/src/v2/protocol-validation.test.ts @@ -2,6 +2,8 @@ import { describe, expect, test } from "bun:test" import { ProtocolValidationError, assertValidProtocolPayload, + dropUnknownReplayEnvelopes, + knownServerNotificationMethods, } from "./protocol-validation" describe("desktop protocol runtime validation", () => { @@ -79,6 +81,78 @@ describe("desktop protocol runtime validation", () => { ).toThrow(ProtocolValidationError) }) + /** + * The settings snapshot carries the raw reasoning-effort selection — + * including the toggle keywords toggle/variant-style models use — while + * the ModelBinding on `session.model` keeps the typed enum. Locks the + * regenerated schema so neither side regresses back to a shared enum. + */ + test("accepts toggle-keyword reasoning selections in session snapshots", () => { + const baseSession = { + id: "session-1", + version: 1, + cwd: "/repo", + createdAt: "2026-08-30T00:00:00Z", + lastActivityAt: "2026-08-30T00:00:00Z", + status: "idle", + flags: [], + archived: false, + ephemeral: false, + queuedCount: 0, + title: null, + titleState: "unset", + parent: null, + forkFromId: null, + atTurnId: null, + preview: "", + model: { provider: "test", model: "alt-model" }, + settings: { + permissionProfile: "default", + reasoningEffort: "enabled", + mode: "plan", + }, + usage: { + total: { + inputTokens: 0, + outputTokens: 0, + cacheCreationInputTokens: 0, + cacheReadInputTokens: 0, + reasoningTokens: 0, + totalTokens: 0, + callCount: 0, + meteredCallCount: 0, + failedCallCount: 0, + cancelledCallCount: 0, + }, + byPurpose: [], + updatedAt: "2026-08-30T00:00:00Z", + }, + } + + const payload = { session: baseSession } + expect( + assertValidProtocolPayload({ + direction: "incomingResult", + method: "session/resume", + payload, + }), + ).toBe(payload) + + // The request-parameter slot on the binding is still the typed enum. + expect(() => + assertValidProtocolPayload({ + direction: "incomingResult", + method: "session/resume", + payload: { + session: { + ...baseSession, + model: { provider: "test", model: "alt-model", reasoningEffort: "enabled" }, + }, + }, + }), + ).toThrow(ProtocolValidationError) + }) + test("validates workspace changes read requests and results", () => { const requestPayload = { sessionId: "s1", @@ -212,4 +286,42 @@ describe("desktop protocol runtime validation", () => { }), ).toThrow(/unknown protocol method/) }) + + test("knows the canonical server notification methods", () => { + const known = knownServerNotificationMethods() + expect(known.has("session/metadataUpdated")).toBe(true) + expect(known.has("item/completed")).toBe(true) + expect(known.has("workspace/changes/updated")).toBe(true) + expect(known.has("session/title/updated")).toBe(false) + }) + + test("drops replay envelopes with unknown notification methods only", () => { + const validEnvelope = { + event: { eventId: "e1", streamId: "session:s1", emittedAt: 0, persisted: true, schemaVersion: 1 }, + notification: { method: "item/completed", params: { item: {} } }, + } + const unknownEnvelope = { + event: { eventId: "e2", streamId: "session:s1", emittedAt: 1, persisted: true, schemaVersion: 1 }, + notification: { method: "session/title/updated", params: { session: {} } }, + } + const payload = { + subscriptionId: "sub_1", + cursors: [], + replay: [validEnvelope, unknownEnvelope, validEnvelope], + } + + const { payload: sanitized, dropped } = dropUnknownReplayEnvelopes(payload) + expect(dropped).toEqual([unknownEnvelope]) + expect((sanitized as { replay: unknown[] }).replay).toEqual([validEnvelope, validEnvelope]) + // Nothing dropped → same reference, no copy. + const cleanPayload = { replay: [validEnvelope] } + expect(dropUnknownReplayEnvelopes(cleanPayload).payload).toBe(cleanPayload) + }) + + test("leaves non-replay payloads untouched", () => { + const payload = { data: [], nextCursor: null } + const { payload: sanitized, dropped } = dropUnknownReplayEnvelopes(payload) + expect(sanitized).toBe(payload) + expect(dropped).toEqual([]) + }) }) diff --git a/apps/desktop/packages/devo-ai-sdk/src/v2/protocol-validation.ts b/apps/desktop/packages/devo-ai-sdk/src/v2/protocol-validation.ts index 41eb5b69..b8439859 100644 --- a/apps/desktop/packages/devo-ai-sdk/src/v2/protocol-validation.ts +++ b/apps/desktop/packages/devo-ai-sdk/src/v2/protocol-validation.ts @@ -89,6 +89,70 @@ function bindingForMethod(method: string): MethodSchemaBinding | undefined { return bundle.methods[method] } +const serverNotificationMethods = new Set() +let serverNotificationMethodsLoaded = false + +/** + * Notification methods the generated schema accepts inside an + * `EventEnvelope.notification` (the `ServerNotification` oneOf). Extracted + * lazily from the generated bundle so it always matches the compiled schema. + */ +export function knownServerNotificationMethods(): ReadonlySet { + if (!serverNotificationMethodsLoaded) { + serverNotificationMethodsLoaded = true + const schema = ( + (bundle.schemas.SubscriptionCreateResult as + | { definitions?: Record } + | undefined)?.definitions?.ServerNotification as + | { oneOf?: Array<{ properties?: { method?: { enum?: unknown[] } } }> } + | undefined + ) + for (const branch of schema?.oneOf ?? []) { + for (const method of branch.properties?.method?.enum ?? []) { + if (typeof method === "string") serverNotificationMethods.add(method) + } + } + } + return serverNotificationMethods +} + +export type ReplaySanitizeResult = { + /** Original payload when nothing was dropped, otherwise a shallow copy. */ + payload: unknown + /** Envelopes removed because their notification method is unknown here. */ + dropped: Array +} + +/** + * Forward compatibility for subscription replay (Postel: be liberal in what + * you accept). The server derives persisted replay events from rollout facts, + * so a server build one generation ahead of (or behind) this client's schema + * can legitimately carry a notification method this bundle does not know. + * Replay processing ignores unknown methods anyway — dropping them before + * validation is semantically identical and keeps a single unknown envelope + * from failing the whole subscription. + */ +export function dropUnknownReplayEnvelopes(payload: unknown): ReplaySanitizeResult { + const record = payload as { replay?: unknown } | null + if (typeof record !== "object" || record === null || !Array.isArray(record.replay)) { + return { payload, dropped: [] } + } + const known = knownServerNotificationMethods() + const kept: unknown[] = [] + const dropped: Array = [] + for (const envelope of record.replay) { + const method = (envelope as { notification?: { method?: unknown } } | null)?.notification + ?.method + if (typeof method === "string" && known.has(method)) { + kept.push(envelope) + } else { + dropped.push(envelope) + } + } + if (dropped.length === 0) return { payload, dropped } + return { payload: { ...record, replay: kept }, dropped } +} + function validatorForSchema( method: string, direction: ProtocolValidationDirection, diff --git a/apps/desktop/packages/ui/src/styles/globals.css b/apps/desktop/packages/ui/src/styles/globals.css index eae9d129..d29a1b29 100644 --- a/apps/desktop/packages/ui/src/styles/globals.css +++ b/apps/desktop/packages/ui/src/styles/globals.css @@ -809,6 +809,40 @@ background-color: var(--ring); } + /* Chat transcript — hover-only scrollbar with stable gutter */ + .scrollbar-chat { + scrollbar-gutter: stable; + scrollbar-width: thin; + scrollbar-color: transparent transparent; + transition: scrollbar-color var(--duration-fast, 0.15s); + } + .scrollbar-chat:hover { + scrollbar-color: var(--border) transparent; + } + .scrollbar-chat:active { + scrollbar-color: var(--ring) transparent; + } + .scrollbar-chat::-webkit-scrollbar { + width: 8px; + height: 8px; + } + .scrollbar-chat::-webkit-scrollbar-track { + background: transparent; + } + .scrollbar-chat::-webkit-scrollbar-thumb { + background-color: transparent; + border-radius: 9999px; + border: 2px solid transparent; + background-clip: content-box; + transition: background-color var(--duration-fast, 0.15s); + } + .scrollbar-chat:hover::-webkit-scrollbar-thumb { + background-color: var(--border); + } + .scrollbar-chat:hover::-webkit-scrollbar-thumb:hover { + background-color: var(--ring); + } + /* Comfort scrollbar — larger hit target for primary navigation panes */ .scrollbar-comfort { scrollbar-width: auto; diff --git a/apps/desktop/src/main/devo-manager.ts b/apps/desktop/src/main/devo-manager.ts index 7a6cca8d..2b152467 100644 --- a/apps/desktop/src/main/devo-manager.ts +++ b/apps/desktop/src/main/devo-manager.ts @@ -1,5 +1,7 @@ import { DESKTOP_INITIALIZE_PARAMS } from "@devo-ai/sdk/v2/client" import type { JsonRpcId, NativeTransport, NativeTransportEvent, NativeTransportListener } from "./native-stdio-client" +import { execFile } from "node:child_process" +import { promisify } from "node:util" import { app } from "electron" import { DEVO_HOME_ENV, @@ -18,6 +20,8 @@ import { waitForEnv } from "./shell-env" const log = createLogger("devo-manager") +const execFileAsync = promisify(execFile) + const STDIO_URL = "stdio://local" const nativeTrafficLogStartupEnv = { [DEVO_HOME_ENV]: process.env[DEVO_HOME_ENV], @@ -80,6 +84,58 @@ export async function restartServer(): Promise { return ensureServer() } +/** + * Minimum spacing between protocol-mismatch recycles so a genuine schema bug + * in the current build cannot restart-loop the app. + */ +const PROTOCOL_RECYCLE_COOLDOWN_MS = 5 * 60_000 +let lastProtocolRecycleAt = 0 + +/** + * Recovers from a stale singleton server. + * + * At most one real devo-server runs per DEVO_HOME (`~/.devo/server.lock`); a + * spawned stdio child that loses the lock silently proxies to the holder + * (server `singleton.rs`). When the holder is an older build, this build's + * protocol schema rejects its traffic (ProtocolValidationError on e.g. + * subscription/create replay). Ask the singleton to shut down, then restart + * our own child, which becomes the real server. + * + * Returns false when the cooldown suppressed the recycle. + */ +export async function recycleServerForProtocolMismatch(reason: string): Promise { + const now = Date.now() + if (now - lastProtocolRecycleAt < PROTOCOL_RECYCLE_COOLDOWN_MS) { + log.warn( + "Skipping Devo server recycle (cooldown); if the error persists, restart the app or run 'devo server --shutdown'", + { reason }, + ) + return false + } + lastProtocolRecycleAt = now + log.warn("Recycling Devo server after protocol mismatch", { reason }) + + const program = resolveDevoProgram({ + appPath: app.getAppPath(), + env: process.env, + isPackaged: app.isPackaged, + resourcesPath: process.resourcesPath, + }) + try { + await execFileAsync(program, ["server", "--shutdown"], { + timeout: 15_000, + windowsHide: true, + }) + } catch (error) { + // No singleton is running (or it died mid-handshake): the restart below + // still brings up a fresh real server of this build. + log.debug("Singleton shutdown call did not succeed", { error: String(error) }) + } + + await restartServer() + return true +} + export async function requestNative( method: string, params?: unknown, @@ -199,5 +255,20 @@ function notifyServerReady(): void { } async function initialize(client: StdioNativeClient): Promise { - await client.request("initialize", DESKTOP_INITIALIZE_PARAMS) + try { + await client.request("initialize", DESKTOP_INITIALIZE_PARAMS) + } catch (error) { + // A timed-out initialize can still leave a warming stdio child alive; retry once + // before surfacing the error to the renderer (common when opening a session cold). + if (client.connected() && isInitializeTimeoutError(error)) { + log.warn("initialize timed out during cold start; retrying once") + await client.request("initialize", DESKTOP_INITIALIZE_PARAMS) + return + } + throw error + } +} + +function isInitializeTimeoutError(error: unknown): boolean { + return error instanceof Error && error.message.startsWith("initialize request ") } diff --git a/apps/desktop/src/main/native-stdio-client.test.ts b/apps/desktop/src/main/native-stdio-client.test.ts index b7033f90..34e11765 100644 --- a/apps/desktop/src/main/native-stdio-client.test.ts +++ b/apps/desktop/src/main/native-stdio-client.test.ts @@ -179,11 +179,13 @@ describe("StdioNativeClient", () => { test("gives MCP admin RPCs a longer timeout than ordinary requests", () => { expect({ + initialize: requestTimeoutMsForMethod("initialize", 10_000), sessionList: requestTimeoutMsForMethod("session/list", 10_000), mcpTools: requestTimeoutMsForMethod("mcp/tools", 10_000), mcpSetEnabled: requestTimeoutMsForMethod("mcp/set_enabled", 5), providerValidate: requestTimeoutMsForMethod("provider/validate", 10_000), }).toEqual({ + initialize: 60_000, sessionList: 10_000, mcpTools: 60_000, mcpSetEnabled: 60_000, diff --git a/apps/desktop/src/main/native-stdio-client.ts b/apps/desktop/src/main/native-stdio-client.ts index 116434b5..a0e730db 100644 --- a/apps/desktop/src/main/native-stdio-client.ts +++ b/apps/desktop/src/main/native-stdio-client.ts @@ -33,10 +33,15 @@ type PendingRequest = { } const REQUEST_TIMEOUT_MS = 10_000 +/** Cold-starting the managed `devo server` process can exceed the default RPC budget. */ +export const INITIALIZE_REQUEST_TIMEOUT_MS = 60_000 /** MCP admin RPCs may start a lazy server before listing tools. */ export const MCP_ADMIN_REQUEST_TIMEOUT_MS = 60_000 export function requestTimeoutMsForMethod(method: string, fallbackMs: number): number | undefined { + if (method === "initialize") { + return Math.max(fallbackMs, INITIALIZE_REQUEST_TIMEOUT_MS) + } if (method === "provider/validate") { return undefined } diff --git a/apps/desktop/src/main/notification-watcher.ts b/apps/desktop/src/main/notification-watcher.ts index 2dff1cf7..a060540d 100644 --- a/apps/desktop/src/main/notification-watcher.ts +++ b/apps/desktop/src/main/notification-watcher.ts @@ -1,4 +1,5 @@ import { createDevoClient, type DevoNativeTransport } from "@devo-ai/sdk/v2/client" +import { ProtocolValidationError } from "@devo-ai/sdk/v2/protocol-validation" import { createLogger } from "./logger" import { applyWatcherEvent, @@ -6,6 +7,7 @@ import { type SessionState as WatcherSessionState, } from "./notification-policy" import { setPermissionResponder, showNotification, updateBadgeCount } from "./notifications" +import { recycleServerForProtocolMismatch } from "./devo-manager" const log = createLogger("notification-watcher") @@ -130,6 +132,19 @@ async function connectWithRetry(client: ReturnType, sig } } catch (err) { if (signal.aborted) break + if (isProtocolValidationError(err)) { + // Our stdio child may be proxying to a stale singleton server + // whose wire shape predates this build's schema. Recycling shuts + // the singleton down and restarts our own server; restartServer() + // stops this watcher and starts a fresh one, so exit the loop. + log.error( + "Native protocol validation failed", + { reason: describeProtocolMismatch(err) }, + describeOffendingPayloadEntry(err), + ) + const recycled = await recycleServerForProtocolMismatch(describeProtocolMismatch(err)) + if (recycled) return + } log.error("Native event stream error, reconnecting", { retryDelay }, err) } @@ -141,6 +156,39 @@ async function connectWithRetry(client: ReturnType, sig } } +function isProtocolValidationError(error: unknown): error is ProtocolValidationError { + if (error instanceof ProtocolValidationError) return true + return ( + typeof error === "object" && + error !== null && + (error as { name?: unknown }).name === "ProtocolValidationError" + ) +} + +function describeProtocolMismatch(error: ProtocolValidationError): string { + return `${error.method} (${error.schemaName ?? error.direction})` +} + +/** + * Extracts the replay entry an ajv failure pointed at (instancePath like + * `/replay/5/notification/method`) so the next occurrence is diagnosable from + * the log alone — the Electron log formatter renders nested objects as + * `[Object]`, hiding the offending method otherwise. + */ +function describeOffendingPayloadEntry(error: ProtocolValidationError): string { + const match = error.errors + .map((ajvError) => /^\/replay\/(\d+)\//.exec(ajvError.instancePath ?? "")) + .find(Boolean) + if (!match) return "" + const entry = (error.payload as { replay?: unknown[] } | null)?.replay?.[Number(match[1])] + if (entry === undefined) return "" + try { + return JSON.stringify(entry).slice(0, 800) + } catch { + return String(entry).slice(0, 800) + } +} + async function consumeNativeEvents(client: ReturnType, signal: AbortSignal): Promise { beginHydration() const result = await client.event.subscribe() diff --git a/apps/desktop/src/main/protocol-mismatch-recovery.test.ts b/apps/desktop/src/main/protocol-mismatch-recovery.test.ts new file mode 100644 index 00000000..2f6f6288 --- /dev/null +++ b/apps/desktop/src/main/protocol-mismatch-recovery.test.ts @@ -0,0 +1,58 @@ +import { readFileSync } from "node:fs" +import { describe, expect, test } from "bun:test" + +const watcherSource = readFileSync(new URL("./notification-watcher.ts", import.meta.url), "utf8") +const managerSource = readFileSync(new URL("./devo-manager.ts", import.meta.url), "utf8") +const sdkPackageJson = readFileSync( + new URL("../../packages/devo-ai-sdk/package.json", import.meta.url), + "utf8", +) + +/** + * A stale singleton server (older build holding `~/.devo/server.lock`) makes our + * stdio child proxy to it, so this build's schema rejects its traffic. The + * watcher must detect that specific failure and recycle the server instead of + * retrying against incompatible code forever. + */ +describe("protocol mismatch recovery", () => { + test("notification watcher recycles the server on ProtocolValidationError", () => { + expect({ + importsError: watcherSource.includes( + 'from "@devo-ai/sdk/v2/protocol-validation"', + ), + detectsError: watcherSource.includes("isProtocolValidationError(err)"), + detectorChecksName: + watcherSource.includes('.name === "ProtocolValidationError"'), + callsRecycle: watcherSource.includes("recycleServerForProtocolMismatch("), + exitsAfterRecycle: watcherSource.includes("if (recycled) return"), + }).toEqual({ + importsError: true, + detectsError: true, + detectorChecksName: true, + callsRecycle: true, + exitsAfterRecycle: true, + }) + }) + + test("recycle shuts the singleton down and restarts the managed server", () => { + expect({ + exportsRecycle: managerSource.includes( + "export async function recycleServerForProtocolMismatch", + ), + shutsSingletonDown: managerSource.includes('["server", "--shutdown"]'), + restartsServer: managerSource.includes("await restartServer()"), + cooldownGuardsLoop: managerSource.includes("PROTOCOL_RECYCLE_COOLDOWN_MS"), + cooldownSkips: managerSource.includes("return false"), + }).toEqual({ + exportsRecycle: true, + shutsSingletonDown: true, + restartsServer: true, + cooldownGuardsLoop: true, + cooldownSkips: true, + }) + }) + + test("SDK exports the protocol-validation module for main-process consumers", () => { + expect(sdkPackageJson.includes('"./v2/protocol-validation"')).toBe(true) + }) +}) diff --git a/apps/desktop/src/main/tray-menu.test.ts b/apps/desktop/src/main/tray-menu.test.ts index 830d4903..0784adae 100644 --- a/apps/desktop/src/main/tray-menu.test.ts +++ b/apps/desktop/src/main/tray-menu.test.ts @@ -14,7 +14,7 @@ function menuShape(items: MenuItemConstructorOptions[]): unknown[] { } describe("buildDevoTrayMenuTemplate", () => { - test("builds a Devo tray menu with running, recent, usage, and actions", () => { + test("builds a Devo tray menu with running, recent, and actions", () => { const liveSessions = new Map([ [ "s1", @@ -141,25 +141,6 @@ describe("buildDevoTrayMenuTemplate", () => { ], }, { label: undefined, sublabel: undefined, enabled: undefined, type: "separator", click: false, submenu: undefined }, - { label: "Usage", sublabel: undefined, enabled: false, type: undefined, click: false, submenu: undefined }, - { label: "Tokens 37.8k", sublabel: undefined, enabled: false, type: undefined, click: false, submenu: undefined }, - { - label: "Input 28k · Output 9.8k", - sublabel: undefined, - enabled: false, - type: undefined, - click: false, - submenu: undefined, - }, - { - label: "Cache read 1.2k", - sublabel: undefined, - enabled: false, - type: undefined, - click: false, - submenu: undefined, - }, - { label: undefined, sublabel: undefined, enabled: undefined, type: "separator", click: false, submenu: undefined }, { label: "New Chat", sublabel: undefined, enabled: undefined, type: undefined, click: true, submenu: undefined }, { label: undefined, sublabel: undefined, enabled: undefined, type: "separator", click: false, submenu: undefined }, { label: "Open Devo", sublabel: undefined, enabled: undefined, type: undefined, click: true, submenu: undefined }, @@ -250,25 +231,6 @@ describe("buildDevoTrayMenuTemplate", () => { submenu: undefined, }, { label: undefined, sublabel: undefined, enabled: undefined, type: "separator", click: false, submenu: undefined }, - { label: "Usage", sublabel: undefined, enabled: false, type: undefined, click: false, submenu: undefined }, - { label: "Tokens 0", sublabel: undefined, enabled: false, type: undefined, click: false, submenu: undefined }, - { - label: "Input 0 · Output 0", - sublabel: undefined, - enabled: false, - type: undefined, - click: false, - submenu: undefined, - }, - { - label: "Cache read 0", - sublabel: undefined, - enabled: false, - type: undefined, - click: false, - submenu: undefined, - }, - { label: undefined, sublabel: undefined, enabled: undefined, type: "separator", click: false, submenu: undefined }, { label: "New Chat", sublabel: undefined, enabled: undefined, type: undefined, click: true, submenu: undefined }, { label: undefined, sublabel: undefined, enabled: undefined, type: "separator", click: false, submenu: undefined }, { label: "Open Devo", sublabel: undefined, enabled: undefined, type: undefined, click: true, submenu: undefined }, diff --git a/apps/desktop/src/main/tray-menu.ts b/apps/desktop/src/main/tray-menu.ts index 1edd1c13..8a00a134 100644 --- a/apps/desktop/src/main/tray-menu.ts +++ b/apps/desktop/src/main/tray-menu.ts @@ -27,20 +27,9 @@ interface TraySession { title: string directory: string updatedAt: number - totalInputTokens: number - totalOutputTokens: number - totalTokens: number - totalCacheReadTokens: number parentId?: string } -interface UsageSummary { - inputTokens: number - outputTokens: number - totalTokens: number - cacheReadTokens: number -} - export function buildDevoTrayMenuTemplate( options: DevoTrayMenuOptions, ): MenuItemConstructorOptions[] { @@ -77,8 +66,6 @@ export function buildDevoTrayMenuTemplate( template.push(separator()) } - template.push(...buildUsageSection(discoverySessions)) - template.push(separator()) template.push({ label: "New Chat", click: options.onNewChat, @@ -112,10 +99,6 @@ function buildRunningSection( title: titleForSession(state.title || discovered?.title), directory: state.directory || discovered?.directory || "", updatedAt: discovered?.updatedAt ?? 0, - totalInputTokens: discovered?.totalInputTokens ?? 0, - totalOutputTokens: discovered?.totalOutputTokens ?? 0, - totalTokens: discovered?.totalTokens ?? 0, - totalCacheReadTokens: discovered?.totalCacheReadTokens ?? 0, } }) .sort((left, right) => right.updatedAt - left.updatedAt) @@ -190,68 +173,20 @@ function sessionMenuItem( } } -function buildUsageSection(sessions: TraySession[]): MenuItemConstructorOptions[] { - const usage = summarizeUsage(sessions) - return [ - { label: "Usage", enabled: false }, - { label: `Tokens ${formatTokenCount(usage.totalTokens)}`, enabled: false }, - { - label: `Input ${formatTokenCount(usage.inputTokens)} · Output ${formatTokenCount( - usage.outputTokens, - )}`, - enabled: false, - }, - { label: `Cache read ${formatTokenCount(usage.cacheReadTokens)}`, enabled: false }, - ] -} - -function summarizeUsage(sessions: TraySession[]): UsageSummary { - return sessions.reduce( - (summary, session) => ({ - inputTokens: summary.inputTokens + session.totalInputTokens, - outputTokens: summary.outputTokens + session.totalOutputTokens, - totalTokens: summary.totalTokens + session.totalTokens, - cacheReadTokens: summary.cacheReadTokens + session.totalCacheReadTokens, - }), - { - inputTokens: 0, - outputTokens: 0, - totalTokens: 0, - cacheReadTokens: 0, - }, - ) -} - function normalizeDiscoverySessions(discovery: DiscoveryCache | null): TraySession[] { if (!discovery) return [] - return discovery.sessions.map((session) => { - const totalInputTokens = numericSessionField(session, "totalInputTokens") - const totalOutputTokens = numericSessionField(session, "totalOutputTokens") - const totalTokens = - numericSessionField(session, "totalTokens") || totalInputTokens + totalOutputTokens - - return { - id: String(session.id), - title: titleForSession(session.title), - directory: String(session.directory ?? ""), - updatedAt: Number(session.time?.updated ?? session.time?.created ?? 0), - parentId: session.parentID ? String(session.parentID) : undefined, - totalInputTokens, - totalOutputTokens, - totalTokens, - totalCacheReadTokens: numericSessionField(session, "totalCacheReadTokens"), - } - }) -} - -function numericSessionField(session: Session, field: string): number { - const value = session[field] - return typeof value === "number" && Number.isFinite(value) ? value : 0 + return discovery.sessions.map((session) => ({ + id: String(session.id), + title: titleForSession(session.title), + directory: String(session.directory ?? ""), + updatedAt: Number(session.time?.updated ?? session.time?.created ?? 0), + parentId: session.parentID ? String(session.parentID) : undefined, + })) } function titleForSession(title: unknown): string { - return typeof title === "string" && title.trim() ? title : "New chat" + return typeof title === "string" && title.trim() ? title : "New Chat" } function truncateTitle(title: string): string { @@ -264,16 +199,6 @@ function projectNameFromDir(directory: string): string { return parts.at(-1) ?? "/" } -function formatTokenCount(tokens: number): string { - if (tokens >= 1_000_000) return `${formatOneDecimal(tokens / 1_000_000)}m` - if (tokens >= 1_000) return `${formatOneDecimal(tokens / 1_000)}k` - return String(tokens) -} - -function formatOneDecimal(value: number): string { - return value.toFixed(1).replace(/\.0$/, "") -} - function separator(): MenuItemConstructorOptions { return { type: "separator" } } diff --git a/apps/desktop/src/renderer/atoms/derived/agents.ts b/apps/desktop/src/renderer/atoms/derived/agents.ts index 31a600d4..fd9d2bef 100644 --- a/apps/desktop/src/renderer/atoms/derived/agents.ts +++ b/apps/desktop/src/renderer/atoms/derived/agents.ts @@ -41,11 +41,14 @@ function agentEqual(prev: Agent | null, next: Agent | null): boolean { // so status changes from descendant sub-agents propagate to the sidebar. prev.currentActivity === next.currentActivity && prev.parentId === next.parentId && + prev.forkFromId === next.forkFromId && + prev.atTurnId === next.atTurnId && prev.worktreePath === next.worktreePath && prev.worktreeBranch === next.worktreeBranch && prev.createdAt === next.createdAt && prev.lastActiveAt === next.lastActiveAt && prev.hasUnreadCompletion === next.hasUnreadCompletion && + prev.titleGenerating === next.titleGenerating && prev.permissions.length === next.permissions.length && prev.questions.length === next.questions.length && prev.permissions[0] === next.permissions[0] && @@ -379,10 +382,16 @@ export const agentFamily = atomFamily((sessionId: string) => { const effectivePerm = get(effectivePermissionFamily(session.id)) const effectiveQ = get(effectiveQuestionFamily(session.id)) + const titleState = + typeof (session as { titleState?: string }).titleState === "string" + ? (session as { titleState?: string }).titleState + : undefined + const next: Agent = { id: session.id, sessionId: session.id, - name: session.title || "Untitled", + name: session.title || "New Chat", + titleGenerating: titleState === "Generating" && !session.title, status: agentStatus, environment: "local" as const, project: projectName, @@ -402,6 +411,8 @@ export const agentFamily = atomFamily((sessionId: string) => { permissions, questions, parentId: session.parentID, + forkFromId: session.forkFromId, + atTurnId: session.atTurnId, worktreePath: entry.worktreePath, worktreeBranch: entry.worktreeBranch, createdAt: created, @@ -426,7 +437,7 @@ export const sessionNameFamily = atomFamily((sessionId: string) => atom((get) => { const entry = get(sessionFamily(sessionId)) if (!entry) return undefined - return entry.session.title || "Untitled" + return entry.session.title || "New Chat" }), ) diff --git a/apps/desktop/src/renderer/atoms/session-composer.test.ts b/apps/desktop/src/renderer/atoms/session-composer.test.ts new file mode 100644 index 00000000..3e4f2ffb --- /dev/null +++ b/apps/desktop/src/renderer/atoms/session-composer.test.ts @@ -0,0 +1,158 @@ +import { describe, expect, test } from "bun:test" +import { + composerFromMessages, + composerFromPersistedModel, + composerFromSessionModel, + EMPTY_COMPOSER_STATE, + hydrateSessionComposerState, + type SessionComposerState, +} from "./session-composer" +import type { Message } from "../lib/types" + +describe("hydrateSessionComposerState", () => { + test("prefers last user message model over project default", () => { + const messages: Message[] = [ + { + id: "m1", + role: "user", + content: "hello", + model: { providerID: "openai", modelID: "gpt-4" }, + variant: "high", + } as Message, + ] + const hydrated = hydrateSessionComposerState(EMPTY_COMPOSER_STATE, messages, { + providerID: "anthropic", + modelID: "claude-3", + }) + expect(hydrated).toEqual({ + model: { providerID: "openai", modelID: "gpt-4" }, + variant: "high", + agent: null, + hasUserOverride: false, + }) + }) + + test("falls back to project default for empty sessions", () => { + const hydrated = hydrateSessionComposerState(EMPTY_COMPOSER_STATE, [], { + providerID: "anthropic", + modelID: "claude-3", + variant: "medium", + agent: "build", + }) + expect(hydrated).toEqual({ + model: { providerID: "anthropic", modelID: "claude-3" }, + variant: "medium", + agent: "build", + hasUserOverride: false, + }) + }) + + test("does not overwrite user overrides", () => { + const current: SessionComposerState = { + model: { providerID: "openai", modelID: "gpt-4o" }, + variant: undefined, + agent: null, + hasUserOverride: true, + } + const messages: Message[] = [ + { + id: "m1", + role: "user", + content: "hello", + model: { providerID: "openai", modelID: "gpt-4" }, + } as Message, + ] + expect(hydrateSessionComposerState(current, messages, undefined)).toEqual(current) + }) +}) + +describe("composerFromMessages", () => { + test("returns null when no user messages carry composer metadata", () => { + const messages: Message[] = [{ id: "m1", role: "assistant", content: "hi" } as Message] + expect(composerFromMessages(messages)).toBeNull() + }) +}) + +describe("composerFromPersistedModel", () => { + test("returns empty state when project default is missing", () => { + expect(composerFromPersistedModel(undefined)).toEqual(EMPTY_COMPOSER_STATE) + }) +}) + +describe("composerFromSessionModel", () => { + const resolve = (seed: { provider?: string; model?: string }) => { + if (seed.provider && seed.provider !== "unknown") { + return { providerID: seed.provider, modelID: seed.model ?? "" } + } + return seed.model === "deepseek-v4-flash" + ? { providerID: "deepseek", modelID: seed.model } + : null + } + + test("builds composer state from the persisted session model", () => { + expect( + composerFromSessionModel( + { model: "deepseek-v4-flash", reasoningEffort: "high" }, + resolve, + ), + ).toEqual({ + model: { providerID: "deepseek", modelID: "deepseek-v4-flash" }, + variant: "high", + agent: null, + hasUserOverride: false, + }) + }) + + test("uses the wire provider id when present", () => { + const seeded = composerFromSessionModel({ provider: "openai", model: "custom-model" }, resolve) + expect(seeded?.model).toEqual({ providerID: "openai", modelID: "custom-model" }) + }) + + test("returns null when the slug resolves to no known provider", () => { + expect(composerFromSessionModel({ model: "gone-model" }, resolve)).toBeNull() + expect(composerFromSessionModel(undefined, resolve)).toBeNull() + }) +}) + +describe("hydrateSessionComposerState with persisted session seed", () => { + test("uses the session seed when history messages carry no model metadata", () => { + const messages: Message[] = [{ id: "m1", role: "user", content: "hello" } as Message] + const seed: SessionComposerState = { + model: { providerID: "deepseek", modelID: "deepseek-v4-flash" }, + variant: "high", + agent: null, + hasUserOverride: false, + } + expect( + hydrateSessionComposerState(EMPTY_COMPOSER_STATE, messages, undefined, seed), + ).toEqual(seed) + }) + + test("current session settings win over older message metadata", () => { + const messages: Message[] = [ + { + id: "m1", + role: "user", + content: "hello", + model: { providerID: "openai", modelID: "gpt-4" }, + } as Message, + ] + const seed: SessionComposerState = { + model: { providerID: "deepseek", modelID: "deepseek-v4-flash" }, + variant: undefined, + agent: null, + hasUserOverride: false, + } + const hydrated = hydrateSessionComposerState(EMPTY_COMPOSER_STATE, messages, undefined, seed) + expect(hydrated).toEqual(seed) + }) + + test("keeps empty composer for non-empty sessions without metadata or seed", () => { + const messages: Message[] = [{ id: "m1", role: "user", content: "hello" } as Message] + const hydrated = hydrateSessionComposerState(EMPTY_COMPOSER_STATE, messages, { + providerID: "anthropic", + modelID: "claude-3", + }) + expect(hydrated.model).toBeNull() + }) +}) diff --git a/apps/desktop/src/renderer/atoms/session-composer.ts b/apps/desktop/src/renderer/atoms/session-composer.ts new file mode 100644 index 00000000..02a53399 --- /dev/null +++ b/apps/desktop/src/renderer/atoms/session-composer.ts @@ -0,0 +1,135 @@ +import { atom } from "jotai" +import { atomFamily } from "jotai/utils" +import type { ModelRef } from "../hooks/use-devo-data" +import type { Message } from "../lib/types" +import type { PersistedModelRef } from "./preferences" + +export interface SessionComposerState { + model: ModelRef | null + variant?: string + agent: string | null + /** Set when the user explicitly changes composer settings for this session. */ + hasUserOverride: boolean +} + +const EMPTY_COMPOSER_STATE: SessionComposerState = { + model: null, + variant: undefined, + agent: null, + hasUserOverride: false, +} + +export { EMPTY_COMPOSER_STATE } + +export const sessionComposerFamily = atomFamily((_sessionId: string) => + atom(EMPTY_COMPOSER_STATE), +) + +export function composerFromPersistedModel( + stored: PersistedModelRef | undefined, +): SessionComposerState { + if (!stored?.providerID || !stored?.modelID) { + return EMPTY_COMPOSER_STATE + } + return { + model: { providerID: stored.providerID, modelID: stored.modelID }, + variant: stored.variant, + agent: stored.agent ?? null, + hasUserOverride: false, + } +} + +export interface SessionModelSeed { + provider?: string + model?: string + reasoningEffort?: string +} + +/** + * Builds composer state from the persisted wire-session model settings. + * `resolveModel` maps the seed to a full ModelRef — preferring the wire + * provider id (`session/resume` carries a real one) and falling back to a + * reverse slug lookup across providers (cold `session/list` snapshots may + * only know `"unknown"`). + */ +export function composerFromSessionModel( + seed: SessionModelSeed | null | undefined, + resolveModel: (seed: SessionModelSeed) => ModelRef | null, +): SessionComposerState | null { + if (!seed?.model) return null + const model = resolveModel(seed) + if (!model) return null + return { + model, + variant: seed.reasoningEffort, + agent: null, + hasUserOverride: false, + } +} + +export function composerFromMessages(messages: Message[]): SessionComposerState | null { + for (let index = messages.length - 1; index >= 0; index--) { + const message = messages[index] + if (message.role !== "user") continue + const dynamic = message as Message & Record + let model: ModelRef | null = null + if ("model" in message && message.model) { + const raw = message.model as { providerID: string; modelID: string } + if (raw.providerID && raw.modelID) { + model = { providerID: raw.providerID, modelID: raw.modelID } + } + } + const variant = + typeof dynamic.variant === "string" && dynamic.variant.length > 0 + ? dynamic.variant + : undefined + const agentName = + typeof dynamic.agent === "string" && dynamic.agent.length > 0 ? dynamic.agent : null + if (model || variant || agentName) { + return { + model, + variant, + agent: agentName, + hasUserOverride: false, + } + } + } + return null +} + +export function hydrateSessionComposerState( + current: SessionComposerState, + messages: Message[], + projectDefault: PersistedModelRef | undefined, + /** Persisted per-session turn settings from the wire session (server restores them). */ + sessionSeed?: SessionComposerState | null, +): SessionComposerState { + if (current.hasUserOverride) return current + // The session snapshot is the current "next turn" configuration. It is + // newer and authoritative over model metadata from an older history item; + // history remains the fallback for legacy sessions without a snapshot seed. + if (sessionSeed) return sessionSeed + const fromMessages = composerFromMessages(messages) + if (fromMessages) return fromMessages + if (messages.length > 0) return current + return composerFromPersistedModel(projectDefault) +} + +export const setSessionComposerAtom = atom( + null, + ( + _get, + set, + args: { + sessionId: string + patch: Partial + userOverride?: boolean + }, + ) => { + set(sessionComposerFamily(args.sessionId), (current) => ({ + ...current, + ...args.patch, + hasUserOverride: args.userOverride ? true : current.hasUserOverride, + })) + }, +) diff --git a/apps/desktop/src/renderer/atoms/ui.ts b/apps/desktop/src/renderer/atoms/ui.ts index 2318c9fc..f46ba2e7 100644 --- a/apps/desktop/src/renderer/atoms/ui.ts +++ b/apps/desktop/src/renderer/atoms/ui.ts @@ -31,8 +31,56 @@ export const settingsOverlayOpenAtom = atom(false) /** Whether Customize is showing in the main content pane (does not change the route). */ export const customizeOpenAtom = atom(false) -/** Last known scrollTop for a session's chat view (used when returning from Settings). */ -export const sessionScrollTopFamily = atomFamily((_sessionId: string) => atom(null)) +export interface SessionScrollSnapshot { + scrollTop: number + atBottom: boolean + /** Distinguishes an unvisited session from a deliberate scrollTop of 0. */ + hasSnapshot: boolean +} + +const EMPTY_SCROLL_SNAPSHOT: SessionScrollSnapshot = { + scrollTop: 0, + atBottom: true, + hasSnapshot: false, +} + +/** Last known scroll snapshot for a session's chat view. */ +export const sessionScrollSnapshotFamily = atomFamily((_sessionId: string) => + atom(EMPTY_SCROLL_SNAPSHOT), +) + +/** @deprecated Use sessionScrollSnapshotFamily */ +export const sessionScrollTopFamily = atomFamily((sessionId: string) => + atom( + (get) => { + const snapshot = get(sessionScrollSnapshotFamily(sessionId)) + return snapshot.hasSnapshot ? snapshot.scrollTop : null + }, + (get, set, scrollTop: number | null) => { + if (scrollTop == null) { + set(sessionScrollSnapshotFamily(sessionId), EMPTY_SCROLL_SNAPSHOT) + return + } + const current = get(sessionScrollSnapshotFamily(sessionId)) + set(sessionScrollSnapshotFamily(sessionId), { + ...current, + scrollTop, + hasSnapshot: true, + }) + }, + ), +) + +/** @deprecated Use sessionScrollSnapshotFamily */ +export const sessionAtBottomFamily = atomFamily((sessionId: string) => + atom( + (get) => get(sessionScrollSnapshotFamily(sessionId)).atBottom, + (get, set, atBottom: boolean) => { + const current = get(sessionScrollSnapshotFamily(sessionId)) + set(sessionScrollSnapshotFamily(sessionId), { ...current, atBottom }) + }, + ), +) // ============================================================ // Review Panel State diff --git a/apps/desktop/src/renderer/components/agent-detail.tsx b/apps/desktop/src/renderer/components/agent-detail.tsx index 8589a8ae..8d32efaa 100644 --- a/apps/desktop/src/renderer/components/agent-detail.tsx +++ b/apps/desktop/src/renderer/components/agent-detail.tsx @@ -77,6 +77,8 @@ interface AgentDetailProps { /** Structured chat turns (for Chat tab) */ chatTurns: ChatTurn[] chatLoading?: boolean + /** True when the initial fetch is in flight and no cached turns exist yet. */ + chatShowLoading?: boolean /** Whether earlier messages are currently being loaded */ chatLoadingEarlier?: boolean /** Whether there are earlier messages that can be loaded */ @@ -120,10 +122,10 @@ interface AgentDetailProps { onRedo?: () => Promise /** Whether the session is in a reverted state */ isReverted?: boolean - /** Revert to a specific message (for per-turn undo) */ - onRevertToMessage?: (messageId: string) => Promise - /** Fork from a turn boundary (messageId of the next turn's user message, or undefined for full fork) */ - onForkFromTurn?: (messageId?: string) => Promise + /** Fork from a turn boundary (protocol turn id, or undefined for tip fork) */ + onForkFromTurn?: (turnId?: string) => Promise + /** Edit and resend the latest user message */ + onEditUserMessage?: (messageId: string, text: string) => Promise /** Delete a specific part from a message (for error recovery) */ onDeletePart?: (sessionId: string, messageId: string, partId: string) => Promise } @@ -132,6 +134,7 @@ export function AgentDetail({ agent, chatTurns, chatLoading, + chatShowLoading, onStop, onApprove, onDeny, @@ -153,8 +156,8 @@ export function AgentDetail({ onUndo, onRedo, isReverted, - onRevertToMessage, onForkFromTurn, + onEditUserMessage, onDeletePart, }: AgentDetailProps) { const navigate = useNavigate() @@ -240,16 +243,15 @@ export function AgentDetail({ onToggleReviewPanel={() => setReviewPanelOpen((prev) => !prev)} /> - {/* Sub-agent breadcrumb -- navigate back to parent */} - {agent.parentId && ( + {/* Sub-agent breadcrumb — navigate back to parent (forks use the in-transcript marker) */} + {agent.parentId && !agent.forkFromId && ( - {label} - - -
{description}
- {isPlan &&
Shift + Tab to toggle
} - - - ) -} - /** - * Instant-scroll when session content finishes loading. - * - * The `` (StickToBottom) uses `initial="instant"` for the first - * paint, but messages are fetched async — by the time they arrive and render, - * the library treats the content growth as a *resize* and applies - * `resize="smooth"`, causing a visible scroll animation from top → bottom. - * - * This component sits inside `` so it can access the - * StickToBottom context. It watches for the loading→loaded transition - * and forces an instant scroll-to-bottom. + * Restores scroll position when session content finishes loading or the session + * remounts after LRU eviction. Respects saved scrollTop unless the user was at bottom. */ -function ScrollOnLoad({ loading, sessionId }: { loading: boolean; sessionId: string }) { - const { scrollToBottom } = useStickToBottomContext() +function ScrollOnLoad({ + loading, + sessionId, + isActive, +}: { + loading: boolean + sessionId: string + isActive: boolean +}) { + const { scrollToBottom, scrollRef, stopScroll } = useStickToBottomContext() const settingsOverlayOpen = useAtomValue(settingsOverlayOpenAtom) const prevLoadingRef = useRef(loading) - const prevSessionRef = useRef(sessionId) + const prevActiveRef = useRef(isActive) useLayoutEffect(() => { const wasLoading = prevLoadingRef.current - const sessionChanged = prevSessionRef.current !== sessionId + const becameActive = !prevActiveRef.current && isActive prevLoadingRef.current = loading - prevSessionRef.current = sessionId + prevActiveRef.current = isActive - if (settingsOverlayOpen || getPendingRestoreScrollTop() != null) return + if (!isActive || settingsOverlayOpen || getPendingRestoreScrollTop() != null) return - // Instant scroll when: loading just finished, or session changed while not loading - // (e.g. messages were already cached in the Jotai store) - if ((wasLoading && !loading) || (sessionChanged && !loading)) { - scrollToBottom("instant") + if ((wasLoading && !loading) || becameActive) { + const snapshot = appStore.get(sessionScrollSnapshotFamily(sessionId)) + const plan = planSessionScrollRestore(snapshot) + if (plan.action === "bottom") { + scrollToBottom("instant") + } else { + markScrollRestored(plan.scrollTop) + restoreSessionScrollWhenReady({ + sessionId, + getElement: () => scrollRef.current, + scrollTop: plan.scrollTop, + stopScroll, + onRestored: markScrollRestored, + }) + } } - }, [loading, sessionId, scrollToBottom, settingsOverlayOpen]) + }, [loading, sessionId, isActive, scrollToBottom, scrollRef, stopScroll, settingsOverlayOpen]) return null } @@ -249,23 +235,29 @@ function ScrollOnLoad({ loading, sessionId }: { loading: boolean; sessionId: str * Tracks scroll position while the session is visible so it can be restored * after returning from Settings (StickToBottom may reset on layout changes). */ -function ScrollPositionTracker({ sessionId }: { sessionId: string }) { +function ScrollPositionTracker({ + sessionId, + isActive, +}: { + sessionId: string + isActive: boolean +}) { const { scrollRef } = useStickToBottomContext() - const setScrollTop = useSetAtom(sessionScrollTopFamily(sessionId)) + const setSnapshot = useSetAtom(sessionScrollSnapshotFamily(sessionId)) const settingsOverlayOpen = useAtomValue(settingsOverlayOpenAtom) useEffect(() => { const element = scrollRef.current - if (!element) return + if (!element || !isActive) return const onScroll = () => { if (settingsOverlayOpen) return - setScrollTop(element.scrollTop) + setSnapshot(snapshotFromScrollElement(element)) } element.addEventListener("scroll", onScroll, { passive: true }) return () => element.removeEventListener("scroll", onScroll) - }, [scrollRef, sessionId, setScrollTop, settingsOverlayOpen]) + }, [scrollRef, sessionId, isActive, setSnapshot, settingsOverlayOpen]) return null } @@ -315,7 +307,7 @@ function SettingsScrollGuard({ sessionId }: { sessionId: string }) { return null } -interface ScrollHandle { +export interface ChatScrollHandle { scrollToBottom: (behavior?: "instant" | "smooth") => void /** Returns the current scrollHeight of the scroll container */ getScrollHeight: () => number @@ -331,7 +323,7 @@ interface ScrollHandle { * can force a scroll-to-bottom even when the user has scrolled away. * Also exposes scroll position helpers for load-earlier anchor restore. */ -function ScrollBridge({ scrollRef }: { scrollRef: React.RefObject }) { +function ScrollBridge({ scrollRef }: { scrollRef: React.RefObject }) { const ctx = useStickToBottomContext() useImperativeHandle( scrollRef, @@ -364,7 +356,7 @@ function LoadEarlierOnScroll({ hasEarlierMessages: boolean loadingEarlier: boolean onLoadEarlier?: () => void | Promise - scrollRef: RefObject + scrollRef: RefObject }) { const { scrollRef: containerRef, stopScroll } = useStickToBottomContext() const sentinelRef = useRef(null) @@ -474,9 +466,10 @@ function turnListRevision(turns: ChatTurn[]): string { interface VirtualizedTurnListProps { turns: ChatTurn[] renderTurn: (turn: ChatTurn, index: number) => ReactNode + sessionId: string } -function VirtualizedTurnList({ turns, renderTurn }: VirtualizedTurnListProps) { +function VirtualizedTurnList({ turns, renderTurn, sessionId }: VirtualizedTurnListProps) { const { scrollRef } = useStickToBottomContext() const turnsRevision = useMemo(() => turnListRevision(turns), [turns]) const virtualizer = useVirtualizer({ @@ -484,12 +477,26 @@ function VirtualizedTurnList({ turns, renderTurn }: VirtualizedTurnListProps) { getScrollElement: () => scrollRef.current, getItemKey: (index) => turns[index]?.id ?? index, estimateSize: (index) => estimateTurnSize(turns[index]), - overscan: 8, + overscan: 5, }) useLayoutEffect(() => { + if (!isRestoringSessionScroll(sessionId)) { + virtualizer.measure() + return + } + const pending = getPendingRestoreScrollTop() + const snapshot = appStore.get(sessionScrollSnapshotFamily(sessionId)) + const plan = planSessionScrollRestore( + pending != null + ? { scrollTop: pending, atBottom: false, hasSnapshot: true } + : snapshot, + ) + if (plan.action === "restore" && scrollRef.current) { + scrollRef.current.scrollTop = plan.scrollTop + } virtualizer.measure() - }, [turnsRevision, virtualizer]) + }, [turnsRevision, virtualizer, scrollRef, sessionId]) return (
Promise onRedo?: () => Promise isReverted?: boolean - /** Revert to a specific message (for per-turn undo) */ - onRevertToMessage?: (messageId: string) => Promise - /** Fork from a turn boundary (messageId of the next turn's user message, or undefined for full fork) */ - onForkFromTurn?: (messageId?: string) => Promise + /** Fork from a turn boundary (protocol turn id, or undefined for tip fork) */ + onForkFromTurn?: (turnId?: string) => Promise + /** Edit and resend the latest user message */ + onEditUserMessage?: (messageId: string, text: string) => Promise /** Delete a specific part from a message (for error recovery) */ onDeletePart?: (sessionId: string, messageId: string, partId: string) => Promise /** Whether the review panel is open (removes max-w constraint) */ reviewPanelOpen?: boolean + /** Parent session title for fork boundary marker */ + parentSessionName?: string + /** Whether this session is the visible panel (gates scroll tracking). */ + isActive?: boolean + /** When false, only render the transcript (used by SessionShell). */ + showComposer?: boolean + /** When false, only render the composer (used by SessionShell). */ + showTranscript?: boolean + /** Shared scroll handle when composer is rendered outside the transcript. */ + externalScrollRef?: RefObject + /** Composer height when rendered outside this ChatView instance. */ + composerInsetPx?: number + /** Registers /side handler when transcript and composer are split. */ + sideQuestionHandlerRef?: MutableRefObject<((question: string) => Promise) | null> +} + +type SideCard = { + id: string + question: string + answer: string + status: "running" | "done" | "failed" +} + +type SelectionPersistPatch = { + modelID?: string + reasoningEffort?: string + mode?: string } /** @@ -713,6 +749,7 @@ interface ChatViewProps { export function ChatView({ turns, loading, + showLoading = false, loadingEarlier, hasEarlierMessages, onLoadEarlier, @@ -732,10 +769,17 @@ export function ChatView({ onUndo, onRedo, isReverted, - onRevertToMessage, onForkFromTurn, + onEditUserMessage, onDeletePart, reviewPanelOpen, + parentSessionName, + isActive = true, + showComposer = true, + showTranscript = true, + externalScrollRef, + composerInsetPx, + sideQuestionHandlerRef, }: ChatViewProps) { const isWorking = agent.status === "running" const settingsOverlayOpen = useAtomValue(settingsOverlayOpenAtom) @@ -757,19 +801,123 @@ export function ChatView({ // Ref to imperatively scroll the conversation to bottom from outside the // tree (e.g. after sending a message or answering a question). - const scrollRef = useRef(null) + const internalScrollRef = useRef(null) + const scrollRef = externalScrollRef ?? internalScrollRef const composerRef = useRef(null) - const [composerInset, setComposerInset] = useState(0) + const [measuredComposerInset, setMeasuredComposerInset] = useState(0) + const composerInset = composerInsetPx ?? measuredComposerInset // Session-level error and setup phase from the session atom const sessionEntry = useAtomValue(sessionFamily(agent.sessionId)) const sessionError = sessionEntry?.error const setupPhase = sessionEntry?.setupPhase const compactionStatus = useAtomValue(compactionStatusFamily(agent.sessionId)) + const [sideCards, setSideCards] = useState([]) + + const startSideQuestion = useCallback( + async (question: string) => { + if (!agent.directory) return + const client = getProjectClient(agent.directory) + if (!client?.task?.startAgent) { + log.error("task.startAgent unavailable", { sessionId: agent.sessionId }) + return + } + const cardId = crypto.randomUUID() + setSideCards((prev) => [ + ...prev, + { id: cardId, question, answer: "", status: "running" }, + ]) + const prompt = + "You are answering a /side side question in a lightweight forked agent.\n" + + "The inherited conversation is reference context only. Do not continue or modify the " + + "main session task. Answer only this side question.\n" + + "You cannot use tools in this fork: do not read files, run commands, search, or modify code. " + + "Produce one concise answer and stop.\n\n" + + `Side question:\n${question}` + try { + const result = await client.task.startAgent({ + sessionID: agent.sessionId, + prompt, + forkTurns: "all", + maxTurns: 1, + toolPolicy: "deny_all", + ephemeral: true, + }) + const itemId = result.data.itemId + const childSessionId = itemId.startsWith("item_") ? itemId.slice("item_".length) : itemId + let answer = "" + for (let attempt = 0; attempt < 40; attempt++) { + await new Promise((resolve) => setTimeout(resolve, 250)) + try { + const messages = await client.session.messages({ + sessionID: childSessionId, + limit: 50, + }) + const texts: string[] = [] + for (const entry of messages.data ?? []) { + if (entry.info.role !== "assistant") continue + for (const part of entry.parts ?? []) { + if (part.type === "text" && typeof part.text === "string" && part.text.trim()) { + texts.push(part.text) + } + } + } + answer = texts.join("\n").trim() + if (answer) break + } catch { + // Child may still be starting; keep polling. + } + } + setSideCards((prev) => + prev.map((card) => + card.id === cardId + ? { + ...card, + answer: answer || "No answer returned.", + status: answer ? "done" : "failed", + } + : card, + ), + ) + } catch (err) { + log.error("slash /side failed", { sessionId: agent.sessionId }, err) + setSideCards((prev) => + prev.map((card) => + card.id === cardId + ? { + ...card, + answer: err instanceof Error ? err.message : "Side question failed.", + status: "failed", + } + : card, + ), + ) + } + }, + [agent.directory, agent.sessionId], + ) + + // Clear ephemeral side cards when switching sessions. + useEffect(() => { + setSideCards([]) + }, [agent.sessionId]) + + useEffect(() => { + if (!sideQuestionHandlerRef || !isActive || showComposer) return + sideQuestionHandlerRef.current = startSideQuestion + return () => { + if (sideQuestionHandlerRef.current === startSideQuestion) { + sideQuestionHandlerRef.current = null + } + } + }, [isActive, showComposer, sideQuestionHandlerRef, startSideQuestion]) useLayoutEffect(() => { + if (composerInsetPx != null || !showComposer) { + return + } if (setupPhase) { - setComposerInset(0) + setMeasuredComposerInset(0) return } @@ -778,7 +926,7 @@ export function ChatView({ const updateComposerInset = () => { const nextInset = Math.ceil(composer.getBoundingClientRect().height) - setComposerInset((currentInset) => + setMeasuredComposerInset((currentInset) => currentInset === nextInset ? currentInset : nextInset, ) } @@ -808,7 +956,7 @@ export function ChatView({ window.removeEventListener("resize", updateComposerInset) } } - }, [setupPhase]) + }, [composerInsetPx, setupPhase, showComposer]) const effectivePermission = useAtomValue(effectivePermissionFamily(agent.sessionId)) const removePermission = useSetAtom(removePermissionAtom) @@ -903,6 +1051,23 @@ export function ChatView({ : "mx-auto w-full min-w-0 max-w-3xl" const retryStatus = sessionEntry?.retryStatus + const latestEditableUserTurnIndex = useMemo(() => { + for (let index = turns.length - 1; index >= 0; index--) { + if (!isSyntheticMessage(turns[index].userMessage)) return index + } + return -1 + }, [turns]) + + const forkBoundaryAfterIndex = useMemo( + () => + forkBoundaryAfterTurnIndex( + turns, + agent.forkFromId, + agent.atTurnId, + agent.createdAt, + ), + [agent.atTurnId, agent.createdAt, agent.forkFromId, turns], + ) const renderTurn = useCallback( (turn: ChatTurn, index: number) => { @@ -920,8 +1085,8 @@ export function ChatView({ ? retryStatus : undefined return ( + { - const nextTurn = turns[index + 1] - return onForkFromTurn(nextTurn?.userMessage.info.id) - } + ? () => onForkFromTurn(turn.turnId) + : undefined + } + onEditUserMessage={ + index === latestEditableUserTurnIndex && onEditUserMessage + ? (text) => onEditUserMessage(turn.userMessage.info.id, text) : undefined } onDeletePart={onDeletePart} @@ -954,20 +1120,31 @@ export function ChatView({ appStore.set(collaborationModeFamily(agent.sessionId), "plan") }} /> + {index === forkBoundaryAfterIndex ? ( + + ) : null} + ) }, [ agent, effectivePermission, compactionStatus, + forkBoundaryAfterIndex, handleApprovePermission, handleDenyPermission, isConnected, isWorking, + latestEditableUserTurnIndex, onDeletePart, onForkFromTurn, - onRevertToMessage, + onEditUserMessage, onSendMessage, + parentSessionName, retryStatus, turns, ], @@ -983,21 +1160,22 @@ export function ChatView({ } > {/* Chat messages -- constrained width for readability */} + {showTranscript ? (
- + - + -
+
- {loading ? ( -
- - Loading chat... -
+ {showLoading ? ( + ) : turns.length > 0 ? ( turns.length > VIRTUALIZE_TURN_THRESHOLD ? ( - + ) : ( turns.map(renderTurn) ) @@ -1024,6 +1203,26 @@ export function ChatView({
)} + {sideCards.map((card) => ( +
+
+ Side + {card.status === "running" ? ( + + ) : null} +
+

{card.question}

+ {card.answer ? ( +

{card.answer}

+ ) : ( +

Thinking…

+ )} +
+ ))} + {/* Session-level error from session.error events */} {showSessionError && sessionErrorText && (
@@ -1048,12 +1247,9 @@ export function ChatView({ className="pointer-events-none absolute inset-x-0 bottom-[var(--chat-composer-inset)] z-10 h-6 bg-gradient-to-t from-background/30 to-transparent" />
+ ) : null} - {/* Bottom input section — hidden during worktree setup since the stub session - cannot accept prompts yet. Extracted into its own component so toolbar, - popover, mention, and model-selection state changes don't re-render the - conversation turn list above. */} - {!setupPhase && ( + {showComposer && !setupPhase && (
Promise onReplyQuestion?: ChatViewProps["onReplyQuestion"] onRejectQuestion?: ChatViewProps["onRejectQuestion"] + onForkFromTurn?: ChatViewProps["onForkFromTurn"] + onStartSideQuestion?: (question: string) => Promise canRedo?: boolean onRedo?: () => Promise isReverted?: boolean - scrollRef: React.RefObject + scrollRef: React.RefObject reviewPanelOpen?: boolean } -function ChatInputSection({ +export function ChatInputSection({ agent, turns, isConnected, @@ -1128,6 +1328,8 @@ function ChatInputSection({ onDeny, onReplyQuestion, onRejectQuestion, + onForkFromTurn, + onStartSideQuestion, canRedo, onRedo, isReverted, @@ -1139,7 +1341,7 @@ function ChatInputSection({ const [activeGoal, setActiveGoal] = useState(null) const [goalAction, setGoalAction] = useState(null) const [skillPickerOpen, setSkillPickerOpen] = useState(false) - const [collaborationMode, setCollaborationMode] = useAtom( + const [collaborationMode, setCollaborationModeAtom] = useAtom( collaborationModeFamily(agent.sessionId), ) @@ -1276,69 +1478,150 @@ function ChatInputSection({ const [, setInterruptCount] = useState(0) const interruptTimerRef = useRef | null>(null) - // Toolbar state - const [selectedModel, setSelectedModel] = useState(null) - const [selectedAgent, setSelectedAgent] = useState(null) - const [selectedVariant, setSelectedVariant] = useState(undefined) - - // Initialize model, variant, and agent from the session's last user message. + // Per-session composer settings (model / variant / agent). const sessionMessages = useAtomValue(messagesFamily(agent.sessionId)) const projectModels = useAtomValue(projectModelsAtom) - const initializedForSessionRef = useRef(null) - const resetForSessionRef = useRef(null) + const composerState = useAtomValue(sessionComposerFamily(agent.sessionId)) + const setComposerState = useSetAtom(setSessionComposerAtom) + const hydratedForMessagesRef = useRef(null) + const sessionEntry = useAtomValue(sessionFamily(agent.sessionId)) + useEffect(() => { - if (resetForSessionRef.current !== agent.sessionId) { - resetForSessionRef.current = agent.sessionId - initializedForSessionRef.current = null - const stored = agent.directory ? projectModels[agent.directory] : undefined - if (stored?.providerID && stored?.modelID) { - setSelectedModel(stored) - setSelectedVariant(stored.variant) - } else { - setSelectedModel(null) - setSelectedVariant(undefined) + const wireSession = sessionEntry?.session as { + model?: { + provider?: string + model?: string + reasoningEffort?: string + reasoning_effort?: string } - setSelectedAgent(stored?.agent || null) - } - - if (initializedForSessionRef.current === agent.sessionId) return - if (!sessionMessages || sessionMessages.length === 0) return - initializedForSessionRef.current = agent.sessionId - - let foundModel = false - let foundAgent = false - for (let i = sessionMessages.length - 1; i >= 0; i--) { - const msg = sessionMessages[i] - if (msg.role !== "user") continue - const dynamic = msg as Record - - if (!foundModel && "model" in msg && msg.model) { - const model = msg.model as { providerID: string; modelID: string } - if (model.providerID && model.modelID) { - setSelectedModel(model) - foundModel = true - const variant = dynamic.variant as string | undefined - if (variant) { - setSelectedVariant(variant) - } else { - setSelectedVariant(undefined) - } + settings?: { + mode?: string + reasoningEffort?: string + reasoning_effort?: string + } + } | null | undefined + const persistedReasoningEffort = + wireSession?.model?.reasoningEffort ?? + wireSession?.model?.reasoning_effort ?? + wireSession?.settings?.reasoningEffort ?? + wireSession?.settings?.reasoning_effort + // Include the wire seed fingerprint so enrichment (session/resume + // replaces the cold list snapshot with full model/settings) re-runs + // the hydration even when the message count has not changed. + const messageKey = `${agent.sessionId}:${sessionMessages.length}:${ + wireSession?.model?.provider ?? + "" + }|${wireSession?.model?.model ?? ""}|${persistedReasoningEffort ?? ""}|${ + wireSession?.settings?.mode ?? "" + }` + if (hydratedForMessagesRef.current === messageKey) return + const projectDefault = agent.directory ? projectModels[agent.directory] : undefined + // Persisted per-session turn settings survive restarts (the server + // restores them on resume); history messages do not carry model + // metadata, so without this seed every restored session would show + // the project-default model / reasoning effort. + const wireModel = wireSession?.model + const wireModelSeed = wireModel + ? { + provider: wireModel.provider, + model: wireModel.model, + reasoningEffort: persistedReasoningEffort, + } + : undefined + const sessionSeed = composerFromSessionModel(wireModelSeed, (seed) => { + const providerList = providers?.providers ?? [] + // Prefer the wire provider id when it names a provider that actually + // serves the model (session/resume carries a real binding); cold + // list snapshots may only carry "unknown", so fall back to a + // reverse lookup by model slug. + if (seed.provider && seed.provider !== "unknown") { + const provider = providerList.find((p) => p.id === seed.provider) + if (provider?.models?.[seed.model ?? ""]) { + return { providerID: seed.provider, modelID: seed.model ?? "" } } } + return seed.model ? modelRefFromSlug(seed.model, providerList) : null + }) + const next = hydrateSessionComposerState( + composerState, + sessionMessages, + projectDefault, + sessionSeed, + ) + if ( + next.model?.providerID !== composerState.model?.providerID || + next.model?.modelID !== composerState.model?.modelID || + next.variant !== composerState.variant || + next.agent !== composerState.agent + ) { + setComposerState({ sessionId: agent.sessionId, patch: next }) + } + // Record the guard key only once hydration is conclusive. A wire seed + // that exists but could not be resolved yet (provider list still + // loading, slug unmatched against current providers) must retry on the + // next dep change — recording the key now would lock the composer into + // the default model/effort forever. Writing the ref does not render, + // so retries are driven purely by dep changes and cannot loop. + if (next.model || !wireSession?.model?.model || composerState.hasUserOverride) { + hydratedForMessagesRef.current = messageKey + } + // Seed the collaboration mode (build/plan) once per session per run so + // restarts restore it without fighting in-run user toggles. + const wireMode = wireSession?.settings?.mode + if ( + (wireMode === "plan" || wireMode === "build") && + !seededCollaborationModes.has(agent.sessionId) + ) { + seededCollaborationModes.add(agent.sessionId) + appStore.set(collaborationModeFamily(agent.sessionId), wireMode) + } + }, [ + agent.directory, + agent.sessionId, + composerState, + projectModels, + providers, + sessionEntry, + sessionMessages, + setComposerState, + ]) + + const selectedModel = composerState.model + const selectedAgent = composerState.agent + const selectedVariant = composerState.variant + + const setSelectedModel = useCallback( + (model: ModelRef | null) => { + setComposerState({ + sessionId: agent.sessionId, + patch: { model, variant: undefined }, + userOverride: true, + }) + }, + [agent.sessionId, setComposerState], + ) - if ( - !foundAgent && - dynamic.agent && - typeof dynamic.agent === "string" && - dynamic.agent.length > 0 - ) { - setSelectedAgent(dynamic.agent) - foundAgent = true - } + const setSelectedAgent = useCallback( + (agentName: string | null) => { + setComposerState({ + sessionId: agent.sessionId, + patch: { agent: agentName }, + userOverride: true, + }) + }, + [agent.sessionId, setComposerState], + ) - if (foundModel && foundAgent) break - } - }, [sessionMessages, agent.sessionId, agent.directory, projectModels]) + const setSelectedVariant = useCallback( + (variant: string | undefined) => { + setComposerState({ + sessionId: agent.sessionId, + patch: { variant }, + userOverride: true, + }) + }, + [agent.sessionId, setComposerState], + ) const { addRecent: addRecentModel } = useModelState() @@ -1369,36 +1652,108 @@ function ChatInputSection({ if (!available.includes(selectedVariant)) { setSelectedVariant(undefined) } - }, [selectedVariant, effectiveModel, providers]) + }, [selectedVariant, effectiveModel, providers, setSelectedVariant]) const modelCapabilities = useMemo( () => getModelInputCapabilities(effectiveModel, providers?.providers ?? []), [effectiveModel, providers], ) + // ── Persist-on-selection ───────────────────────────────────────────── + // Composer selections (model / reasoning effort / mode) are persisted to + // the session record the moment they change, debounced and coalesced into + // one session/metadata/update per burst, so they survive a restart even + // if no message is ever sent. The send path still passes them per turn + // (and re-persists), acting as a backstop. + const pendingSelectionPersistRef = useRef(null) + const selectionPersistTimerRef = useRef | null>(null) + + const flushSelectionPersist = useCallback((): Promise => { + if (selectionPersistTimerRef.current !== null) { + clearTimeout(selectionPersistTimerRef.current) + selectionPersistTimerRef.current = null + } + const pending = pendingSelectionPersistRef.current + pendingSelectionPersistRef.current = null + if (!pending || !agent.sessionId) return Promise.resolve() + try { + const client = + (agent.directory ? getProjectClient(agent.directory) : null) ?? getBaseClient() + if (!client) { + log.warn("session settings persist skipped: not connected", { + sessionId: agent.sessionId, + }) + return Promise.resolve() + } + const updateSettings = client.session?.updateSettings + if (typeof updateSettings !== "function") { + log.warn("session settings persist skipped: client API unavailable", { + sessionId: agent.sessionId, + }) + return Promise.resolve() + } + // Resolved (not rejected) when the write lands or has failed and + // been logged — callers await this to order the flush ahead of a + // turn without taking on error handling. + return Promise.resolve() + .then(() => updateSettings.call(client.session, { sessionID: agent.sessionId, ...pending })) + .catch((error: unknown) => { + log.warn("session settings persist failed", { sessionId: agent.sessionId }, error) + }) + } catch (error) { + log.warn("session settings persist failed", { sessionId: agent.sessionId }, error) + return Promise.resolve() + } + }, [agent.directory, agent.sessionId]) + + const scheduleSelectionPersist = useCallback( + (patch: SelectionPersistPatch) => { + pendingSelectionPersistRef.current = { ...pendingSelectionPersistRef.current, ...patch } + if (selectionPersistTimerRef.current !== null) { + clearTimeout(selectionPersistTimerRef.current) + } + selectionPersistTimerRef.current = setTimeout( + () => flushSelectionPersist(), + SELECTION_PERSIST_DEBOUNCE_MS, + ) + }, + [flushSelectionPersist], + ) + + // Unmount / session switch flushes whatever is still pending. + useEffect(() => { + return () => { + flushSelectionPersist() + } + }, [flushSelectionPersist]) + const handleModelSelect = useCallback( (model: ModelRef | null) => { setSelectedModel(model) - setSelectedVariant(undefined) if (!model) return addRecentModel(model) - if (!agent.directory) return - void persistRuntimeModelSelection(agent.directory, model).catch((err) => { - console.error("Failed to persist model selection:", err) - }) + scheduleSelectionPersist({ modelID: model.modelID }) }, - [addRecentModel, agent.directory], + [addRecentModel, scheduleSelectionPersist, setSelectedModel], ) const handleVariantSelect = useCallback( (variant: string | undefined) => { setSelectedVariant(variant) - if (!variant || !agent.directory) return - void persistRuntimeModelConfigOption(agent.directory, "thought_level", variant).catch((err) => { - console.error("Failed to persist reasoning effort selection:", err) - }) + if (typeof variant === "string" && variant.length > 0) { + scheduleSelectionPersist({ reasoningEffort: variant }) + } }, - [agent.directory], + [scheduleSelectionPersist, setSelectedVariant], + ) + + /** Mode toggle that also persists the choice to the session record. */ + const changeCollaborationMode = useCallback( + (next: CollaborationMode) => { + setCollaborationModeAtom(next) + scheduleSelectionPersist({ mode: next }) + }, + [scheduleSelectionPersist, setCollaborationModeAtom], ) const slashCommandRef = useRef<{ @@ -1472,10 +1827,11 @@ function ChatInputSection({ const spaceIndex = trimmed.indexOf(" ") const cmdName = spaceIndex === -1 ? trimmed.slice(1) : trimmed.slice(1, spaceIndex) + const args = spaceIndex === -1 ? "" : trimmed.slice(spaceIndex + 1).trim() // Product requirement: Desktop slash commands are limited to first-party - // entries. Compact executes immediately; Goal/Plan become footer trigger - // chips; Research stays as slash text so Native can run it after a question. + // entries. Compact executes immediately; Goal becomes a footer trigger + // chip; /plan switches collaboration mode; Research stays as slash text. switch (cmdName.toLowerCase()) { case "compact": if (agent.directory && effectiveModel) { @@ -1493,12 +1849,29 @@ function ChatInputSection({ } } return true + case "fork": + if (onForkFromTurn) { + try { + await onForkFromTurn() + } catch (err) { + log.error("slash /fork failed", { sessionId: agent.sessionId }, err) + } + } + return true + case "side": + case "btw": { + if (!args) { + slashCommandRef.current?.setText("/side ") + return true + } + await onStartSideQuestion?.(args) + return true + } case "goal": setActiveTrigger("goal") return true case "plan": - setActiveTrigger("plan") - setCollaborationMode("plan") + changeCollaborationMode("plan") return true case "skills": setSkillPickerOpen(true) @@ -1509,7 +1882,14 @@ function ChatInputSection({ return false } }, - [agent.directory, agent.sessionId, effectiveModel, setCollaborationMode], + [ + agent.directory, + agent.sessionId, + changeCollaborationMode, + effectiveModel, + onForkFromTurn, + onStartSideQuestion, + ], ) const submitTriggeredPrompt = useCallback( @@ -1533,17 +1913,29 @@ function ChatInputSection({ url: file.url, }) } + // Flush any still-debounced composer selection so the turn starts + // from exactly what the UI shows, then send only the explicit + // selection — never a fallback-resolved model, which would + // overwrite the persisted per-session choice. + await flushSelectionPersist() await client.session.promptAsync({ sessionID: agent.sessionId, parts, - model: effectiveModel - ? { providerID: effectiveModel.providerID, modelID: effectiveModel.modelID } + model: selectedModel + ? { providerID: selectedModel.providerID, modelID: selectedModel.modelID } : undefined, agent: selectedAgent || undefined, variant: selectedVariant, }) }, - [agent.directory, agent.sessionId, effectiveModel, selectedAgent, selectedVariant], + [ + agent.directory, + agent.sessionId, + flushSelectionPersist, + selectedModel, + selectedAgent, + selectedVariant, + ], ) const handleSend = useCallback( @@ -1575,11 +1967,15 @@ function ChatInputSection({ setSending(true) try { - if (effectiveModel && agent.directory) { + // Only an explicit composer selection may become the project's + // default model — a fallback-resolved model would poison the + // preference (and then every future fallback) with request + // slugs or defaults the user never chose. + if (selectedModel && agent.directory) { appStore.set(setProjectModelAtom, { directory: agent.directory, model: { - ...effectiveModel, + ...selectedModel, variant: selectedVariant, agent: selectedAgent || undefined, }, @@ -1611,8 +2007,12 @@ function ChatInputSection({ setTimeout(() => void refreshGoalStatus(), 1_200) } } else { + // Land any still-debounced selection before the turn, and + // send only the explicit selection — the server keeps the + // session's persisted model otherwise. + await flushSelectionPersist() await onSendMessage?.(agent, finalText, { - model: effectiveModel ?? undefined, + model: selectedModel ?? undefined, agentName: selectedAgent || undefined, variant: selectedVariant, files, @@ -1640,7 +2040,8 @@ function ChatInputSection({ onSendMessage, sending, agent, - effectiveModel, + selectedModel, + flushSelectionPersist, selectedAgent, selectedVariant, clearDraft, @@ -1791,7 +2192,7 @@ function ChatInputSection({ e.preventDefault() handleSlashClose() handleMentionClose() - setCollaborationMode((current) => (current === "plan" ? "build" : "plan")) + changeCollaborationMode(collaborationMode === "plan" ? "build" : "plan") return } @@ -1806,7 +2207,7 @@ function ChatInputSection({ handleEscapeAbort() } }, - [handleEscapeAbort, handleSlashClose, handleMentionClose, setCollaborationMode], + [handleEscapeAbort, handleSlashClose, handleMentionClose, changeCollaborationMode, collaborationMode], ) // Width constraint class: remove max-w when review panel is open @@ -1958,26 +2359,17 @@ function ChatInputSection({ onSelectVariant={handleVariantSelect} disabled={!isConnected} /> - - {activeTrigger && ( - changeCollaborationMode("build")} + /> + )} + {activeTrigger === "goal" && ( + setActiveTrigger(null)} /> )} diff --git a/apps/desktop/src/renderer/components/chat/composer-mode-chip.tsx b/apps/desktop/src/renderer/components/chat/composer-mode-chip.tsx new file mode 100644 index 00000000..407468e0 --- /dev/null +++ b/apps/desktop/src/renderer/components/chat/composer-mode-chip.tsx @@ -0,0 +1,76 @@ +import { Tooltip, TooltipContent, TooltipTrigger } from "@devo/ui/components/tooltip" +import { cn } from "@devo/ui/lib/utils" +import { GoalIcon, ListTodoIcon, XIcon } from "lucide-react" + +export type ComposerModeChipVariant = "plan" | "goal" + +const CHIP_CONFIG = { + plan: { + icon: ListTodoIcon, + label: "Plan", + description: "Plan mode — the agent will propose a plan before building", + tooltipExtra: "Shift + Tab to toggle", + }, + goal: { + icon: GoalIcon, + label: "Goal", + description: "Goal — the next message sets a session goal", + tooltipExtra: null, + }, +} as const + +export function ComposerModeChip({ + variant, + onRemove, + disabled = false, +}: { + variant: ComposerModeChipVariant + onRemove: () => void + disabled?: boolean +}) { + const config = CHIP_CONFIG[variant] + const Icon = config.icon + const isPlan = variant === "plan" + + return ( + + + } + > + + {config.label} + + +
{config.description}
+ {config.tooltipExtra ? ( +
{config.tooltipExtra}
+ ) : null} + + + ) +} diff --git a/apps/desktop/src/renderer/components/chat/fork-boundary-divider.tsx b/apps/desktop/src/renderer/components/chat/fork-boundary-divider.tsx new file mode 100644 index 00000000..69250dde --- /dev/null +++ b/apps/desktop/src/renderer/components/chat/fork-boundary-divider.tsx @@ -0,0 +1,57 @@ +import { cn } from "@devo/ui/lib/utils" +import { useNavigate } from "@tanstack/react-router" +import { SplitIcon } from "lucide-react" + +export function ForkBoundaryDivider({ + parentName, + sourceSessionId, + projectSlug, + className, +}: { + parentName?: string + sourceSessionId?: string + projectSlug?: string + className?: string +}) { + const navigate = useNavigate() + const displayName = parentName || "source session" + const label = `Forked from ${displayName}` + + const handleNavigateToSource = () => { + if (!sourceSessionId || !projectSlug) return + navigate({ + to: "/project/$projectSlug/session/$sessionId", + params: { projectSlug, sessionId: sourceSessionId }, + }) + } + + const canNavigate = Boolean(sourceSessionId && projectSlug) + + return ( +
+