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'