From dbcdcda5e7a964cc0bef3cdd0003100453925fad Mon Sep 17 00:00:00 2001 From: Croissant Le Doux Date: Thu, 9 Jul 2026 00:22:12 -0400 Subject: [PATCH] feat: stream Reginald's replies token-by-token MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The 26b is slow (~30s/call); the chat now shows the answer forming instead of freezing until it's done. The final prose streams over SSE; tool-calling turns stay structured (no partial tokens), so streaming kicks in for the narration. core (@commitea/core): - chat-client.complete gains an optional onToken — when set, it requests stream:true and parses the OpenAI SSE stream, emitting content deltas and assembling streamed tool-call argument fragments into the final result. - GiteaHttpResponse exposes the optional `body` stream (real fetch has it; stubs don't). agent-loop threads onToken to each completion. app: - model:chat forwards each delta to the renderer (event.sender.send); preload exposes model.onToken(cb) → unsubscribe. useChat accumulates the live stream into a growing bubble (with a cursor), replaced by the authoritative final content when the turn resolves. Unconfigured → scripted reply, unchanged. Verified: 118 core tests green (2 streaming: SSE content deltas + tool-call fragment assembly), desktop typecheck clean, 14 fixture e2e green. Live: a real turn against gemma-4-26b assembles the correct answer via the streaming path (live-reginald green) — the reply now renders token-by-token. Co-Authored-By: Claude Opus 4.8 (1M context) --- apps/desktop/src/main/model.ts | 9 ++- apps/desktop/src/preload/index.ts | 6 ++ .../src/components/shell/chat-panel.tsx | 13 +++- apps/desktop/src/renderer/src/global.d.ts | 2 + apps/desktop/src/renderer/src/lib/use-chat.ts | 16 ++++- packages/core/src/agent/agent-loop.ts | 8 ++- packages/core/src/agent/agent.test.ts | 42 ++++++++++++ packages/core/src/agent/chat-client.ts | 67 ++++++++++++++++++- packages/core/src/gitea/types.ts | 2 + packages/core/src/index.ts | 2 +- 10 files changed, 152 insertions(+), 15 deletions(-) diff --git a/apps/desktop/src/main/model.ts b/apps/desktop/src/main/model.ts index 2859397..416c905 100644 --- a/apps/desktop/src/main/model.ts +++ b/apps/desktop/src/main/model.ts @@ -86,11 +86,15 @@ export function registerModelIpc(): void { return { configured: true, model } }) - ipcMain.handle('model:chat', async (_event, messages: ChatMessage[]) => { + ipcMain.handle('model:chat', async (event, messages: ChatMessage[]) => { if (!router) return { ok: false as const, reason: 'unconfigured' as const } const client = getGiteaClient() const model = await resolveLoadedModel(router.small.baseUrl, router.small.model) const chat = createChatClient({ ...router.small, model }, fetch) + // stream the model's prose to the renderer token-by-token + const onToken = (delta: string) => { + if (!event.sender.isDestroyed()) event.sender.send('model:chat:token', delta) + } // Proposals the model formulates this turn; the renderer approves them (the // write happens through gitea:applyChange, never inside the loop). @@ -129,10 +133,11 @@ export function registerModelIpc(): void { try { const turn = await runAgentTurn({ - complete: (m, t) => chat.complete(m, t), + complete: (m, t, ot) => chat.complete(m, t, ot), messages: [{ role: 'system', content: REGINALD_SYSTEM }, ...messages], tools: REGINALD_TOOLS, execute, + onToken, }) return { ok: true as const, content: turn.content, steps: turn.steps, proposals } } catch (e) { diff --git a/apps/desktop/src/preload/index.ts b/apps/desktop/src/preload/index.ts index a253128..4705f6c 100644 --- a/apps/desktop/src/preload/index.ts +++ b/apps/desktop/src/preload/index.ts @@ -25,6 +25,12 @@ const api = { status: () => ipcRenderer.invoke('model:status'), /** One agent turn: messages in, Reginald's prose + the tools it consulted out. */ chat: (messages: unknown) => ipcRenderer.invoke('model:chat', messages), + /** Subscribe to streamed prose tokens for the in-flight turn; returns an unsubscribe. */ + onToken: (cb: (delta: string) => void) => { + const listener = (_e: unknown, delta: string) => cb(delta) + ipcRenderer.on('model:chat:token', listener) + return () => ipcRenderer.removeListener('model:chat:token', listener) + }, /** Decompose a braindump into a proposed issue set (capture_work). */ capture: (braindump: string) => ipcRenderer.invoke('model:capture', braindump), }, diff --git a/apps/desktop/src/renderer/src/components/shell/chat-panel.tsx b/apps/desktop/src/renderer/src/components/shell/chat-panel.tsx index ca4bd6e..082f7e4 100644 --- a/apps/desktop/src/renderer/src/components/shell/chat-panel.tsx +++ b/apps/desktop/src/renderer/src/components/shell/chat-panel.tsx @@ -19,7 +19,7 @@ export interface ChatPanelProps { } export function ChatPanel({ onOpenDirectives, offline, onApplyChange }: ChatPanelProps) { - const { msgs, thinking, live, model, steps, proposals, send: sendChat, approve, dismiss } = useChat(onApplyChange) + const { msgs, thinking, live, model, steps, proposals, streaming, send: sendChat, approve, dismiss } = useChat(onApplyChange) // shorten "google/gemma-4-26b-a4b-qat" → "gemma-4-26b" for the header chip const modelLabel = model ? (model.split('/').pop() ?? model).replace(/-(qat|instruct|it|gguf)$/i, '') : 'gemma-4' const [text, setText] = useState('') @@ -28,7 +28,7 @@ export function ChatPanel({ onOpenDirectives, offline, onApplyChange }: ChatPane useEffect(() => { const el = scrollRef.current if (el) el.scrollTop = el.scrollHeight - }, [msgs, thinking]) + }, [msgs, thinking, streaming]) const send = () => { const t = text.trim() @@ -98,7 +98,14 @@ export function ChatPanel({ onOpenDirectives, offline, onApplyChange }: ChatPane The model is away from its desk. Reads still work; writes will wait their turn. ) : null} - {thinking ?
considering…
: null} + {streaming ? ( +
+ {streaming} + ▊ +
+ ) : thinking ? ( +
considering…
+ ) : null} {!thinking && steps.length ? (
consulted {Array.from(new Set(steps.map((s) => s.replace('query_project', 'the project').replace('propose_change', 'the labels').replace('record_directive', 'the directive ledger')))).join(', ')} diff --git a/apps/desktop/src/renderer/src/global.d.ts b/apps/desktop/src/renderer/src/global.d.ts index 3235a0a..93434ba 100644 --- a/apps/desktop/src/renderer/src/global.d.ts +++ b/apps/desktop/src/renderer/src/global.d.ts @@ -68,6 +68,8 @@ export interface ModelBridge { status(): Promise<{ configured: boolean; model: string | null }> chat(messages: ChatMessage[]): Promise capture(braindump: string): Promise + /** Subscribe to streamed prose tokens; returns an unsubscribe fn. */ + onToken(cb: (delta: string) => void): () => void } /** The result of reading the directive ledger. */ diff --git a/apps/desktop/src/renderer/src/lib/use-chat.ts b/apps/desktop/src/renderer/src/lib/use-chat.ts index ae12830..a0eb33b 100644 --- a/apps/desktop/src/renderer/src/lib/use-chat.ts +++ b/apps/desktop/src/renderer/src/lib/use-chat.ts @@ -20,6 +20,8 @@ export interface ChatState { steps: string[] /** Changes Reginald has proposed and is awaiting approval on. */ proposals: ChangeProposal[] + /** The in-flight streamed prose (grows token-by-token) before the turn finalizes. */ + streaming: string send: (text: string) => void approve: (p: ChangeProposal) => void dismiss: (p: ChangeProposal) => void @@ -41,6 +43,7 @@ export function useChat(onApplyChange?: (change: IssueChange) => Promise<{ ok: b const [model, setModel] = useState(null) const [steps, setSteps] = useState([]) const [proposals, setProposals] = useState([]) + const [streaming, setStreaming] = useState('') const convoRef = useRef(convo) convoRef.current = convo @@ -86,10 +89,17 @@ export function useChat(onApplyChange?: (change: IssueChange) => Promise<{ ok: b role: m.from === 'user' ? 'user' : 'assistant', content: m.text, })) + setStreaming('') + const unsubscribe = window.commitea.model.onToken((delta) => setStreaming((s) => s + delta)) + const finish = () => { + unsubscribe() + setThinking(false) + setStreaming('') + } window.commitea.model .chat(wire) .then((res) => { - setThinking(false) + finish() if (res.ok) { setSteps(res.steps.map((s) => s.tool)) setProposals(res.proposals) @@ -105,7 +115,7 @@ export function useChat(onApplyChange?: (change: IssueChange) => Promise<{ ok: b } }) .catch(() => { - setThinking(false) + finish() setConvo((c) => [...c, { from: 'agent', text: 'I could not reach the model.' }]) }) }, @@ -137,5 +147,5 @@ export function useChat(onApplyChange?: (change: IssueChange) => Promise<{ ok: b setConvo((c) => [...c, { from: 'agent', text: `Left #${p.change.issue} as it was.` }]) }, []) - return { msgs: [...seed, ...convo], thinking, live, model, steps, proposals, send, approve, dismiss } + return { msgs: [...seed, ...convo], thinking, live, model, steps, proposals, streaming, send, approve, dismiss } } diff --git a/packages/core/src/agent/agent-loop.ts b/packages/core/src/agent/agent-loop.ts index dc174d8..f8783d7 100644 --- a/packages/core/src/agent/agent-loop.ts +++ b/packages/core/src/agent/agent-loop.ts @@ -31,18 +31,20 @@ function stringify(result: unknown): string { } export async function runAgentTurn(opts: { - complete: (messages: ChatMessage[], tools?: ToolDecl[]) => Promise + complete: (messages: ChatMessage[], tools?: ToolDecl[], onToken?: (delta: string) => void) => Promise messages: ChatMessage[] tools: ToolDecl[] execute: ToolExecutor maxSteps?: number + /** Streams content deltas as the model produces prose (final-answer streaming). */ + onToken?: (delta: string) => void }): Promise { const maxSteps = opts.maxSteps ?? DEFAULT_MAX_STEPS const convo: ChatMessage[] = [...opts.messages] const steps: AgentStep[] = [] for (let step = 0; step < maxSteps; step++) { - const { content, toolCalls } = await opts.complete(convo, opts.tools) + const { content, toolCalls } = await opts.complete(convo, opts.tools, opts.onToken) if (toolCalls.length === 0) { convo.push({ role: 'assistant', content }) return { content, steps, messages: convo } @@ -63,7 +65,7 @@ export async function runAgentTurn(opts: { } // Out of tool budget — force a final prose answer with tools withheld. - const final = await opts.complete(convo, []) + const final = await opts.complete(convo, [], opts.onToken) convo.push({ role: 'assistant', content: final.content }) return { content: final.content, steps, messages: convo } } diff --git a/packages/core/src/agent/agent.test.ts b/packages/core/src/agent/agent.test.ts index b94ba20..dda1602 100644 --- a/packages/core/src/agent/agent.test.ts +++ b/packages/core/src/agent/agent.test.ts @@ -25,6 +25,48 @@ describe('createChatClient', () => { return { fetch, calls } } + function sseStub(chunks: string[]) { + const calls: { url: string; body: unknown }[] = [] + const fetch: FetchLike = (url, init) => { + calls.push({ url, body: init?.body ? JSON.parse(init.body) : undefined }) + const enc = new TextEncoder() + const body = new ReadableStream({ + start(c) { + for (const ch of chunks) c.enqueue(enc.encode(ch)) + c.close() + }, + }) + return Promise.resolve({ ok: true, status: 200, body, json: () => Promise.resolve({}), text: () => Promise.resolve('') }) + } + return { fetch, calls } + } + + it('streams content deltas via onToken and returns the assembled result', async () => { + const { fetch, calls } = sseStub([ + 'data: {"choices":[{"delta":{"content":"Right "}}]}\n\n', + 'data: {"choices":[{"delta":{"content":"now: #2."}}]}\n\n', + 'data: [DONE]\n\n', + ]) + const client = createChatClient({ baseUrl: 'http://x/v1', model: 'm' }, fetch) + const tokens: string[] = [] + const res = await client.complete([{ role: 'user', content: 'now?' }], undefined, (d) => tokens.push(d)) + + expect(tokens).toEqual(['Right ', 'now: #2.']) + expect(res.content).toBe('Right now: #2.') + expect((calls[0].body as { stream?: boolean }).stream).toBe(true) + }) + + it('assembles a streamed tool call from argument deltas', async () => { + const { fetch } = sseStub([ + 'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"id":"c1","function":{"name":"query_project","arguments":"{\\"view\\""}}]}}]}\n\n', + 'data: {"choices":[{"delta":{"tool_calls":[{"index":0,"function":{"arguments":":\\"focus\\"}"}}]}}]}\n\n', + 'data: [DONE]\n\n', + ]) + const client = createChatClient({ baseUrl: 'http://x/v1', model: 'm' }, fetch) + const res = await client.complete([{ role: 'user', content: 'x' }], [{ name: 'query_project', description: '', parameters: {} }], () => {}) + expect(res.toolCalls).toEqual([{ id: 'c1', name: 'query_project', arguments: '{"view":"focus"}' }]) + }) + it('POSTs to /chat/completions and parses content', async () => { const { fetch, calls } = stub({ choices: [{ message: { content: 'the focus is #2' } }] }) const client = createChatClient({ baseUrl: 'http://localhost:1234/v1', model: 'gemma' }, fetch) diff --git a/packages/core/src/agent/chat-client.ts b/packages/core/src/agent/chat-client.ts index 79280e2..94444c8 100644 --- a/packages/core/src/agent/chat-client.ts +++ b/packages/core/src/agent/chat-client.ts @@ -48,8 +48,12 @@ export interface CompletionResult { toolCalls: ToolCall[] } +/** Called with each streamed content delta (final-prose streaming). */ +export type OnToken = (delta: string) => void + export interface ChatClient { - complete(messages: ChatMessage[], tools?: ToolDecl[]): Promise + /** When `onToken` is given, the response streams (SSE) and each content delta is emitted. */ + complete(messages: ChatMessage[], tools?: ToolDecl[], onToken?: OnToken): Promise } /** Map our message shape to the OpenAI wire shape. */ @@ -85,7 +89,7 @@ export function createChatClient(config: ModelConfig, fetchImpl: FetchLike): Cha const url = `${config.baseUrl.replace(/\/+$/, '')}/chat/completions` return { - async complete(messages, tools) { + async complete(messages, tools, onToken) { const body: Record = { model: config.model, messages: messages.map(toWireMessage), @@ -95,11 +99,13 @@ export function createChatClient(config: ModelConfig, fetchImpl: FetchLike): Cha body.tools = tools.map(toWireTool) body.tool_choice = 'auto' } + const stream = !!onToken + if (stream) body.stream = true const res = await fetchImpl(url, { method: 'POST', headers: { 'Content-Type': 'application/json', - Accept: 'application/json', + Accept: stream ? 'text/event-stream' : 'application/json', ...(config.apiKey ? { Authorization: `Bearer ${config.apiKey}` } : {}), }, body: JSON.stringify(body), @@ -108,6 +114,8 @@ export function createChatClient(config: ModelConfig, fetchImpl: FetchLike): Cha const text = await res.text().catch(() => '') throw new Error(`model completion failed (${res.status}): ${text.slice(0, 200)}`) } + if (stream && res.body) return readStream(res.body, onToken!) + const json = (await res.json()) as { choices?: { message: RawChoiceMessage }[] } const msg = json.choices?.[0]?.message return { @@ -121,3 +129,56 @@ export function createChatClient(config: ModelConfig, fetchImpl: FetchLike): Cha }, } } + +/** Streamed tool-call delta: name arrives first, arguments accumulate across chunks. */ +interface RawToolDelta { + index: number + id?: string + function?: { name?: string; arguments?: string } +} + +/** Parse an OpenAI SSE stream: emit content deltas via onToken, accumulate the final result. */ +async function readStream(body: ReadableStream, onToken: OnToken): Promise { + const reader = body.getReader() + const decoder = new TextDecoder() + let buffer = '' + let content = '' + const toolAcc: { id: string; name: string; arguments: string }[] = [] + + const handle = (data: string) => { + if (data === '[DONE]') return + let chunk: { choices?: { delta?: { content?: string; tool_calls?: RawToolDelta[] } }[] } + try { + chunk = JSON.parse(data) + } catch { + return + } + const delta = chunk.choices?.[0]?.delta + if (!delta) return + if (delta.content) { + content += delta.content + onToken(delta.content) + } + for (const tc of delta.tool_calls ?? []) { + const slot = (toolAcc[tc.index] ??= { id: '', name: '', arguments: '' }) + if (tc.id) slot.id = tc.id + if (tc.function?.name) slot.name = tc.function.name + if (tc.function?.arguments) slot.arguments += tc.function.arguments + } + } + + for (;;) { + const { done, value } = await reader.read() + if (done) break + buffer += decoder.decode(value, { stream: true }) + const lines = buffer.split('\n') + buffer = lines.pop() ?? '' + for (const line of lines) { + const t = line.trim() + if (t.startsWith('data:')) handle(t.slice(5).trim()) + } + } + if (buffer.trim().startsWith('data:')) handle(buffer.trim().slice(5).trim()) + + return { content, toolCalls: toolAcc.filter((t) => t.name).map((t) => ({ id: t.id, name: t.name, arguments: t.arguments })) } +} diff --git a/packages/core/src/gitea/types.ts b/packages/core/src/gitea/types.ts index 81b9c1d..c39a1bf 100644 --- a/packages/core/src/gitea/types.ts +++ b/packages/core/src/gitea/types.ts @@ -31,6 +31,8 @@ export interface GiteaHttpResponse { status: number json(): Promise text(): Promise + /** Present on the real fetch Response; used for SSE streaming (chat). */ + body?: ReadableStream | null } export type FetchLike = (url: string, init?: GiteaRequestInit) => Promise diff --git a/packages/core/src/index.ts b/packages/core/src/index.ts index d190e3a..a08ec87 100644 --- a/packages/core/src/index.ts +++ b/packages/core/src/index.ts @@ -80,7 +80,7 @@ export type { } from './calibration/calibration-v0.js' export { createChatClient } from './agent/chat-client.js' -export type { ChatClient, ChatMessage, CompletionResult, ModelConfig, ToolCall, ToolDecl } from './agent/chat-client.js' +export type { ChatClient, ChatMessage, CompletionResult, ModelConfig, OnToken, ToolCall, ToolDecl } from './agent/chat-client.js' export { pickModel } from './agent/model-router.js' export type { ModelRouter, TaskKind } from './agent/model-router.js' export { runAgentTurn } from './agent/agent-loop.js' -- 2.49.1