mirror of
https://github.com/rowboatlabs/rowboat.git
synced 2026-07-12 21:02:17 +02:00
feat(x): sessions IPC + renderer data layer with testable stores (stage 5a)
The wire and state layer for the UI cutover; App.tsx integration follows
separately.
- shared/ipc.ts: eleven sessions:* invoke channels + the sessions:events
push feed (SessionBusEvent via z.custom, like runs:events). The generic
preload bridge needs no changes.
- main: sessions:* handlers as thin pass-throughs to the DI'd sessions
service; startSessionsWatcher forwards the session bus to all windows;
startup awaits the session-index scan before the renderer can list.
- renderer architecture per review guidance — all logic in framework-
agnostic, dependency-injected modules; hooks are thin
useSyncExternalStore subscriptions; components will consume pre-digested
view models:
- client.ts: narrow SessionsClient over window.ipc (fakeable).
- feed.ts: one shared sessions:events consumer with fan-out (factory
for tests).
- turn-view.ts: pure derivations — live overlay (deltas accumulate,
canonical events clear), TurnState -> ConversationItem[], and the
session chat state (permission/ask-human maps re-manufactured in the
runs-era shapes so existing components render unchanged;
isProcessing/isThinking contract preserved).
- store.ts: SessionChatStore (seed via getSession/getTurn, shared
reduceTurn over live events, prior-turn freezing, unknown-turn
reconciliation, stale-load guard, action routing) + SessionListStore.
- hooks/useSessionChat + useSessions: thin wrappers, deps injectable.
- renderer test infra added from scratch (vitest + jsdom +
testing-library); 29 tests across stores, pure views, feed, and hooks.
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
Ramnique Singh
parent
1d5782fa75
commit
df607ea510
17 changed files with 1840 additions and 5 deletions
|
|
@ -24,6 +24,8 @@ const execFileAsync = promisify(execFile);
|
|||
|
||||
import { RunEvent } from '@x/shared/dist/runs.js';
|
||||
import { ServiceEvent } from '@x/shared/dist/service-events.js';
|
||||
import type { SessionBusEvent } from '@x/shared/dist/sessions.js';
|
||||
import type { ISessions, EmitterSessionBus } from '@x/core/dist/sessions/index.js';
|
||||
import container from '@x/core/dist/di/container.js';
|
||||
import { listOnboardingModels } from '@x/core/dist/models/models-dev.js';
|
||||
import { testModelConnection, listModelsForProvider, generateOneShot } from '@x/core/dist/models/models.js';
|
||||
|
|
@ -634,6 +636,25 @@ export async function startRunsWatcher(): Promise<void> {
|
|||
});
|
||||
}
|
||||
|
||||
// New runtime: session bus → renderer windows (session-design.md §10).
|
||||
function emitSessionEvent(event: SessionBusEvent): void {
|
||||
const windows = BrowserWindow.getAllWindows();
|
||||
for (const win of windows) {
|
||||
if (!win.isDestroyed() && win.webContents) {
|
||||
win.webContents.send('sessions:events', event);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
let sessionsWatcher: (() => void) | null = null;
|
||||
export function startSessionsWatcher(): void {
|
||||
if (sessionsWatcher) {
|
||||
return;
|
||||
}
|
||||
const sessionBus = container.resolve<EmitterSessionBus>('sessionBus');
|
||||
sessionsWatcher = sessionBus.subscribe((event) => emitSessionEvent(event));
|
||||
}
|
||||
|
||||
let servicesWatcher: (() => void) | null = null;
|
||||
export async function startServicesWatcher(): Promise<void> {
|
||||
if (servicesWatcher) {
|
||||
|
|
@ -841,6 +862,51 @@ export function setupIpcHandlers() {
|
|||
await runsCore.deleteRun(args.runId);
|
||||
return { success: true };
|
||||
},
|
||||
// ── New runtime: sessions + turns ─────────────────────────
|
||||
// Thin pass-throughs to the sessions service. sendMessage returns the
|
||||
// turnId immediately; the turn advances in the background and the
|
||||
// renderer reconciles via the sessions:events feed. Input-routing calls
|
||||
// settle with that advance's outcome (the renderer fire-and-forgets).
|
||||
'sessions:create': async (_event, args) => {
|
||||
const sessionId = await container.resolve<ISessions>('sessions').createSession(args);
|
||||
return { sessionId };
|
||||
},
|
||||
'sessions:list': async () => {
|
||||
return { sessions: container.resolve<ISessions>('sessions').listSessions() };
|
||||
},
|
||||
'sessions:get': async (_event, args) => {
|
||||
return container.resolve<ISessions>('sessions').getSession(args.sessionId);
|
||||
},
|
||||
'sessions:getTurn': async (_event, args) => {
|
||||
return container.resolve<ISessions>('sessions').getTurn(args.turnId);
|
||||
},
|
||||
'sessions:sendMessage': async (_event, args) => {
|
||||
return container.resolve<ISessions>('sessions').sendMessage(args.sessionId, args.input, args.config);
|
||||
},
|
||||
'sessions:respondToPermission': async (_event, args) => {
|
||||
await container.resolve<ISessions>('sessions').respondToPermission(args.turnId, args.toolCallId, args.decision, args.metadata);
|
||||
return { success: true };
|
||||
},
|
||||
'sessions:respondToAskHuman': async (_event, args) => {
|
||||
await container.resolve<ISessions>('sessions').respondToAskHuman(args.turnId, args.toolCallId, args.answer);
|
||||
return { success: true };
|
||||
},
|
||||
'sessions:stopTurn': async (_event, args) => {
|
||||
await container.resolve<ISessions>('sessions').stopTurn(args.turnId, args.reason);
|
||||
return { success: true };
|
||||
},
|
||||
'sessions:resumeTurn': async (_event, args) => {
|
||||
await container.resolve<ISessions>('sessions').resumeTurn(args.sessionId);
|
||||
return { success: true };
|
||||
},
|
||||
'sessions:setTitle': async (_event, args) => {
|
||||
await container.resolve<ISessions>('sessions').setTitle(args.sessionId, args.title);
|
||||
return { success: true };
|
||||
},
|
||||
'sessions:delete': async (_event, args) => {
|
||||
await container.resolve<ISessions>('sessions').deleteSession(args.sessionId);
|
||||
return { success: true };
|
||||
},
|
||||
'runs:downloadLog': async (event, args) => {
|
||||
const runFileName = `${args.runId}.jsonl`;
|
||||
if (path.basename(runFileName) !== runFileName) {
|
||||
|
|
|
|||
|
|
@ -2,7 +2,7 @@ import { app, BrowserWindow, desktopCapturer, protocol, net, shell, session, typ
|
|||
import path from "node:path";
|
||||
import {
|
||||
setupIpcHandlers,
|
||||
startRunsWatcher,
|
||||
startRunsWatcher, startSessionsWatcher,
|
||||
startCodeSessionStatusWatcher,
|
||||
startServicesWatcher,
|
||||
startLiveNoteAgentWatcher,
|
||||
|
|
@ -45,6 +45,7 @@ import { execFileSync } from "node:child_process";
|
|||
import { init as initChromeSync } from "@x/core/dist/knowledge/chrome-extension/server/server.js";
|
||||
import container, { registerBrowserControlService, registerNotificationService } from "@x/core/dist/di/container.js";
|
||||
import type { CodeModeManager } from "@x/core/dist/code-mode/acp/manager.js";
|
||||
import type { ISessions } from "@x/core/dist/sessions/index.js";
|
||||
import { browserViewManager, BROWSER_PARTITION } from "./browser/view.js";
|
||||
import { setupBrowserEventForwarding } from "./browser/ipc.js";
|
||||
import { ElectronBrowserControlService } from "./browser/control-service.js";
|
||||
|
|
@ -388,6 +389,11 @@ app.whenReady().then(async () => {
|
|||
// start runs watcher
|
||||
startRunsWatcher();
|
||||
|
||||
// New runtime: build the in-memory session index (startup scan) before the
|
||||
// renderer can list sessions, then forward the session bus to windows.
|
||||
await container.resolve<ISessions>('sessions').initialize();
|
||||
startSessionsWatcher();
|
||||
|
||||
// start code-session status tracker (derives working/needs-you/idle + notifications)
|
||||
startCodeSessionStatusWatcher();
|
||||
|
||||
|
|
|
|||
|
|
@ -6,7 +6,9 @@
|
|||
"dev": "vite",
|
||||
"build": "tsc -b && vite build",
|
||||
"lint": "eslint .",
|
||||
"preview": "vite preview"
|
||||
"preview": "vite preview",
|
||||
"test": "vitest run",
|
||||
"test:watch": "vitest"
|
||||
},
|
||||
"dependencies": {
|
||||
"@codemirror/language": "^6.12.3",
|
||||
|
|
@ -83,6 +85,9 @@
|
|||
},
|
||||
"devDependencies": {
|
||||
"@eslint/js": "^9.39.1",
|
||||
"@testing-library/dom": "^10.4.1",
|
||||
"@testing-library/jest-dom": "^6.9.1",
|
||||
"@testing-library/react": "^16.3.2",
|
||||
"@types/node": "^24.10.1",
|
||||
"@types/react": "^19.2.5",
|
||||
"@types/react-dom": "^19.2.3",
|
||||
|
|
@ -91,9 +96,11 @@
|
|||
"eslint-plugin-react-hooks": "^7.0.1",
|
||||
"eslint-plugin-react-refresh": "^0.4.24",
|
||||
"globals": "^16.5.0",
|
||||
"jsdom": "^29.1.1",
|
||||
"tw-animate-css": "^1.4.0",
|
||||
"typescript": "~5.9.3",
|
||||
"typescript-eslint": "^8.46.4",
|
||||
"vite": "^7.2.4"
|
||||
"vite": "^7.2.4",
|
||||
"vitest": "catalog:"
|
||||
}
|
||||
}
|
||||
|
|
|
|||
118
apps/x/apps/renderer/src/hooks/useSessionChat.test.tsx
Normal file
118
apps/x/apps/renderer/src/hooks/useSessionChat.test.tsx
Normal file
|
|
@ -0,0 +1,118 @@
|
|||
import { act, renderHook, waitFor } from '@testing-library/react'
|
||||
import { describe, expect, it } from 'vitest'
|
||||
import type { SessionBusEvent } from '@x/shared/src/sessions.js'
|
||||
import type { SessionsClient } from '@/lib/session-chat/client'
|
||||
import type { SessionFeedListener } from '@/lib/session-chat/feed'
|
||||
import {
|
||||
assistantText,
|
||||
completed,
|
||||
completedTurnLog,
|
||||
created,
|
||||
requested,
|
||||
sessionState,
|
||||
turnCompleted,
|
||||
user,
|
||||
} from '@/lib/session-chat/test-fixtures'
|
||||
import { isChatMessage } from '@/lib/chat-conversation'
|
||||
import { useSessionChat } from './useSessionChat'
|
||||
|
||||
const S1 = 'sess-1'
|
||||
|
||||
function makeDeps() {
|
||||
const calls: Array<{ method: string; args: unknown[] }> = []
|
||||
let emit: SessionFeedListener = () => undefined
|
||||
let unsubscribed = 0
|
||||
const sessions = new Map([[S1, sessionState(S1, ['turn-1'])]])
|
||||
const turns = new Map([['turn-1', completedTurnLog('turn-1', S1, 'q1', 'a1')]])
|
||||
const client: SessionsClient = {
|
||||
create: async () => ({ sessionId: 'x' }),
|
||||
list: async () => ({ sessions: [] }),
|
||||
get: async (sessionId) => {
|
||||
const state = sessions.get(sessionId)
|
||||
if (!state) throw new Error('session not found')
|
||||
return state
|
||||
},
|
||||
getTurn: async (turnId) => ({ turnId, events: turns.get(turnId) ?? [] }),
|
||||
sendMessage: async (...args) => {
|
||||
calls.push({ method: 'sendMessage', args })
|
||||
return { turnId: 'turn-2' }
|
||||
},
|
||||
respondToPermission: async (...args) => {
|
||||
calls.push({ method: 'respondToPermission', args })
|
||||
},
|
||||
respondToAskHuman: async (...args) => {
|
||||
calls.push({ method: 'respondToAskHuman', args })
|
||||
},
|
||||
stopTurn: async (...args) => {
|
||||
calls.push({ method: 'stopTurn', args })
|
||||
},
|
||||
resumeTurn: async () => undefined,
|
||||
setTitle: async () => undefined,
|
||||
delete: async () => undefined,
|
||||
}
|
||||
return {
|
||||
deps: {
|
||||
client,
|
||||
subscribeFeed: (listener: SessionFeedListener) => {
|
||||
emit = listener
|
||||
return () => {
|
||||
unsubscribed += 1
|
||||
}
|
||||
},
|
||||
},
|
||||
calls,
|
||||
emit: (event: SessionBusEvent) => emit(event),
|
||||
getUnsubscribed: () => unsubscribed,
|
||||
}
|
||||
}
|
||||
|
||||
describe('useSessionChat', () => {
|
||||
it('seeds from the session, follows live events, and routes actions', async () => {
|
||||
const { deps, calls, emit } = makeDeps()
|
||||
const { result } = renderHook(() => useSessionChat(S1, deps))
|
||||
|
||||
await waitFor(() => {
|
||||
expect(result.current.latestTurnId).toBe('turn-1')
|
||||
})
|
||||
expect(
|
||||
result.current.chatState?.conversation.filter(isChatMessage).map((m) => m.content),
|
||||
).toEqual(['q1', 'a1'])
|
||||
|
||||
// A new turn streams in over the feed.
|
||||
act(() => {
|
||||
emit({ kind: 'turn-event', sessionId: S1, turnId: 'turn-2', event: created('turn-2', S1, user('q2')) })
|
||||
emit({ kind: 'turn-event', sessionId: S1, turnId: 'turn-2', event: requested('turn-2', 0, [user('q2')]) })
|
||||
emit({
|
||||
kind: 'turn-event',
|
||||
sessionId: S1,
|
||||
turnId: 'turn-2',
|
||||
event: { type: 'text_delta', turnId: 'turn-2', modelCallIndex: 0, delta: 'a2…' },
|
||||
})
|
||||
})
|
||||
expect(result.current.latestTurnId).toBe('turn-2')
|
||||
expect(result.current.chatState?.currentAssistantMessage).toBe('a2…')
|
||||
expect(result.current.chatState?.isProcessing).toBe(true)
|
||||
|
||||
act(() => {
|
||||
emit({ kind: 'turn-event', sessionId: S1, turnId: 'turn-2', event: completed('turn-2', 0, assistantText('a2')) })
|
||||
emit({ kind: 'turn-event', sessionId: S1, turnId: 'turn-2', event: turnCompleted('turn-2', 'a2') })
|
||||
})
|
||||
expect(result.current.chatState?.isProcessing).toBe(false)
|
||||
|
||||
await act(async () => {
|
||||
await result.current.respondToPermission('tc1', 'deny')
|
||||
await result.current.stop()
|
||||
})
|
||||
expect(calls).toEqual([
|
||||
{ method: 'respondToPermission', args: ['turn-2', 'tc1', 'deny', undefined] },
|
||||
{ method: 'stopTurn', args: ['turn-2'] },
|
||||
])
|
||||
})
|
||||
|
||||
it('unsubscribes from the feed on unmount', async () => {
|
||||
const { deps, getUnsubscribed } = makeDeps()
|
||||
const { unmount } = renderHook(() => useSessionChat(S1, deps))
|
||||
unmount()
|
||||
expect(getUnsubscribed()).toBe(1)
|
||||
})
|
||||
})
|
||||
36
apps/x/apps/renderer/src/hooks/useSessionChat.ts
Normal file
36
apps/x/apps/renderer/src/hooks/useSessionChat.ts
Normal file
|
|
@ -0,0 +1,36 @@
|
|||
import { useEffect, useMemo, useState, useSyncExternalStore } from 'react'
|
||||
import { ipcSessionsClient } from '@/lib/session-chat/client'
|
||||
import { subscribeSessionFeed } from '@/lib/session-chat/feed'
|
||||
import { SessionChatStore, type SessionChatStoreDeps } from '@/lib/session-chat/store'
|
||||
|
||||
const defaultDeps: SessionChatStoreDeps = {
|
||||
client: ipcSessionsClient,
|
||||
subscribeFeed: subscribeSessionFeed,
|
||||
}
|
||||
|
||||
// Thin subscription over SessionChatStore — all logic (seeding, feed events,
|
||||
// reducer, overlay, action routing) lives in the store, which is unit-tested
|
||||
// without React. `deps` is injectable for tests.
|
||||
export function useSessionChat(
|
||||
sessionId: string | null,
|
||||
deps: SessionChatStoreDeps = defaultDeps,
|
||||
) {
|
||||
const [store] = useState(() => new SessionChatStore(deps))
|
||||
useEffect(() => () => store.dispose(), [store])
|
||||
useEffect(() => {
|
||||
void store.setSession(sessionId)
|
||||
}, [store, sessionId])
|
||||
const snapshot = useSyncExternalStore(store.subscribe, store.getSnapshot)
|
||||
return useMemo(
|
||||
() => ({
|
||||
...snapshot,
|
||||
sendMessage: store.sendMessage,
|
||||
respondToPermission: store.respondToPermission,
|
||||
answerAskHuman: store.answerAskHuman,
|
||||
stop: store.stop,
|
||||
}),
|
||||
[snapshot, store],
|
||||
)
|
||||
}
|
||||
|
||||
export type SessionChat = ReturnType<typeof useSessionChat>
|
||||
32
apps/x/apps/renderer/src/hooks/useSessions.ts
Normal file
32
apps/x/apps/renderer/src/hooks/useSessions.ts
Normal file
|
|
@ -0,0 +1,32 @@
|
|||
import { useEffect, useMemo, useState, useSyncExternalStore } from 'react'
|
||||
import { ipcSessionsClient } from '@/lib/session-chat/client'
|
||||
import { subscribeSessionFeed } from '@/lib/session-chat/feed'
|
||||
import { SessionListStore, type SessionChatStoreDeps } from '@/lib/session-chat/store'
|
||||
|
||||
const defaultDeps: SessionChatStoreDeps = {
|
||||
client: ipcSessionsClient,
|
||||
subscribeFeed: subscribeSessionFeed,
|
||||
}
|
||||
|
||||
// The session list (chat history sidebar): seeded from sessions:list, kept
|
||||
// current by index-changed feed events. Logic lives in SessionListStore.
|
||||
export function useSessions(deps: SessionChatStoreDeps = defaultDeps) {
|
||||
const [store] = useState(() => new SessionListStore(deps))
|
||||
useEffect(() => {
|
||||
void store.load()
|
||||
return () => store.dispose()
|
||||
}, [store])
|
||||
const snapshot = useSyncExternalStore(store.subscribe, store.getSnapshot)
|
||||
const client = deps.client
|
||||
return useMemo(
|
||||
() => ({
|
||||
...snapshot,
|
||||
createSession: (input: { title?: string } = {}) => client.create(input),
|
||||
deleteSession: (sessionId: string) => client.delete(sessionId),
|
||||
setTitle: (sessionId: string, title: string) => client.setTitle(sessionId, title),
|
||||
}),
|
||||
[snapshot, client],
|
||||
)
|
||||
}
|
||||
|
||||
export type Sessions = ReturnType<typeof useSessions>
|
||||
70
apps/x/apps/renderer/src/lib/session-chat/client.ts
Normal file
70
apps/x/apps/renderer/src/lib/session-chat/client.ts
Normal file
|
|
@ -0,0 +1,70 @@
|
|||
import type { z } from 'zod'
|
||||
import type { UserMessage } from '@x/shared/src/message.js'
|
||||
import type { SessionIndexEntry, SessionState } from '@x/shared/src/sessions.js'
|
||||
import type { JsonValue, RequestedAgent, TurnEvent } from '@x/shared/src/turns.js'
|
||||
|
||||
// Narrow, injectable surface over the sessions IPC channels so stores are
|
||||
// testable with a plain fake instead of a window.ipc stub.
|
||||
export interface SendMessageConfig {
|
||||
agent: z.infer<typeof RequestedAgent>
|
||||
autoPermission?: boolean
|
||||
maxModelCalls?: number
|
||||
}
|
||||
|
||||
export interface SessionsClient {
|
||||
create(input: { title?: string }): Promise<{ sessionId: string }>
|
||||
list(): Promise<{ sessions: SessionIndexEntry[] }>
|
||||
get(sessionId: string): Promise<SessionState>
|
||||
getTurn(turnId: string): Promise<{ turnId: string; events: Array<z.infer<typeof TurnEvent>> }>
|
||||
sendMessage(
|
||||
sessionId: string,
|
||||
input: z.infer<typeof UserMessage>,
|
||||
config: SendMessageConfig,
|
||||
): Promise<{ turnId: string }>
|
||||
respondToPermission(
|
||||
turnId: string,
|
||||
toolCallId: string,
|
||||
decision: 'allow' | 'deny',
|
||||
metadata?: JsonValue,
|
||||
): Promise<void>
|
||||
respondToAskHuman(turnId: string, toolCallId: string, answer: string): Promise<void>
|
||||
stopTurn(turnId: string, reason?: string): Promise<void>
|
||||
resumeTurn(sessionId: string): Promise<void>
|
||||
setTitle(sessionId: string, title: string): Promise<void>
|
||||
delete(sessionId: string): Promise<void>
|
||||
}
|
||||
|
||||
export const ipcSessionsClient: SessionsClient = {
|
||||
create: (input) => window.ipc.invoke('sessions:create', input),
|
||||
list: () => window.ipc.invoke('sessions:list', {}),
|
||||
get: (sessionId) => window.ipc.invoke('sessions:get', { sessionId }),
|
||||
getTurn: (turnId) => window.ipc.invoke('sessions:getTurn', { turnId }),
|
||||
sendMessage: (sessionId, input, config) =>
|
||||
window.ipc.invoke('sessions:sendMessage', { sessionId, input, config }),
|
||||
respondToPermission: async (turnId, toolCallId, decision, metadata) => {
|
||||
await window.ipc.invoke('sessions:respondToPermission', {
|
||||
turnId,
|
||||
toolCallId,
|
||||
decision,
|
||||
...(metadata === undefined ? {} : { metadata }),
|
||||
})
|
||||
},
|
||||
respondToAskHuman: async (turnId, toolCallId, answer) => {
|
||||
await window.ipc.invoke('sessions:respondToAskHuman', { turnId, toolCallId, answer })
|
||||
},
|
||||
stopTurn: async (turnId, reason) => {
|
||||
await window.ipc.invoke('sessions:stopTurn', {
|
||||
turnId,
|
||||
...(reason === undefined ? {} : { reason }),
|
||||
})
|
||||
},
|
||||
resumeTurn: async (sessionId) => {
|
||||
await window.ipc.invoke('sessions:resumeTurn', { sessionId })
|
||||
},
|
||||
setTitle: async (sessionId, title) => {
|
||||
await window.ipc.invoke('sessions:setTitle', { sessionId, title })
|
||||
},
|
||||
delete: async (sessionId) => {
|
||||
await window.ipc.invoke('sessions:delete', { sessionId })
|
||||
},
|
||||
}
|
||||
52
apps/x/apps/renderer/src/lib/session-chat/feed.test.ts
Normal file
52
apps/x/apps/renderer/src/lib/session-chat/feed.test.ts
Normal file
|
|
@ -0,0 +1,52 @@
|
|||
import { describe, expect, it } from 'vitest'
|
||||
import type { SessionBusEvent } from '@x/shared/src/sessions.js'
|
||||
import { createSessionFeed, type SessionFeedListener } from './feed'
|
||||
|
||||
const event: SessionBusEvent = {
|
||||
kind: 'index-changed',
|
||||
sessionId: 's1',
|
||||
entry: null,
|
||||
}
|
||||
|
||||
describe('createSessionFeed', () => {
|
||||
it('starts the source lazily on first subscribe and fans events out', () => {
|
||||
let sourceStarted = 0
|
||||
let push: SessionFeedListener = () => undefined
|
||||
const feed = createSessionFeed((listener) => {
|
||||
sourceStarted += 1
|
||||
push = listener
|
||||
return () => undefined
|
||||
})
|
||||
expect(sourceStarted).toBe(0)
|
||||
|
||||
const seenA: SessionBusEvent[] = []
|
||||
const seenB: SessionBusEvent[] = []
|
||||
feed.subscribe((e) => seenA.push(e))
|
||||
feed.subscribe((e) => seenB.push(e))
|
||||
expect(sourceStarted).toBe(1) // one shared IPC listener
|
||||
|
||||
push(event)
|
||||
expect(seenA).toEqual([event])
|
||||
expect(seenB).toEqual([event])
|
||||
})
|
||||
|
||||
it('unsubscribes cleanly and isolates a throwing listener', () => {
|
||||
let push: SessionFeedListener = () => undefined
|
||||
const feed = createSessionFeed((listener) => {
|
||||
push = listener
|
||||
return () => undefined
|
||||
})
|
||||
const seen: SessionBusEvent[] = []
|
||||
feed.subscribe(() => {
|
||||
throw new Error('bad subscriber')
|
||||
})
|
||||
const unsubscribe = feed.subscribe((e) => seen.push(e))
|
||||
|
||||
push(event)
|
||||
expect(seen).toEqual([event])
|
||||
|
||||
unsubscribe()
|
||||
push(event)
|
||||
expect(seen).toEqual([event])
|
||||
})
|
||||
})
|
||||
42
apps/x/apps/renderer/src/lib/session-chat/feed.ts
Normal file
42
apps/x/apps/renderer/src/lib/session-chat/feed.ts
Normal file
|
|
@ -0,0 +1,42 @@
|
|||
import type { SessionBusEvent } from '@x/shared/src/sessions.js'
|
||||
|
||||
export type SessionFeedListener = (event: SessionBusEvent) => void
|
||||
export type SessionFeedSource = (listener: SessionFeedListener) => () => void
|
||||
|
||||
// One shared consumer of the sessions:events push channel; stores tap this
|
||||
// fan-out instead of each opening their own IPC listener. Factory so tests
|
||||
// can drive a fake source.
|
||||
export function createSessionFeed(source: SessionFeedSource) {
|
||||
const listeners = new Set<SessionFeedListener>()
|
||||
let detach: (() => void) | null = null
|
||||
|
||||
const ensureStarted = () => {
|
||||
if (detach) return
|
||||
detach = source((event) => {
|
||||
// Copy so (un)subscribing during dispatch is safe.
|
||||
for (const listener of [...listeners]) {
|
||||
try {
|
||||
listener(event)
|
||||
} catch {
|
||||
// A misbehaving subscriber must never break the feed.
|
||||
}
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
return {
|
||||
subscribe(listener: SessionFeedListener): () => void {
|
||||
ensureStarted()
|
||||
listeners.add(listener)
|
||||
return () => {
|
||||
listeners.delete(listener)
|
||||
}
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
const appFeed = createSessionFeed((listener) => window.ipc.on('sessions:events', listener))
|
||||
|
||||
export function subscribeSessionFeed(listener: SessionFeedListener): () => void {
|
||||
return appFeed.subscribe(listener)
|
||||
}
|
||||
283
apps/x/apps/renderer/src/lib/session-chat/store.test.ts
Normal file
283
apps/x/apps/renderer/src/lib/session-chat/store.test.ts
Normal file
|
|
@ -0,0 +1,283 @@
|
|||
import { describe, expect, it } from 'vitest'
|
||||
import type { SessionBusEvent, SessionState } from '@x/shared/src/sessions.js'
|
||||
import { isChatMessage } from '@/lib/chat-conversation'
|
||||
import type { SessionsClient } from './client'
|
||||
import type { SessionFeedListener } from './feed'
|
||||
import { SessionChatStore, SessionListStore } from './store'
|
||||
import {
|
||||
assistantText,
|
||||
completed,
|
||||
completedTurnLog,
|
||||
created,
|
||||
indexEntry,
|
||||
requested,
|
||||
sessionState,
|
||||
turnCompleted,
|
||||
user,
|
||||
type TEvent,
|
||||
} from './test-fixtures'
|
||||
|
||||
const S1 = 'sess-1'
|
||||
|
||||
class FakeClient implements SessionsClient {
|
||||
sessions = new Map<string, SessionState>()
|
||||
turns = new Map<string, TEvent[]>()
|
||||
calls: Array<{ method: string; args: unknown[] }> = []
|
||||
// When set, get() defers until the returned resolver is invoked.
|
||||
deferredGet: (() => void) | null = null
|
||||
|
||||
private record(method: string, ...args: unknown[]) {
|
||||
this.calls.push({ method, args })
|
||||
}
|
||||
|
||||
async create(input: { title?: string }) {
|
||||
this.record('create', input)
|
||||
return { sessionId: 'new-session' }
|
||||
}
|
||||
async list() {
|
||||
this.record('list')
|
||||
return { sessions: [...this.sessions.keys()].map((id) => indexEntry(id)) }
|
||||
}
|
||||
async get(sessionId: string) {
|
||||
this.record('get', sessionId)
|
||||
if (this.deferredGet === null) {
|
||||
const state = this.sessions.get(sessionId)
|
||||
if (!state) throw new Error(`session not found: ${sessionId}`)
|
||||
return state
|
||||
}
|
||||
return new Promise<SessionState>((resolve, reject) => {
|
||||
this.deferredGet = () => {
|
||||
const state = this.sessions.get(sessionId)
|
||||
if (state) resolve(state)
|
||||
else reject(new Error(`session not found: ${sessionId}`))
|
||||
}
|
||||
})
|
||||
}
|
||||
async getTurn(turnId: string) {
|
||||
this.record('getTurn', turnId)
|
||||
const events = this.turns.get(turnId)
|
||||
if (!events) throw new Error(`turn not found: ${turnId}`)
|
||||
return { turnId, events }
|
||||
}
|
||||
async sendMessage(sessionId: string, input: unknown, config: unknown) {
|
||||
this.record('sendMessage', sessionId, input, config)
|
||||
return { turnId: 'turn-next' }
|
||||
}
|
||||
async respondToPermission(...args: unknown[]) {
|
||||
this.record('respondToPermission', ...args)
|
||||
}
|
||||
async respondToAskHuman(...args: unknown[]) {
|
||||
this.record('respondToAskHuman', ...args)
|
||||
}
|
||||
async stopTurn(...args: unknown[]) {
|
||||
this.record('stopTurn', ...args)
|
||||
}
|
||||
async resumeTurn(...args: unknown[]) {
|
||||
this.record('resumeTurn', ...args)
|
||||
}
|
||||
async setTitle(...args: unknown[]) {
|
||||
this.record('setTitle', ...args)
|
||||
}
|
||||
async delete(...args: unknown[]) {
|
||||
this.record('delete', ...args)
|
||||
}
|
||||
}
|
||||
|
||||
function makeStore() {
|
||||
const client = new FakeClient()
|
||||
let emit: SessionFeedListener = () => undefined
|
||||
let unsubscribed = 0
|
||||
const store = new SessionChatStore({
|
||||
client,
|
||||
subscribeFeed: (listener) => {
|
||||
emit = listener
|
||||
return () => {
|
||||
unsubscribed += 1
|
||||
}
|
||||
},
|
||||
})
|
||||
return {
|
||||
client,
|
||||
store,
|
||||
emit: (event: SessionBusEvent) => emit(event),
|
||||
getUnsubscribed: () => unsubscribed,
|
||||
}
|
||||
}
|
||||
|
||||
function turnEvent(
|
||||
sessionId: string,
|
||||
turnId: string,
|
||||
event:
|
||||
| TEvent
|
||||
| { type: 'text_delta' | 'reasoning_delta'; turnId: string; modelCallIndex: number; delta: string },
|
||||
): SessionBusEvent {
|
||||
return { kind: 'turn-event', sessionId, turnId, event }
|
||||
}
|
||||
|
||||
describe('SessionChatStore', () => {
|
||||
it('seeds a session: prior turns frozen, latest live, conversation composed', async () => {
|
||||
const { client, store } = makeStore()
|
||||
client.sessions.set(S1, sessionState(S1, ['turn-1', 'turn-2']))
|
||||
client.turns.set('turn-1', completedTurnLog('turn-1', S1, 'first?', 'first answer'))
|
||||
client.turns.set('turn-2', completedTurnLog('turn-2', S1, 'second?', 'second answer'))
|
||||
|
||||
await store.setSession(S1)
|
||||
const snapshot = store.getSnapshot()
|
||||
expect(snapshot.loading).toBe(false)
|
||||
expect(snapshot.latestTurnId).toBe('turn-2')
|
||||
expect(
|
||||
snapshot.chatState?.conversation.filter(isChatMessage).map((m) => m.content),
|
||||
).toEqual(['first?', 'first answer', 'second?', 'second answer'])
|
||||
expect(snapshot.chatState?.isProcessing).toBe(false)
|
||||
})
|
||||
|
||||
it('applies live durable events through the shared reducer', async () => {
|
||||
const { client, store, emit } = makeStore()
|
||||
client.sessions.set(S1, sessionState(S1, []))
|
||||
await store.setSession(S1)
|
||||
|
||||
emit(turnEvent(S1, 'turn-1', created('turn-1', S1, user('go'))))
|
||||
emit(turnEvent(S1, 'turn-1', requested('turn-1', 0, [user('go')])))
|
||||
expect(store.getSnapshot().chatState?.isThinking).toBe(true)
|
||||
|
||||
emit(turnEvent(S1, 'turn-1', completed('turn-1', 0, assistantText('done'))))
|
||||
emit(turnEvent(S1, 'turn-1', turnCompleted('turn-1')))
|
||||
const snapshot = store.getSnapshot()
|
||||
expect(snapshot.latestTurnId).toBe('turn-1')
|
||||
expect(snapshot.chatState?.isProcessing).toBe(false)
|
||||
expect(
|
||||
snapshot.chatState?.conversation.filter(isChatMessage).map((m) => m.content),
|
||||
).toEqual(['go', 'done'])
|
||||
})
|
||||
|
||||
it('accumulates text deltas and clears them on the canonical response', async () => {
|
||||
const { client, store, emit } = makeStore()
|
||||
client.sessions.set(S1, sessionState(S1, []))
|
||||
await store.setSession(S1)
|
||||
emit(turnEvent(S1, 'turn-1', created('turn-1', S1, user('go'))))
|
||||
emit(turnEvent(S1, 'turn-1', requested('turn-1', 0, [user('go')])))
|
||||
emit(turnEvent(S1, 'turn-1', { type: 'text_delta', turnId: 'turn-1', modelCallIndex: 0, delta: 'he' }))
|
||||
emit(turnEvent(S1, 'turn-1', { type: 'text_delta', turnId: 'turn-1', modelCallIndex: 0, delta: 'y' }))
|
||||
expect(store.getSnapshot().chatState?.currentAssistantMessage).toBe('hey')
|
||||
|
||||
emit(turnEvent(S1, 'turn-1', completed('turn-1', 0, assistantText('hey'))))
|
||||
expect(store.getSnapshot().chatState?.currentAssistantMessage).toBe('')
|
||||
})
|
||||
|
||||
it('freezes the previous latest turn when a new turn starts', async () => {
|
||||
const { client, store, emit } = makeStore()
|
||||
client.sessions.set(S1, sessionState(S1, ['turn-1']))
|
||||
client.turns.set('turn-1', completedTurnLog('turn-1', S1, 'q1', 'a1'))
|
||||
await store.setSession(S1)
|
||||
|
||||
emit(turnEvent(S1, 'turn-2', created('turn-2', S1, user('q2'))))
|
||||
const snapshot = store.getSnapshot()
|
||||
expect(snapshot.latestTurnId).toBe('turn-2')
|
||||
expect(
|
||||
snapshot.chatState?.conversation.filter(isChatMessage).map((m) => m.content),
|
||||
).toEqual(['q1', 'a1', 'q2'])
|
||||
})
|
||||
|
||||
it('ignores events for other sessions', async () => {
|
||||
const { client, store, emit } = makeStore()
|
||||
client.sessions.set(S1, sessionState(S1, []))
|
||||
await store.setSession(S1)
|
||||
emit(turnEvent('other-session', 'turn-x', created('turn-x', 'other-session')))
|
||||
expect(store.getSnapshot().chatState?.conversation).toEqual([])
|
||||
})
|
||||
|
||||
it('reconciles an unknown mid-turn event by refetching the turn', async () => {
|
||||
const { client, store, emit } = makeStore()
|
||||
client.sessions.set(S1, sessionState(S1, []))
|
||||
await store.setSession(S1)
|
||||
// The feed attached mid-turn: we never saw turn_created for turn-9.
|
||||
client.turns.set('turn-9', [
|
||||
created('turn-9', S1, user('missed')),
|
||||
requested('turn-9', 0, [user('missed')]),
|
||||
])
|
||||
emit(turnEvent(S1, 'turn-9', completed('turn-9', 0, assistantText('caught up'))))
|
||||
await Promise.resolve()
|
||||
await Promise.resolve()
|
||||
expect(store.getSnapshot().latestTurnId).toBe('turn-9')
|
||||
})
|
||||
|
||||
it('routes actions against the latest turn', async () => {
|
||||
const { client, store, emit } = makeStore()
|
||||
client.sessions.set(S1, sessionState(S1, []))
|
||||
await store.setSession(S1)
|
||||
emit(turnEvent(S1, 'turn-1', created('turn-1', S1)))
|
||||
|
||||
await store.sendMessage(user('next'), { agent: { agentId: 'copilot' } })
|
||||
await store.respondToPermission('tc1', 'allow')
|
||||
await store.answerAskHuman('ah1', '42')
|
||||
await store.stop()
|
||||
|
||||
expect(client.calls.filter((c) => c.method !== 'get' && c.method !== 'getTurn')).toEqual([
|
||||
{ method: 'sendMessage', args: [S1, user('next'), { agent: { agentId: 'copilot' } }] },
|
||||
{ method: 'respondToPermission', args: ['turn-1', 'tc1', 'allow', undefined] },
|
||||
{ method: 'respondToAskHuman', args: ['turn-1', 'ah1', '42'] },
|
||||
{ method: 'stopTurn', args: ['turn-1'] },
|
||||
])
|
||||
})
|
||||
|
||||
it('rejects sendMessage without an active session', async () => {
|
||||
const { store } = makeStore()
|
||||
await expect(
|
||||
store.sendMessage(user('x'), { agent: { agentId: 'copilot' } }),
|
||||
).rejects.toThrowError('No active session')
|
||||
})
|
||||
|
||||
it('drops stale loads after a session switch', async () => {
|
||||
const { client, store } = makeStore()
|
||||
client.sessions.set(S1, sessionState(S1, ['turn-1']))
|
||||
client.turns.set('turn-1', completedTurnLog('turn-1', S1, 'old', 'old answer'))
|
||||
client.sessions.set('sess-2', sessionState('sess-2', []))
|
||||
|
||||
client.deferredGet = () => undefined // arm deferral
|
||||
const first = store.setSession(S1)
|
||||
const release = client.deferredGet
|
||||
client.deferredGet = null
|
||||
const second = store.setSession('sess-2')
|
||||
release?.() // S1's get resolves after the switch
|
||||
await Promise.all([first, second])
|
||||
|
||||
expect(store.getSnapshot().sessionId).toBe('sess-2')
|
||||
expect(store.getSnapshot().chatState?.conversation).toEqual([])
|
||||
})
|
||||
|
||||
it('surfaces load errors and disposes its feed subscription', async () => {
|
||||
const { store, getUnsubscribed } = makeStore()
|
||||
await store.setSession('missing-session')
|
||||
expect(store.getSnapshot().error).toMatch(/session not found/)
|
||||
store.dispose()
|
||||
expect(getUnsubscribed()).toBe(1)
|
||||
})
|
||||
})
|
||||
|
||||
describe('SessionListStore', () => {
|
||||
it('loads, applies index updates and deletions, and sorts by updatedAt', async () => {
|
||||
const client = new FakeClient()
|
||||
client.sessions.set('a', sessionState('a', []))
|
||||
client.sessions.set('b', sessionState('b', []))
|
||||
let emit: SessionFeedListener = () => undefined
|
||||
const store = new SessionListStore({
|
||||
client,
|
||||
subscribeFeed: (listener) => {
|
||||
emit = listener
|
||||
return () => undefined
|
||||
},
|
||||
})
|
||||
await store.load()
|
||||
expect(store.getSnapshot().sessions.map((s) => s.sessionId).sort()).toEqual(['a', 'b'])
|
||||
|
||||
emit({
|
||||
kind: 'index-changed',
|
||||
sessionId: 'c',
|
||||
entry: indexEntry('c', { updatedAt: '2026-07-03T00:00:00Z' }),
|
||||
})
|
||||
expect(store.getSnapshot().sessions[0].sessionId).toBe('c')
|
||||
|
||||
emit({ kind: 'index-changed', sessionId: 'a', entry: null })
|
||||
expect(store.getSnapshot().sessions.map((s) => s.sessionId)).toEqual(['c', 'b'])
|
||||
})
|
||||
})
|
||||
297
apps/x/apps/renderer/src/lib/session-chat/store.ts
Normal file
297
apps/x/apps/renderer/src/lib/session-chat/store.ts
Normal file
|
|
@ -0,0 +1,297 @@
|
|||
import type { z } from 'zod'
|
||||
import type { UserMessage } from '@x/shared/src/message.js'
|
||||
import type { SessionBusEvent, SessionIndexEntry } from '@x/shared/src/sessions.js'
|
||||
import {
|
||||
reduceTurn,
|
||||
type JsonValue,
|
||||
type TurnEvent,
|
||||
type TurnState,
|
||||
type TurnStreamEvent,
|
||||
} from '@x/shared/src/turns.js'
|
||||
import type { SendMessageConfig, SessionsClient } from './client'
|
||||
import type { SessionFeedListener } from './feed'
|
||||
import {
|
||||
applyOverlay,
|
||||
buildSessionChatState,
|
||||
emptyOverlay,
|
||||
type LiveOverlay,
|
||||
type SessionChatState,
|
||||
} from './turn-view'
|
||||
|
||||
type TEvent = z.infer<typeof TurnEvent>
|
||||
|
||||
export interface SessionChatSnapshot {
|
||||
sessionId: string | null
|
||||
chatState: SessionChatState | null
|
||||
latestTurnId: string | null
|
||||
loading: boolean
|
||||
error: string | null
|
||||
}
|
||||
|
||||
export interface SessionChatStoreDeps {
|
||||
client: SessionsClient
|
||||
subscribeFeed: (listener: SessionFeedListener) => () => void
|
||||
}
|
||||
|
||||
// Framework-agnostic controller for one active session's chat. Owns all the
|
||||
// logic (seeding via getSession/getTurn, applying live feed events with the
|
||||
// shared reducer, the ephemeral overlay, action routing); the useSessionChat
|
||||
// hook is a thin useSyncExternalStore subscription over it.
|
||||
export class SessionChatStore {
|
||||
private readonly client: SessionsClient
|
||||
private readonly unsubscribeFeed: () => void
|
||||
private readonly listeners = new Set<() => void>()
|
||||
|
||||
private sessionId: string | null = null
|
||||
// Settled earlier turns, reduced once and frozen.
|
||||
private priorTurns: TurnState[] = []
|
||||
// The latest turn's raw event log; re-reduced on each durable event.
|
||||
private latestEvents: TEvent[] | null = null
|
||||
private overlay: LiveOverlay = emptyOverlay()
|
||||
private loading = false
|
||||
private error: string | null = null
|
||||
// Guards stale async loads after a session switch.
|
||||
private generation = 0
|
||||
|
||||
private snapshot: SessionChatSnapshot = {
|
||||
sessionId: null,
|
||||
chatState: null,
|
||||
latestTurnId: null,
|
||||
loading: false,
|
||||
error: null,
|
||||
}
|
||||
|
||||
constructor(deps: SessionChatStoreDeps) {
|
||||
this.client = deps.client
|
||||
this.unsubscribeFeed = deps.subscribeFeed(this.onFeedEvent)
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
this.unsubscribeFeed()
|
||||
this.listeners.clear()
|
||||
}
|
||||
|
||||
subscribe = (onChange: () => void): (() => void) => {
|
||||
this.listeners.add(onChange)
|
||||
return () => {
|
||||
this.listeners.delete(onChange)
|
||||
}
|
||||
}
|
||||
|
||||
getSnapshot = (): SessionChatSnapshot => this.snapshot
|
||||
|
||||
async setSession(sessionId: string | null): Promise<void> {
|
||||
if (sessionId === this.sessionId) return
|
||||
this.generation += 1
|
||||
const generation = this.generation
|
||||
this.sessionId = sessionId
|
||||
this.priorTurns = []
|
||||
this.latestEvents = null
|
||||
this.overlay = emptyOverlay()
|
||||
this.error = null
|
||||
this.loading = sessionId !== null
|
||||
this.emit()
|
||||
if (sessionId === null) return
|
||||
|
||||
try {
|
||||
const state = await this.client.get(sessionId)
|
||||
const turns = await Promise.all(
|
||||
state.turns.map((ref) => this.client.getTurn(ref.turnId)),
|
||||
)
|
||||
if (generation !== this.generation) return
|
||||
const reduced = turns.map((turn) => reduceTurn(turn.events))
|
||||
this.priorTurns = reduced.slice(0, -1)
|
||||
this.latestEvents = turns.length > 0 ? turns[turns.length - 1].events : null
|
||||
this.loading = false
|
||||
this.emit()
|
||||
} catch (error) {
|
||||
if (generation !== this.generation) return
|
||||
this.loading = false
|
||||
this.error = error instanceof Error ? error.message : String(error)
|
||||
this.emit()
|
||||
}
|
||||
}
|
||||
|
||||
private onFeedEvent: SessionFeedListener = (event: SessionBusEvent) => {
|
||||
if (event.kind !== 'turn-event' || event.sessionId !== this.sessionId) return
|
||||
const turnEvent = event.event
|
||||
if (isDurable(turnEvent)) {
|
||||
if (turnEvent.type === 'turn_created') {
|
||||
// A new turn started for this session: freeze the previous latest.
|
||||
this.freezeLatest()
|
||||
this.latestEvents = [turnEvent]
|
||||
this.overlay = emptyOverlay()
|
||||
} else if (this.latestEvents && this.latestEvents[0].turnId === turnEvent.turnId) {
|
||||
this.latestEvents.push(turnEvent)
|
||||
} else {
|
||||
// An event for a turn we haven't seen (missed turn_created, e.g. the
|
||||
// feed attached mid-turn): reconcile by refetching that turn.
|
||||
void this.reloadTurn(event.turnId)
|
||||
return
|
||||
}
|
||||
}
|
||||
this.overlay = applyOverlay(this.overlay, turnEvent)
|
||||
this.emit()
|
||||
}
|
||||
|
||||
private freezeLatest(): void {
|
||||
if (!this.latestEvents) return
|
||||
try {
|
||||
this.priorTurns = [...this.priorTurns, reduceTurn(this.latestEvents)]
|
||||
} catch {
|
||||
// A turn we can't reduce is dropped from history rather than wedging
|
||||
// the whole conversation.
|
||||
}
|
||||
this.latestEvents = null
|
||||
}
|
||||
|
||||
private async reloadTurn(turnId: string): Promise<void> {
|
||||
const generation = this.generation
|
||||
try {
|
||||
const turn = await this.client.getTurn(turnId)
|
||||
if (generation !== this.generation) return
|
||||
if (this.latestEvents && this.latestEvents[0].turnId !== turnId) {
|
||||
this.freezeLatest()
|
||||
}
|
||||
this.latestEvents = turn.events
|
||||
this.emit()
|
||||
} catch {
|
||||
// The next snapshot-worthy event will retry.
|
||||
}
|
||||
}
|
||||
|
||||
// ── Actions ─────────────────────────────────────────────────────────────
|
||||
|
||||
sendMessage = async (
|
||||
input: z.infer<typeof UserMessage>,
|
||||
config: SendMessageConfig,
|
||||
): Promise<{ turnId: string }> => {
|
||||
if (!this.sessionId) throw new Error('No active session')
|
||||
return this.client.sendMessage(this.sessionId, input, config)
|
||||
}
|
||||
|
||||
respondToPermission = async (
|
||||
toolCallId: string,
|
||||
decision: 'allow' | 'deny',
|
||||
metadata?: JsonValue,
|
||||
): Promise<void> => {
|
||||
const turnId = this.snapshot.latestTurnId
|
||||
if (!turnId) return
|
||||
await this.client.respondToPermission(turnId, toolCallId, decision, metadata)
|
||||
}
|
||||
|
||||
answerAskHuman = async (toolCallId: string, answer: string): Promise<void> => {
|
||||
const turnId = this.snapshot.latestTurnId
|
||||
if (!turnId) return
|
||||
await this.client.respondToAskHuman(turnId, toolCallId, answer)
|
||||
}
|
||||
|
||||
stop = async (): Promise<void> => {
|
||||
const turnId = this.snapshot.latestTurnId
|
||||
if (!turnId) return
|
||||
await this.client.stopTurn(turnId)
|
||||
}
|
||||
|
||||
// ── Derivation ──────────────────────────────────────────────────────────
|
||||
|
||||
private emit(): void {
|
||||
this.snapshot = this.derive()
|
||||
for (const listener of [...this.listeners]) {
|
||||
listener()
|
||||
}
|
||||
}
|
||||
|
||||
private derive(): SessionChatSnapshot {
|
||||
let turns = this.priorTurns
|
||||
let error = this.error
|
||||
if (this.latestEvents) {
|
||||
try {
|
||||
turns = [...this.priorTurns, reduceTurn(this.latestEvents)]
|
||||
} catch (reduceError) {
|
||||
error =
|
||||
reduceError instanceof Error ? reduceError.message : String(reduceError)
|
||||
}
|
||||
}
|
||||
const latest = turns[turns.length - 1]
|
||||
return {
|
||||
sessionId: this.sessionId,
|
||||
chatState:
|
||||
this.sessionId !== null && !this.loading
|
||||
? buildSessionChatState(turns, this.overlay)
|
||||
: null,
|
||||
latestTurnId: latest?.definition.turnId ?? null,
|
||||
loading: this.loading,
|
||||
error,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
function isDurable(event: TurnStreamEvent): event is TEvent {
|
||||
return event.type !== 'text_delta' && event.type !== 'reasoning_delta'
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Session list store
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export interface SessionListSnapshot {
|
||||
sessions: SessionIndexEntry[]
|
||||
loading: boolean
|
||||
}
|
||||
|
||||
export class SessionListStore {
|
||||
private readonly client: SessionsClient
|
||||
private readonly unsubscribeFeed: () => void
|
||||
private readonly listeners = new Set<() => void>()
|
||||
private entries = new Map<string, SessionIndexEntry>()
|
||||
private loading = true
|
||||
private snapshot: SessionListSnapshot = { sessions: [], loading: true }
|
||||
|
||||
constructor(deps: SessionChatStoreDeps) {
|
||||
this.client = deps.client
|
||||
this.unsubscribeFeed = deps.subscribeFeed(this.onFeedEvent)
|
||||
}
|
||||
|
||||
dispose(): void {
|
||||
this.unsubscribeFeed()
|
||||
this.listeners.clear()
|
||||
}
|
||||
|
||||
subscribe = (onChange: () => void): (() => void) => {
|
||||
this.listeners.add(onChange)
|
||||
return () => {
|
||||
this.listeners.delete(onChange)
|
||||
}
|
||||
}
|
||||
|
||||
getSnapshot = (): SessionListSnapshot => this.snapshot
|
||||
|
||||
async load(): Promise<void> {
|
||||
const { sessions } = await this.client.list()
|
||||
this.entries = new Map(sessions.map((entry) => [entry.sessionId, entry]))
|
||||
this.loading = false
|
||||
this.emit()
|
||||
}
|
||||
|
||||
private onFeedEvent: SessionFeedListener = (event: SessionBusEvent) => {
|
||||
if (event.kind !== 'index-changed') return
|
||||
if (event.entry === null) {
|
||||
this.entries.delete(event.sessionId)
|
||||
} else {
|
||||
this.entries.set(event.sessionId, event.entry)
|
||||
}
|
||||
this.emit()
|
||||
}
|
||||
|
||||
private emit(): void {
|
||||
this.snapshot = {
|
||||
sessions: [...this.entries.values()].sort((a, b) =>
|
||||
b.updatedAt.localeCompare(a.updatedAt),
|
||||
),
|
||||
loading: this.loading,
|
||||
}
|
||||
for (const listener of [...this.listeners]) {
|
||||
listener()
|
||||
}
|
||||
}
|
||||
}
|
||||
169
apps/x/apps/renderer/src/lib/session-chat/test-fixtures.ts
Normal file
169
apps/x/apps/renderer/src/lib/session-chat/test-fixtures.ts
Normal file
|
|
@ -0,0 +1,169 @@
|
|||
import type { z } from 'zod'
|
||||
import type { SessionIndexEntry, SessionState } from '@x/shared/src/sessions.js'
|
||||
import type { ResolvedAgent, TurnEvent } from '@x/shared/src/turns.js'
|
||||
|
||||
// Compact turn-event builders for renderer tests. Sequences must satisfy the
|
||||
// shared reducer's invariants (reduceTurn is what the store runs on them).
|
||||
|
||||
export type TEvent = z.infer<typeof TurnEvent>
|
||||
|
||||
export const TS = '2026-07-02T10:00:00Z'
|
||||
|
||||
export const FIXTURE_AGENT: z.infer<typeof ResolvedAgent> = {
|
||||
agentId: 'copilot',
|
||||
systemPrompt: 'SYS',
|
||||
model: { provider: 'openai', model: 'gpt-fixture' },
|
||||
tools: [],
|
||||
}
|
||||
|
||||
export function user(text: string) {
|
||||
return { role: 'user' as const, content: text }
|
||||
}
|
||||
|
||||
export function assistantText(text: string) {
|
||||
return { role: 'assistant' as const, content: text }
|
||||
}
|
||||
|
||||
export function toolCallPart(id: string, name: string, args: unknown = {}) {
|
||||
return { type: 'tool-call' as const, toolCallId: id, toolName: name, arguments: args }
|
||||
}
|
||||
|
||||
export function created(
|
||||
turnId: string,
|
||||
sessionId: string,
|
||||
input: ReturnType<typeof user> = user('hello'),
|
||||
): TEvent {
|
||||
return {
|
||||
type: 'turn_created',
|
||||
schemaVersion: 1,
|
||||
turnId,
|
||||
ts: TS,
|
||||
sessionId,
|
||||
agent: { requested: { agentId: 'copilot' }, resolved: FIXTURE_AGENT },
|
||||
context: [],
|
||||
input,
|
||||
config: { autoPermission: false, humanAvailable: true, maxModelCalls: 20 },
|
||||
}
|
||||
}
|
||||
|
||||
export function requested(turnId: string, index: number, messages: unknown[]): TEvent {
|
||||
return {
|
||||
type: 'model_call_requested',
|
||||
turnId,
|
||||
ts: TS,
|
||||
modelCallIndex: index,
|
||||
request: {
|
||||
systemPrompt: 'SYS',
|
||||
messages: messages as never,
|
||||
tools: [],
|
||||
parameters: {},
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
export function completed(
|
||||
turnId: string,
|
||||
index: number,
|
||||
message: { role: 'assistant'; content: unknown },
|
||||
): TEvent {
|
||||
return {
|
||||
type: 'model_call_completed',
|
||||
turnId,
|
||||
ts: TS,
|
||||
modelCallIndex: index,
|
||||
message: message as never,
|
||||
finishReason: 'stop',
|
||||
usage: {},
|
||||
}
|
||||
}
|
||||
|
||||
export function invocation(turnId: string, toolCallId: string, name: string): TEvent {
|
||||
return {
|
||||
type: 'tool_invocation_requested',
|
||||
turnId,
|
||||
ts: TS,
|
||||
toolCallId,
|
||||
toolId: `builtin:${name}`,
|
||||
toolName: name,
|
||||
execution: 'sync',
|
||||
input: {},
|
||||
}
|
||||
}
|
||||
|
||||
export function toolResult(
|
||||
turnId: string,
|
||||
toolCallId: string,
|
||||
name: string,
|
||||
output: unknown = 'ok',
|
||||
): TEvent {
|
||||
return {
|
||||
type: 'tool_result',
|
||||
turnId,
|
||||
ts: TS,
|
||||
toolCallId,
|
||||
toolName: name,
|
||||
source: 'sync',
|
||||
result: { output: output as never, isError: false },
|
||||
}
|
||||
}
|
||||
|
||||
export function turnCompleted(turnId: string, text = 'done'): TEvent {
|
||||
return {
|
||||
type: 'turn_completed',
|
||||
turnId,
|
||||
ts: TS,
|
||||
output: assistantText(text),
|
||||
finishReason: 'stop',
|
||||
usage: {},
|
||||
}
|
||||
}
|
||||
|
||||
// A settled single-response turn.
|
||||
export function completedTurnLog(
|
||||
turnId: string,
|
||||
sessionId: string,
|
||||
question: string,
|
||||
answer: string,
|
||||
): TEvent[] {
|
||||
return [
|
||||
created(turnId, sessionId, user(question)),
|
||||
requested(turnId, 0, [user(question)]),
|
||||
completed(turnId, 0, assistantText(answer)),
|
||||
turnCompleted(turnId, answer),
|
||||
]
|
||||
}
|
||||
|
||||
export function indexEntry(
|
||||
sessionId: string,
|
||||
overrides: Partial<SessionIndexEntry> = {},
|
||||
): SessionIndexEntry {
|
||||
return {
|
||||
sessionId,
|
||||
title: `Session ${sessionId}`,
|
||||
createdAt: TS,
|
||||
updatedAt: TS,
|
||||
turnCount: 1,
|
||||
latestTurnStatus: 'completed',
|
||||
...overrides,
|
||||
}
|
||||
}
|
||||
|
||||
export function sessionState(
|
||||
sessionId: string,
|
||||
turnIds: string[],
|
||||
): SessionState {
|
||||
return {
|
||||
definition: { type: 'session_created', schemaVersion: 1, sessionId, ts: TS },
|
||||
title: `Session ${sessionId}`,
|
||||
turns: turnIds.map((turnId, i) => ({
|
||||
turnId,
|
||||
sessionSeq: i + 1,
|
||||
agentId: 'copilot',
|
||||
model: FIXTURE_AGENT.model,
|
||||
ts: TS,
|
||||
})),
|
||||
latestTurnId: turnIds[turnIds.length - 1],
|
||||
createdAt: TS,
|
||||
updatedAt: TS,
|
||||
}
|
||||
}
|
||||
272
apps/x/apps/renderer/src/lib/session-chat/turn-view.test.ts
Normal file
272
apps/x/apps/renderer/src/lib/session-chat/turn-view.test.ts
Normal file
|
|
@ -0,0 +1,272 @@
|
|||
import { describe, expect, it } from 'vitest'
|
||||
import { reduceTurn } from '@x/shared/src/turns.js'
|
||||
import { isChatMessage, isErrorMessage, isToolCall } from '@/lib/chat-conversation'
|
||||
import {
|
||||
applyOverlay,
|
||||
buildSessionChatState,
|
||||
buildTurnConversation,
|
||||
emptyOverlay,
|
||||
} from './turn-view'
|
||||
import {
|
||||
TS,
|
||||
assistantText,
|
||||
completed,
|
||||
completedTurnLog,
|
||||
created,
|
||||
invocation,
|
||||
requested,
|
||||
toolCallPart,
|
||||
toolResult,
|
||||
turnCompleted,
|
||||
user,
|
||||
type TEvent,
|
||||
} from './test-fixtures'
|
||||
|
||||
const T1 = 'turn-1'
|
||||
const S1 = 'sess-1'
|
||||
|
||||
function assistantCalls(...parts: Array<ReturnType<typeof toolCallPart>>) {
|
||||
return { role: 'assistant' as const, content: parts }
|
||||
}
|
||||
|
||||
describe('applyOverlay', () => {
|
||||
it('accumulates text and reasoning deltas', () => {
|
||||
let overlay = emptyOverlay()
|
||||
overlay = applyOverlay(overlay, { type: 'text_delta', turnId: T1, modelCallIndex: 0, delta: 'he' })
|
||||
overlay = applyOverlay(overlay, { type: 'text_delta', turnId: T1, modelCallIndex: 0, delta: 'y' })
|
||||
overlay = applyOverlay(overlay, {
|
||||
type: 'reasoning_delta',
|
||||
turnId: T1,
|
||||
modelCallIndex: 0,
|
||||
delta: 'hmm',
|
||||
})
|
||||
expect(overlay.text).toBe('hey')
|
||||
expect(overlay.reasoning).toBe('hmm')
|
||||
})
|
||||
|
||||
it('clears text/reasoning when the canonical model response arrives', () => {
|
||||
let overlay = { ...emptyOverlay(), text: 'streaming', reasoning: 'thinking' }
|
||||
overlay = applyOverlay(overlay, completed(T1, 0, assistantText('final')))
|
||||
expect(overlay.text).toBe('')
|
||||
expect(overlay.reasoning).toBe('')
|
||||
})
|
||||
|
||||
it('accumulates tool-output progress chunks and drops them on the terminal result', () => {
|
||||
let overlay = emptyOverlay()
|
||||
const progress = (chunk: string): TEvent => ({
|
||||
type: 'tool_progress',
|
||||
turnId: T1,
|
||||
ts: TS,
|
||||
toolCallId: 'tc1',
|
||||
source: 'sync',
|
||||
progress: { kind: 'tool-output', chunk },
|
||||
})
|
||||
overlay = applyOverlay(overlay, progress('$ ls\n'))
|
||||
overlay = applyOverlay(overlay, progress('README.md\n'))
|
||||
expect(overlay.toolOutput.tc1).toBe('$ ls\nREADME.md\n')
|
||||
|
||||
overlay = applyOverlay(overlay, toolResult(T1, 'tc1', 'executeCommand'))
|
||||
expect(overlay.toolOutput.tc1).toBeUndefined()
|
||||
})
|
||||
|
||||
it('ignores non-output progress payloads', () => {
|
||||
const overlay = applyOverlay(emptyOverlay(), {
|
||||
type: 'tool_progress',
|
||||
turnId: T1,
|
||||
ts: TS,
|
||||
toolCallId: 'tc1',
|
||||
source: 'sync',
|
||||
progress: { pct: 50 },
|
||||
})
|
||||
expect(overlay.toolOutput).toEqual({})
|
||||
})
|
||||
})
|
||||
|
||||
describe('buildTurnConversation', () => {
|
||||
it('maps user input, assistant text, and settled tool calls', () => {
|
||||
const call = assistantCalls(toolCallPart('tc1', 'echo', { x: 1 }))
|
||||
const state = reduceTurn([
|
||||
created(T1, S1, user('run it')),
|
||||
requested(T1, 0, [user('run it')]),
|
||||
completed(T1, 0, call),
|
||||
invocation(T1, 'tc1', 'echo'),
|
||||
toolResult(T1, 'tc1', 'echo', { echoed: true }),
|
||||
requested(T1, 1, [
|
||||
user('run it'),
|
||||
call,
|
||||
{ role: 'tool', content: '{"echoed":true}', toolCallId: 'tc1', toolName: 'echo' },
|
||||
]),
|
||||
completed(T1, 1, assistantText('all done')),
|
||||
turnCompleted(T1, 'all done'),
|
||||
])
|
||||
const items = buildTurnConversation(state)
|
||||
expect(items.map((i) => (isToolCall(i) ? `tool:${i.name}:${i.status}` : isChatMessage(i) ? `${i.role}` : 'x'))).toEqual([
|
||||
'user',
|
||||
'tool:echo:completed',
|
||||
'assistant',
|
||||
])
|
||||
const tool = items.find(isToolCall)
|
||||
expect(tool?.result).toEqual({ echoed: true })
|
||||
expect(tool?.input).toEqual({ x: 1 })
|
||||
})
|
||||
|
||||
it('marks permission-pending tools pending and running tools running', () => {
|
||||
const state = reduceTurn([
|
||||
created(T1, S1),
|
||||
requested(T1, 0, [user('hello')]),
|
||||
completed(T1, 0, assistantCalls(toolCallPart('p1', 'executeCommand'), toolCallPart('r1', 'echo'))),
|
||||
{
|
||||
type: 'tool_permission_required',
|
||||
turnId: T1,
|
||||
ts: TS,
|
||||
toolCallId: 'p1',
|
||||
toolName: 'executeCommand',
|
||||
request: { kind: 'command', commandNames: ['rm'] },
|
||||
},
|
||||
invocation(T1, 'r1', 'echo'),
|
||||
])
|
||||
const items = buildTurnConversation(state).filter(isToolCall)
|
||||
expect(items.map((i) => i.status)).toEqual(['pending', 'running'])
|
||||
})
|
||||
|
||||
it('renders user attachments and a failed turn as an error item', () => {
|
||||
const input = {
|
||||
role: 'user' as const,
|
||||
content: [
|
||||
{ type: 'text' as const, text: 'see file' },
|
||||
{ type: 'attachment' as const, path: '/tmp/a.png', filename: 'a.png', mimeType: 'image/png' },
|
||||
],
|
||||
}
|
||||
const state = reduceTurn([
|
||||
created(T1, S1, input as never),
|
||||
requested(T1, 0, [input]),
|
||||
{ type: 'model_call_failed', turnId: T1, ts: TS, modelCallIndex: 0, error: 'boom' },
|
||||
{ type: 'turn_failed', turnId: T1, ts: TS, error: 'boom', usage: {} },
|
||||
])
|
||||
const items = buildTurnConversation(state)
|
||||
expect(isChatMessage(items[0]) && items[0].attachments?.[0].filename).toBe('a.png')
|
||||
expect(isErrorMessage(items[1]) && items[1].message).toBe('boom')
|
||||
})
|
||||
})
|
||||
|
||||
describe('buildSessionChatState', () => {
|
||||
it('composes the conversation across turns and derives processing flags', () => {
|
||||
const prior = reduceTurn(completedTurnLog('turn-1', S1, 'first?', 'first answer'))
|
||||
const latest = reduceTurn([
|
||||
created('turn-2', S1, user('second?')),
|
||||
requested('turn-2', 0, [user('second?')]),
|
||||
])
|
||||
const state = buildSessionChatState([prior, latest], emptyOverlay())
|
||||
expect(
|
||||
state.conversation.filter(isChatMessage).map((m) => m.content),
|
||||
).toEqual(['first?', 'first answer', 'second?'])
|
||||
expect(state.isProcessing).toBe(true) // latest turn idle = actively working
|
||||
expect(state.isThinking).toBe(true)
|
||||
})
|
||||
|
||||
it('is settled (not processing) when the latest turn is terminal', () => {
|
||||
const turn = reduceTurn(completedTurnLog(T1, S1, 'q', 'a'))
|
||||
const state = buildSessionChatState([turn], emptyOverlay())
|
||||
expect(state.isProcessing).toBe(false)
|
||||
expect(state.isThinking).toBe(false)
|
||||
})
|
||||
|
||||
it('exposes pending permissions as request events; waiting is not thinking', () => {
|
||||
const turn = reduceTurn([
|
||||
created(T1, S1),
|
||||
requested(T1, 0, [user('hello')]),
|
||||
completed(T1, 0, assistantCalls(toolCallPart('p1', 'executeCommand', { command: 'rm -rf /' }))),
|
||||
{
|
||||
type: 'tool_permission_required',
|
||||
turnId: T1,
|
||||
ts: TS,
|
||||
toolCallId: 'p1',
|
||||
toolName: 'executeCommand',
|
||||
request: { kind: 'command', commandNames: ['rm'] },
|
||||
},
|
||||
{
|
||||
type: 'turn_suspended',
|
||||
turnId: T1,
|
||||
ts: TS,
|
||||
pendingPermissions: [
|
||||
{ toolCallId: 'p1', toolName: 'executeCommand', request: { kind: 'command', commandNames: ['rm'] } },
|
||||
],
|
||||
pendingAsyncTools: [],
|
||||
usage: {},
|
||||
},
|
||||
])
|
||||
const state = buildSessionChatState([turn], emptyOverlay())
|
||||
const request = state.allPermissionRequests.get('p1')
|
||||
expect(request).toMatchObject({
|
||||
type: 'tool-permission-request',
|
||||
toolCall: { toolCallId: 'p1', toolName: 'executeCommand' },
|
||||
permission: { kind: 'command', commandNames: ['rm'] },
|
||||
})
|
||||
expect(state.isProcessing).toBe(true)
|
||||
expect(state.isThinking).toBe(false)
|
||||
})
|
||||
|
||||
it('maps human decisions to responses and classifier decisions to auto-decisions', () => {
|
||||
const turn = reduceTurn([
|
||||
created(T1, S1),
|
||||
requested(T1, 0, [user('hello')]),
|
||||
completed(T1, 0, assistantCalls(toolCallPart('h1', 'echo'), toolCallPart('c1', 'echo'))),
|
||||
{ type: 'tool_permission_required', turnId: T1, ts: TS, toolCallId: 'h1', toolName: 'echo', request: { kind: 'command', commandNames: ['x'] } },
|
||||
{ type: 'tool_permission_required', turnId: T1, ts: TS, toolCallId: 'c1', toolName: 'echo', request: { kind: 'command', commandNames: ['y'] } },
|
||||
{ type: 'tool_permission_resolved', turnId: T1, ts: TS, toolCallId: 'h1', decision: 'allow', source: 'human' },
|
||||
{ type: 'tool_permission_resolved', turnId: T1, ts: TS, toolCallId: 'c1', decision: 'deny', source: 'classifier', reason: 'risky' },
|
||||
{ type: 'tool_result', turnId: T1, ts: TS, toolCallId: 'c1', toolName: 'echo', source: 'runtime', result: { output: 'denied', isError: true } },
|
||||
invocation(T1, 'h1', 'echo'),
|
||||
toolResult(T1, 'h1', 'echo'),
|
||||
])
|
||||
const state = buildSessionChatState([turn], emptyOverlay())
|
||||
expect(state.permissionResponses.get('h1')).toBe('approve')
|
||||
expect(state.autoPermissionDecisions.get('c1')).toMatchObject({
|
||||
decision: 'deny',
|
||||
reason: 'risky',
|
||||
})
|
||||
})
|
||||
|
||||
it('exposes pending ask-human calls with question and options', () => {
|
||||
const turn = reduceTurn([
|
||||
created(T1, S1),
|
||||
requested(T1, 0, [user('hello')]),
|
||||
completed(T1, 0, assistantCalls(toolCallPart('ah1', 'ask-human', { question: 'Deploy?', options: ['Yes', 'No'] }))),
|
||||
{
|
||||
type: 'tool_invocation_requested',
|
||||
turnId: T1,
|
||||
ts: TS,
|
||||
toolCallId: 'ah1',
|
||||
toolId: 'builtin:ask-human',
|
||||
toolName: 'ask-human',
|
||||
execution: 'async',
|
||||
input: { question: 'Deploy?', options: ['Yes', 'No'] },
|
||||
},
|
||||
])
|
||||
const state = buildSessionChatState([turn], emptyOverlay())
|
||||
expect(state.pendingAskHumanRequests.get('ah1')).toMatchObject({
|
||||
query: 'Deploy?',
|
||||
options: ['Yes', 'No'],
|
||||
})
|
||||
expect(state.isThinking).toBe(false) // waiting on the human, no shimmer
|
||||
})
|
||||
|
||||
it('stitches live tool output onto the matching tool item', () => {
|
||||
const turn = reduceTurn([
|
||||
created(T1, S1),
|
||||
requested(T1, 0, [user('hello')]),
|
||||
completed(T1, 0, assistantCalls(toolCallPart('tc1', 'executeCommand'))),
|
||||
invocation(T1, 'tc1', 'executeCommand'),
|
||||
])
|
||||
const overlay = { ...emptyOverlay(), toolOutput: { tc1: 'partial output' } }
|
||||
const state = buildSessionChatState([turn], overlay)
|
||||
const tool = state.conversation.filter(isToolCall)[0]
|
||||
expect(tool.streamingOutput).toBe('partial output')
|
||||
})
|
||||
|
||||
it('surfaces streaming text as currentAssistantMessage', () => {
|
||||
const turn = reduceTurn([created(T1, S1), requested(T1, 0, [user('hello')])])
|
||||
const state = buildSessionChatState([turn], { ...emptyOverlay(), text: 'typing…' })
|
||||
expect(state.currentAssistantMessage).toBe('typing…')
|
||||
})
|
||||
})
|
||||
300
apps/x/apps/renderer/src/lib/session-chat/turn-view.ts
Normal file
300
apps/x/apps/renderer/src/lib/session-chat/turn-view.ts
Normal file
|
|
@ -0,0 +1,300 @@
|
|||
import type { z } from 'zod'
|
||||
import type {
|
||||
AskHumanRequestEvent,
|
||||
ToolPermissionAutoDecisionEvent,
|
||||
ToolPermissionMetadata,
|
||||
ToolPermissionRequestEvent,
|
||||
} from '@x/shared/src/runs.js'
|
||||
import {
|
||||
deriveTurnStatus,
|
||||
outstandingAsyncTools,
|
||||
outstandingPermissions,
|
||||
type ToolCallState,
|
||||
type TurnState,
|
||||
type TurnStreamEvent,
|
||||
} from '@x/shared/src/turns.js'
|
||||
import type {
|
||||
ChatMessage,
|
||||
ConversationItem,
|
||||
ErrorMessage,
|
||||
MessageAttachment,
|
||||
PermissionResponse,
|
||||
ToolCall,
|
||||
} from '@/lib/chat-conversation'
|
||||
|
||||
// Pure derivations from reduced turn state (+ the ephemeral live overlay) to
|
||||
// the view shapes the existing chat components already consume. No IPC, no
|
||||
// React — everything here is unit-testable with plain state values.
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Live overlay: ephemeral streaming buffers (turn-runtime-design.md §14.4).
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
export type LiveOverlay = {
|
||||
text: string
|
||||
reasoning: string
|
||||
toolOutput: Record<string, string>
|
||||
}
|
||||
|
||||
export const emptyOverlay = (): LiveOverlay => ({ text: '', reasoning: '', toolOutput: {} })
|
||||
|
||||
// Accumulates deltas; canonical durable events supersede the buffers (the
|
||||
// committed transcript now contains what was streaming).
|
||||
export function applyOverlay(overlay: LiveOverlay, event: TurnStreamEvent): LiveOverlay {
|
||||
switch (event.type) {
|
||||
case 'text_delta':
|
||||
return { ...overlay, text: overlay.text + event.delta }
|
||||
case 'reasoning_delta':
|
||||
return { ...overlay, reasoning: overlay.reasoning + event.delta }
|
||||
case 'model_call_completed':
|
||||
return { ...overlay, text: '', reasoning: '' }
|
||||
case 'tool_progress': {
|
||||
const progress = event.progress
|
||||
if (
|
||||
progress &&
|
||||
typeof progress === 'object' &&
|
||||
!Array.isArray(progress) &&
|
||||
(progress as { kind?: unknown }).kind === 'tool-output' &&
|
||||
typeof (progress as { chunk?: unknown }).chunk === 'string'
|
||||
) {
|
||||
const chunk = (progress as { chunk: string }).chunk
|
||||
return {
|
||||
...overlay,
|
||||
toolOutput: {
|
||||
...overlay.toolOutput,
|
||||
[event.toolCallId]: (overlay.toolOutput[event.toolCallId] ?? '') + chunk,
|
||||
},
|
||||
}
|
||||
}
|
||||
return overlay
|
||||
}
|
||||
case 'tool_result': {
|
||||
if (!(event.toolCallId in overlay.toolOutput)) return overlay
|
||||
const toolOutput = { ...overlay.toolOutput }
|
||||
delete toolOutput[event.toolCallId]
|
||||
return { ...overlay, toolOutput }
|
||||
}
|
||||
default:
|
||||
return overlay
|
||||
}
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Conversation items
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
type UserContent = TurnState['definition']['input']['content']
|
||||
|
||||
function extractText(content: UserContent): string {
|
||||
if (typeof content === 'string') return content
|
||||
return content
|
||||
.map((part) => (part.type === 'text' ? part.text : ''))
|
||||
.filter(Boolean)
|
||||
.join('\n')
|
||||
}
|
||||
|
||||
function extractAttachments(content: UserContent): MessageAttachment[] | undefined {
|
||||
if (typeof content === 'string') return undefined
|
||||
const attachments = content.flatMap((part) =>
|
||||
part.type === 'attachment'
|
||||
? [
|
||||
{
|
||||
path: part.path,
|
||||
filename: part.filename,
|
||||
mimeType: part.mimeType,
|
||||
...(part.size === undefined ? {} : { size: part.size }),
|
||||
},
|
||||
]
|
||||
: [],
|
||||
)
|
||||
return attachments.length > 0 ? attachments : undefined
|
||||
}
|
||||
|
||||
function toolStatus(tc: ToolCallState): ToolCall['status'] {
|
||||
if (tc.result) return 'completed'
|
||||
if (tc.permission && !tc.permission.resolved) return 'pending'
|
||||
return 'running'
|
||||
}
|
||||
|
||||
// One turn's contribution to the conversation: the user input, then per
|
||||
// completed model call its text and tool calls (with live status/results).
|
||||
export function buildTurnConversation(state: TurnState): ConversationItem[] {
|
||||
const items: ConversationItem[] = []
|
||||
const turnId = state.definition.turnId
|
||||
let seq = 0
|
||||
const ts = () => Date.parse(state.definition.ts) + seq++
|
||||
|
||||
const userText = extractText(state.definition.input.content)
|
||||
const attachments = extractAttachments(state.definition.input.content)
|
||||
if (userText || attachments) {
|
||||
items.push({
|
||||
id: `${turnId}:user`,
|
||||
role: 'user',
|
||||
content: userText,
|
||||
timestamp: ts(),
|
||||
...(attachments ? { attachments } : {}),
|
||||
} satisfies ChatMessage)
|
||||
}
|
||||
|
||||
const toolCallsById = new Map(state.toolCalls.map((tc) => [tc.toolCallId, tc]))
|
||||
for (const call of state.modelCalls) {
|
||||
if (call.response === undefined) continue
|
||||
const content = call.response.content
|
||||
const text =
|
||||
typeof content === 'string'
|
||||
? content
|
||||
: content
|
||||
.map((part) => (part.type === 'text' ? part.text : ''))
|
||||
.filter(Boolean)
|
||||
.join('\n')
|
||||
if (text) {
|
||||
items.push({
|
||||
id: `${turnId}:a${call.index}`,
|
||||
role: 'assistant',
|
||||
content: text,
|
||||
timestamp: ts(),
|
||||
} satisfies ChatMessage)
|
||||
}
|
||||
if (Array.isArray(content)) {
|
||||
for (const part of content) {
|
||||
if (part.type !== 'tool-call') continue
|
||||
const tc = toolCallsById.get(part.toolCallId)
|
||||
items.push({
|
||||
id: part.toolCallId,
|
||||
name: part.toolName,
|
||||
input: part.arguments as ToolCall['input'],
|
||||
...(tc?.result ? { result: tc.result.result.output as ToolCall['result'] } : {}),
|
||||
status: tc ? toolStatus(tc) : 'running',
|
||||
timestamp: ts(),
|
||||
} satisfies ToolCall)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (state.terminal?.type === 'turn_failed') {
|
||||
items.push({
|
||||
id: `${turnId}:error`,
|
||||
kind: 'error',
|
||||
message: state.terminal.error,
|
||||
timestamp: ts(),
|
||||
} satisfies ErrorMessage)
|
||||
}
|
||||
|
||||
return items
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Session chat state (superset of the ChatTabViewState fields)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
type PermMeta = z.infer<typeof ToolPermissionMetadata>
|
||||
|
||||
export type SessionChatState = {
|
||||
conversation: ConversationItem[]
|
||||
currentAssistantMessage: string
|
||||
pendingAskHumanRequests: Map<string, z.infer<typeof AskHumanRequestEvent>>
|
||||
allPermissionRequests: Map<string, z.infer<typeof ToolPermissionRequestEvent>>
|
||||
permissionResponses: Map<string, PermissionResponse>
|
||||
autoPermissionDecisions: Map<string, z.infer<typeof ToolPermissionAutoDecisionEvent>>
|
||||
// Composer blocked / Stop shown until the latest turn settles (waiting on a
|
||||
// permission or ask-human still counts as processing).
|
||||
isProcessing: boolean
|
||||
// Actively working (non-terminal, nothing waiting on the user) — drives the
|
||||
// "Thinking…" shimmer; deliberately false under permission/ask-human cards.
|
||||
isThinking: boolean
|
||||
}
|
||||
|
||||
function toolCallPartOf(tc: ToolCallState) {
|
||||
return {
|
||||
type: 'tool-call' as const,
|
||||
toolCallId: tc.toolCallId,
|
||||
toolName: tc.toolName,
|
||||
arguments: tc.input,
|
||||
}
|
||||
}
|
||||
|
||||
// Compose the whole session's conversation: prior (settled) turns plus the
|
||||
// latest turn, with the live overlay stitched onto the latest turn's items.
|
||||
export function buildSessionChatState(
|
||||
turns: TurnState[],
|
||||
overlay: LiveOverlay,
|
||||
): SessionChatState {
|
||||
const conversation: ConversationItem[] = []
|
||||
for (const turn of turns) {
|
||||
conversation.push(...buildTurnConversation(turn))
|
||||
}
|
||||
for (let i = 0; i < conversation.length; i++) {
|
||||
const item = conversation[i]
|
||||
if ('name' in item && overlay.toolOutput[item.id]) {
|
||||
conversation[i] = { ...item, streamingOutput: overlay.toolOutput[item.id] }
|
||||
}
|
||||
}
|
||||
|
||||
const latest = turns[turns.length - 1]
|
||||
const status = latest ? deriveTurnStatus(latest) : undefined
|
||||
const latestTurnId = latest?.definition.turnId ?? ''
|
||||
|
||||
const allPermissionRequests = new Map<string, z.infer<typeof ToolPermissionRequestEvent>>()
|
||||
const permissionResponses = new Map<string, PermissionResponse>()
|
||||
const autoPermissionDecisions = new Map<
|
||||
string,
|
||||
z.infer<typeof ToolPermissionAutoDecisionEvent>
|
||||
>()
|
||||
const pendingAskHumanRequests = new Map<string, z.infer<typeof AskHumanRequestEvent>>()
|
||||
|
||||
if (latest) {
|
||||
for (const tc of outstandingPermissions(latest)) {
|
||||
allPermissionRequests.set(tc.toolCallId, {
|
||||
runId: latestTurnId,
|
||||
type: 'tool-permission-request',
|
||||
subflow: [],
|
||||
toolCall: toolCallPartOf(tc),
|
||||
permission: tc.permission?.required.request as PermMeta,
|
||||
})
|
||||
}
|
||||
for (const tc of latest.toolCalls) {
|
||||
const resolved = tc.permission?.resolved
|
||||
if (!resolved) continue
|
||||
if (resolved.source === 'human') {
|
||||
permissionResponses.set(tc.toolCallId, resolved.decision === 'allow' ? 'approve' : 'deny')
|
||||
} else if (resolved.source === 'classifier') {
|
||||
autoPermissionDecisions.set(tc.toolCallId, {
|
||||
runId: latestTurnId,
|
||||
type: 'tool-permission-auto-decision',
|
||||
subflow: [],
|
||||
toolCallId: tc.toolCallId,
|
||||
toolCall: toolCallPartOf(tc),
|
||||
permission: tc.permission?.required.request as PermMeta,
|
||||
decision: resolved.decision,
|
||||
reason: resolved.reason ?? '',
|
||||
})
|
||||
}
|
||||
}
|
||||
for (const tc of outstandingAsyncTools(latest)) {
|
||||
if (tc.toolName !== 'ask-human') continue
|
||||
const input = (tc.input ?? {}) as { question?: unknown; options?: unknown }
|
||||
pendingAskHumanRequests.set(tc.toolCallId, {
|
||||
runId: latestTurnId,
|
||||
type: 'ask-human-request',
|
||||
toolCallId: tc.toolCallId,
|
||||
subflow: [],
|
||||
query: typeof input.question === 'string' ? input.question : '',
|
||||
...(Array.isArray(input.options) && input.options.every((o) => typeof o === 'string')
|
||||
? { options: input.options }
|
||||
: {}),
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
const settled = status === 'completed' || status === 'failed' || status === 'cancelled'
|
||||
return {
|
||||
conversation,
|
||||
currentAssistantMessage: overlay.text,
|
||||
pendingAskHumanRequests,
|
||||
allPermissionRequests,
|
||||
permissionResponses,
|
||||
autoPermissionDecisions,
|
||||
isProcessing: latest !== undefined && !settled,
|
||||
isThinking: status === 'idle',
|
||||
}
|
||||
}
|
||||
1
apps/x/apps/renderer/src/test/setup.ts
Normal file
1
apps/x/apps/renderer/src/test/setup.ts
Normal file
|
|
@ -0,0 +1 @@
|
|||
import '@testing-library/jest-dom/vitest'
|
||||
|
|
@ -1,5 +1,5 @@
|
|||
import path from "path"
|
||||
import { defineConfig } from 'vite'
|
||||
import { defineConfig } from 'vitest/config'
|
||||
import react from '@vitejs/plugin-react'
|
||||
import tailwindcss from '@tailwindcss/vite'
|
||||
|
||||
|
|
@ -18,4 +18,10 @@ export default defineConfig({
|
|||
build: {
|
||||
outDir: 'dist',
|
||||
},
|
||||
test: {
|
||||
environment: 'jsdom',
|
||||
setupFiles: ['./src/test/setup.ts'],
|
||||
include: ['src/**/*.test.{ts,tsx}'],
|
||||
css: false,
|
||||
},
|
||||
})
|
||||
|
|
|
|||
|
|
@ -13,7 +13,9 @@ import {
|
|||
BackgroundTaskSummarySchema,
|
||||
TriggersSchema,
|
||||
} from './background-task.js';
|
||||
import { UserMessageContent } from './message.js';
|
||||
import { UserMessage, UserMessageContent } from './message.js';
|
||||
import { RequestedAgent, type TurnEvent } from './turns.js';
|
||||
import type { SessionBusEvent, SessionIndexEntry, SessionState } from './sessions.js';
|
||||
import { RowboatApiConfig } from './rowboat-account.js';
|
||||
import { ZListToolkitsResponse } from './composio.js';
|
||||
import { BrowserStateSchema } from './browser-control.js';
|
||||
|
|
@ -375,6 +377,82 @@ const ipcSchemas = {
|
|||
req: z.null(),
|
||||
res: z.null(),
|
||||
},
|
||||
// ── New runtime: sessions + turns (session-design.md) ────────────────────
|
||||
// Turn-mutating calls return quickly; the renderer follows progress through
|
||||
// the sessions:events feed and the shared reduceTurn reducer.
|
||||
'sessions:create': {
|
||||
req: z.object({ title: z.string().optional() }),
|
||||
res: z.object({ sessionId: z.string() }),
|
||||
},
|
||||
'sessions:list': {
|
||||
req: z.object({}),
|
||||
res: z.object({ sessions: z.array(z.custom<SessionIndexEntry>()) }),
|
||||
},
|
||||
'sessions:get': {
|
||||
req: z.object({ sessionId: z.string() }),
|
||||
res: z.custom<SessionState>(),
|
||||
},
|
||||
'sessions:getTurn': {
|
||||
// Events are strictly validated at the repository read; typed via
|
||||
// z.custom to avoid re-validating potentially large logs per IPC hop.
|
||||
req: z.object({ turnId: z.string() }),
|
||||
res: z.custom<{ turnId: string; events: Array<z.infer<typeof TurnEvent>> }>(),
|
||||
},
|
||||
'sessions:sendMessage': {
|
||||
req: z.object({
|
||||
sessionId: z.string(),
|
||||
input: UserMessage,
|
||||
config: z.object({
|
||||
agent: RequestedAgent,
|
||||
autoPermission: z.boolean().optional(),
|
||||
maxModelCalls: z.number().int().positive().optional(),
|
||||
}),
|
||||
}),
|
||||
res: z.object({ turnId: z.string() }),
|
||||
},
|
||||
'sessions:respondToPermission': {
|
||||
req: z.object({
|
||||
turnId: z.string(),
|
||||
toolCallId: z.string(),
|
||||
decision: z.enum(['allow', 'deny']),
|
||||
metadata: z.json().optional(),
|
||||
}),
|
||||
res: z.object({ success: z.literal(true) }),
|
||||
},
|
||||
'sessions:respondToAskHuman': {
|
||||
req: z.object({
|
||||
turnId: z.string(),
|
||||
toolCallId: z.string(),
|
||||
answer: z.string(),
|
||||
}),
|
||||
res: z.object({ success: z.literal(true) }),
|
||||
},
|
||||
'sessions:stopTurn': {
|
||||
req: z.object({
|
||||
turnId: z.string(),
|
||||
reason: z.string().optional(),
|
||||
}),
|
||||
res: z.object({ success: z.literal(true) }),
|
||||
},
|
||||
'sessions:resumeTurn': {
|
||||
req: z.object({ sessionId: z.string() }),
|
||||
res: z.object({ success: z.literal(true) }),
|
||||
},
|
||||
'sessions:setTitle': {
|
||||
req: z.object({ sessionId: z.string(), title: z.string() }),
|
||||
res: z.object({ success: z.literal(true) }),
|
||||
},
|
||||
'sessions:delete': {
|
||||
req: z.object({ sessionId: z.string() }),
|
||||
res: z.object({ success: z.literal(true) }),
|
||||
},
|
||||
'sessions:events': {
|
||||
// Typed via z.custom so the renderer's `on` handler is typed without
|
||||
// runtime validation (the broadcast path bypasses preload validation,
|
||||
// like runs:events).
|
||||
req: z.custom<SessionBusEvent>(),
|
||||
res: z.null(),
|
||||
},
|
||||
'services:events': {
|
||||
req: ServiceEvent,
|
||||
res: z.null(),
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue