Compare commits
5 Commits
p4/directi
...
dbcdcda5e7
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
dbcdcda5e7 | ||
| 403aef9d16 | |||
|
|
2f6636684e | ||
|
|
fca36f9075 | ||
| aab7fd6eeb |
37
apps/desktop/e2e/live-persistence.spec.ts
Normal file
37
apps/desktop/e2e/live-persistence.spec.ts
Normal file
@@ -0,0 +1,37 @@
|
||||
import { dirname, join } from 'node:path'
|
||||
import { fileURLToPath } from 'node:url'
|
||||
|
||||
import { _electron as electron, expect, test } from '@playwright/test'
|
||||
|
||||
const here = dirname(fileURLToPath(import.meta.url))
|
||||
const MAIN = join(here, '..', 'out', 'main', 'index.js')
|
||||
|
||||
// Opt-in (GITEA_LIVE=1). A real launch persists the snapshot; a second launch with
|
||||
// gitea unreachable must still show the board + real scheduler output from the
|
||||
// persisted cache (offline reads). No model needed.
|
||||
test.describe('live persistence', () => {
|
||||
test('offline: serves the persisted snapshot', async () => {
|
||||
test.skip(!process.env.GITEA_LIVE, 'GITEA_LIVE not set — opt-in live test')
|
||||
test.setTimeout(120_000)
|
||||
|
||||
// launch 1 — real reconcile writes the snapshot to disk
|
||||
const app1 = await electron.launch({ args: [MAIN], env: { ...process.env } })
|
||||
const w1 = await app1.firstWindow()
|
||||
await w1.waitForLoadState('domcontentloaded')
|
||||
// a scheduler-only phrase confirms real data reconciled (never emitted by fixtures)
|
||||
await expect(w1.getByText(/on the critical path|unblocks #|waits on #|· ready/).first()).toBeVisible({ timeout: 30000 })
|
||||
await app1.close()
|
||||
|
||||
// launch 2 — gitea unreachable; the reconcile must fall back to the persisted snapshot
|
||||
const app2 = await electron.launch({
|
||||
args: [MAIN],
|
||||
env: { ...process.env, GITEA_BASE_URL: 'http://127.0.0.1:9' },
|
||||
})
|
||||
const w2 = await app2.firstWindow()
|
||||
await w2.waitForLoadState('domcontentloaded')
|
||||
// Focus still renders real scheduler output — proving it came from the cache, offline
|
||||
await expect(w2.getByText(/on the critical path|unblocks #|waits on #|· ready/).first()).toBeVisible({ timeout: 20000 })
|
||||
await w2.screenshot({ path: join(here, '.artifacts', 'screens', 'live-offline.png'), fullPage: true, animations: 'disabled' })
|
||||
await app2.close()
|
||||
})
|
||||
})
|
||||
@@ -27,6 +27,8 @@ import {
|
||||
} from '@commitea/core'
|
||||
import { ipcMain } from 'electron'
|
||||
|
||||
import { loadSnapshot, saveSnapshot } from './snapshot-store.js'
|
||||
|
||||
/** Walk up from cwd looking for a .env.local with a GITEA_TOKEN (dev convenience). */
|
||||
function loadEnvLocalToken(): string | undefined {
|
||||
let dir = process.cwd()
|
||||
@@ -126,16 +128,81 @@ export async function reconcileSnapshot(
|
||||
return { issues, milestones, deps, timelines }
|
||||
}
|
||||
|
||||
type Snapshot = Awaited<ReturnType<typeof reconcileSnapshot>>
|
||||
|
||||
/**
|
||||
* A single in-memory reconcile cache shared across the app. A full reconcile is
|
||||
* ~2N gitea calls (deps + timelines per issue); without this, every agent tool
|
||||
* call refetched the whole repo. Reads within `maxAgeMs` reuse the cache;
|
||||
* `getSnapshot({ maxAgeMs: 0 })` forces a fresh pull (the explicit UI reconcile),
|
||||
* and any write calls `invalidateSnapshot()` so the next read sees it. The cache
|
||||
* is rebuildable — the durable truth stays in gitea (the purity split, D4).
|
||||
*/
|
||||
let snapshotCache: { snap: Snapshot; at: number } | null = null
|
||||
|
||||
export async function getSnapshot(client: GiteaClient, opts?: { maxAgeMs?: number }): Promise<Snapshot> {
|
||||
const maxAgeMs = opts?.maxAgeMs ?? 0
|
||||
if (snapshotCache && maxAgeMs > 0 && Date.now() - snapshotCache.at <= maxAgeMs) {
|
||||
return snapshotCache.snap
|
||||
}
|
||||
const snap = await reconcileSnapshot(client)
|
||||
snapshotCache = { snap, at: Date.now() }
|
||||
saveSnapshot(snap, new Date().toISOString()) // persist for instant boot + offline
|
||||
return snap
|
||||
}
|
||||
|
||||
/** Drop the cache so the next read reflects a just-made write. */
|
||||
export function invalidateSnapshot(): void {
|
||||
snapshotCache = null
|
||||
}
|
||||
|
||||
/**
|
||||
* The last persisted snapshot (from a previous session), for instant boot. The
|
||||
* renderer shows it immediately, then a real reconcile supersedes it
|
||||
* (stale-while-revalidate). Returns null when there's nothing on disk; its
|
||||
* `savedAt` marks staleness. It does NOT seed the cache — agent tool calls
|
||||
* always reconcile fresh so they never reason over stale data.
|
||||
*/
|
||||
export function bootSnapshot(): (Snapshot & { savedAt: string }) | null {
|
||||
const persisted = loadSnapshot()
|
||||
if (!persisted) return null
|
||||
return {
|
||||
issues: persisted.issues,
|
||||
milestones: persisted.milestones,
|
||||
deps: persisted.deps,
|
||||
timelines: persisted.timelines,
|
||||
savedAt: persisted.savedAt,
|
||||
} as unknown as Snapshot & { savedAt: string }
|
||||
}
|
||||
|
||||
/** Agent tool calls tolerate a slightly stale snapshot (seconds) to stay responsive. */
|
||||
export const AGENT_SNAPSHOT_TTL_MS = 30_000
|
||||
|
||||
export function registerGiteaIpc(): void {
|
||||
const client = getGiteaClient()
|
||||
const repo = client ? `${process.env.GITEA_OWNER ?? 'christian'}/${process.env.GITEA_REPO ?? 'commitea'}` : null
|
||||
|
||||
ipcMain.handle('gitea:status', () => ({ configured: !!client, repo }))
|
||||
|
||||
// Instant boot: the last persisted snapshot, shown before the fresh reconcile lands.
|
||||
ipcMain.handle('gitea:boot', () => {
|
||||
if (!client) return { configured: false }
|
||||
const persisted = bootSnapshot()
|
||||
return persisted ? { configured: true, cached: true, ...persisted } : { configured: true, cached: false }
|
||||
})
|
||||
|
||||
ipcMain.handle('gitea:reconcile', async () => {
|
||||
if (!client) return { configured: false, issues: [], milestones: [], deps: [], timelines: {} }
|
||||
const snap = await reconcileSnapshot(client)
|
||||
return { configured: true, ...snap }
|
||||
try {
|
||||
// explicit UI sync — force fresh, and warm the cache for agent tool calls
|
||||
const snap = await getSnapshot(client, { maxAgeMs: 0 })
|
||||
return { configured: true, stale: false, ...snap }
|
||||
} catch (e) {
|
||||
// offline / gitea down — serve the last persisted snapshot rather than error out
|
||||
const persisted = bootSnapshot()
|
||||
if (persisted) return { configured: true, stale: true, ...persisted }
|
||||
throw e
|
||||
}
|
||||
})
|
||||
|
||||
ipcMain.handle('gitea:getIssue', async (_event, index: number) => {
|
||||
@@ -167,6 +234,7 @@ export function registerGiteaIpc(): void {
|
||||
const ids = await resolveLabelIds(plan.labels)
|
||||
await client.setIssueLabels(change.issue, ids)
|
||||
const issue = await client.getIssue(change.issue)
|
||||
invalidateSnapshot() // the board + forecast must reflect the label change
|
||||
return { ok: true as const, plan, issue }
|
||||
})
|
||||
|
||||
@@ -183,6 +251,7 @@ export function registerGiteaIpc(): void {
|
||||
const issue = await client.createIssue({ title: it.title, body: it.body, labelIds })
|
||||
created.push({ number: issue.number, title: issue.title })
|
||||
}
|
||||
if (created.length) invalidateSnapshot() // new issues enter the board/scope
|
||||
return { ok: true as const, created }
|
||||
},
|
||||
)
|
||||
|
||||
@@ -26,7 +26,13 @@ import {
|
||||
} from '@commitea/core'
|
||||
import { ipcMain } from 'electron'
|
||||
|
||||
import { appendDirectiveEntry, getGiteaClient, getPmStateClient, reconcileSnapshot } from './gitea.js'
|
||||
import {
|
||||
AGENT_SNAPSHOT_TTL_MS,
|
||||
appendDirectiveEntry,
|
||||
getGiteaClient,
|
||||
getPmStateClient,
|
||||
getSnapshot,
|
||||
} from './gitea.js'
|
||||
|
||||
/** Small local model for prose + the read tool; big model reserved for later decomposition. */
|
||||
function resolveModelRouter(): ModelRouter | null {
|
||||
@@ -80,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).
|
||||
@@ -93,7 +103,8 @@ export function registerModelIpc(): void {
|
||||
const execute = async (name: string, args: unknown) => {
|
||||
if (!client) return { error: 'gitea is not configured' }
|
||||
if (name === 'query_project') {
|
||||
const snap = await reconcileSnapshot(client)
|
||||
// reuse a recent reconcile — a multi-tool turn shouldn't refetch the repo each call
|
||||
const snap = await getSnapshot(client, { maxAgeMs: AGENT_SNAPSHOT_TTL_MS })
|
||||
const a = (args ?? {}) as { view: ProjectView; filters?: QueryFilters }
|
||||
return buildProjectView(a.view, a.filters, snap, new Date())
|
||||
}
|
||||
@@ -122,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) {
|
||||
|
||||
47
apps/desktop/src/main/snapshot-store.ts
Normal file
47
apps/desktop/src/main/snapshot-store.ts
Normal file
@@ -0,0 +1,47 @@
|
||||
/**
|
||||
* Durable snapshot store — the reconcile cache, persisted to disk. On boot the
|
||||
* app shows the last snapshot instantly (stale-while-revalidate) instead of a
|
||||
* blank board while ~2N gitea calls run; if gitea is unreachable, reads fall
|
||||
* back to it (offline). It's a rebuildable mirror — the durable truth stays in
|
||||
* gitea (the purity split, D4). A plain JSON file: the whole snapshot fits in
|
||||
* memory at this scale, so indexed SQL buys nothing yet (see the PR).
|
||||
*/
|
||||
|
||||
import { readFileSync, writeFileSync } from 'node:fs'
|
||||
import { join } from 'node:path'
|
||||
|
||||
import { app } from 'electron'
|
||||
|
||||
/** The shape we persist — kept loose so a schema drift degrades to "no cache", not a crash. */
|
||||
export interface PersistedSnapshot {
|
||||
issues: unknown[]
|
||||
milestones: unknown[]
|
||||
deps: unknown[]
|
||||
timelines: Record<number, unknown[]>
|
||||
/** ISO time the snapshot was reconciled — shown as "cached since". */
|
||||
savedAt: string
|
||||
}
|
||||
|
||||
function snapshotPath(): string {
|
||||
return join(app.getPath('userData'), 'commitea-snapshot.json')
|
||||
}
|
||||
|
||||
/** Load the last persisted snapshot, or null if absent/corrupt. Never throws. */
|
||||
export function loadSnapshot(): PersistedSnapshot | null {
|
||||
try {
|
||||
const parsed = JSON.parse(readFileSync(snapshotPath(), 'utf8')) as PersistedSnapshot
|
||||
if (parsed && Array.isArray(parsed.issues)) return parsed
|
||||
return null
|
||||
} catch {
|
||||
return null // missing file, bad JSON, or drift — treat as no cache
|
||||
}
|
||||
}
|
||||
|
||||
/** Persist a freshly reconciled snapshot. Best-effort — a write failure never breaks a reconcile. */
|
||||
export function saveSnapshot(snap: Omit<PersistedSnapshot, 'savedAt'>, savedAt: string): void {
|
||||
try {
|
||||
writeFileSync(snapshotPath(), JSON.stringify({ ...snap, savedAt }), 'utf8')
|
||||
} catch {
|
||||
// disk full / permissions — the in-memory cache still works this session
|
||||
}
|
||||
}
|
||||
@@ -5,6 +5,8 @@ const api = {
|
||||
gitea: {
|
||||
/** Whether the main process has a gitea token + target repo configured. */
|
||||
status: () => ipcRenderer.invoke('gitea:status'),
|
||||
/** The last persisted snapshot, for instant boot before the fresh reconcile. */
|
||||
boot: () => ipcRenderer.invoke('gitea:boot'),
|
||||
/** Full read of the managed repo — every issue + milestone. */
|
||||
reconcile: () => ipcRenderer.invoke('gitea:reconcile'),
|
||||
/** One issue by index, normalized (or null if unconfigured). */
|
||||
@@ -23,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),
|
||||
},
|
||||
|
||||
@@ -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.
|
||||
</div>
|
||||
) : null}
|
||||
{thinking ? <div style={{ font: 'var(--text-agent)', color: 'var(--ink-3)' }}>considering…</div> : null}
|
||||
{streaming ? (
|
||||
<div style={{ font: 'var(--text-agent)', color: 'var(--ink-1)', lineHeight: 1.55 }}>
|
||||
{streaming}
|
||||
<span style={{ opacity: 0.5 }}>▊</span>
|
||||
</div>
|
||||
) : thinking ? (
|
||||
<div style={{ font: 'var(--text-agent)', color: 'var(--ink-3)' }}>considering…</div>
|
||||
) : null}
|
||||
{!thinking && steps.length ? (
|
||||
<div style={{ font: 'var(--text-caption)', color: 'var(--ink-3)', display: 'flex', alignItems: 'center', gap: 5 }}>
|
||||
<Icon name="eye" size={11} /> 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(', ')}
|
||||
|
||||
32
apps/desktop/src/renderer/src/global.d.ts
vendored
32
apps/desktop/src/renderer/src/global.d.ts
vendored
@@ -28,17 +28,31 @@ export type CaptureResult =
|
||||
| { ok: false; reason: 'unconfigured' | 'error'; message?: string }
|
||||
| ({ ok: true } & CaptureProposal)
|
||||
|
||||
/** A reconciled snapshot as it crosses the bridge. */
|
||||
export interface SnapshotPayload {
|
||||
configured: boolean
|
||||
issues: GiteaIssue[]
|
||||
milestones: GiteaMilestone[]
|
||||
deps: DependencyEdge[]
|
||||
/** Normalized lifecycle events keyed by issue number. */
|
||||
timelines: Record<number, LifecycleEvent[]>
|
||||
/** true when served from the persisted cache (offline / instant boot). */
|
||||
stale?: boolean
|
||||
/** ISO time the persisted snapshot was reconciled (present on cached reads). */
|
||||
savedAt?: string
|
||||
}
|
||||
|
||||
/** Boot payload — the persisted snapshot, or a marker that there's none yet. */
|
||||
export type BootPayload =
|
||||
| { configured: false }
|
||||
| { configured: true; cached: false }
|
||||
| ({ configured: true; cached: true } & Omit<SnapshotPayload, 'configured'>)
|
||||
|
||||
/** The gitea bridge exposed by the preload over IPC (main-process backed). */
|
||||
export interface GiteaBridge {
|
||||
status(): Promise<{ configured: boolean; repo: string | null }>
|
||||
reconcile(): Promise<{
|
||||
configured: boolean
|
||||
issues: GiteaIssue[]
|
||||
milestones: GiteaMilestone[]
|
||||
deps: DependencyEdge[]
|
||||
/** Normalized lifecycle events keyed by issue number. */
|
||||
timelines: Record<number, LifecycleEvent[]>
|
||||
}>
|
||||
boot(): Promise<BootPayload>
|
||||
reconcile(): Promise<SnapshotPayload>
|
||||
getIssue(index: number): Promise<GiteaIssue | null>
|
||||
applyChange(change: IssueChange): Promise<ApplyChangeResult>
|
||||
createIssues(issues: ProposedIssue[]): Promise<CreateIssuesResult>
|
||||
@@ -54,6 +68,8 @@ export interface ModelBridge {
|
||||
status(): Promise<{ configured: boolean; model: string | null }>
|
||||
chat(messages: ChatMessage[]): Promise<ChatResult>
|
||||
capture(braindump: string): Promise<CaptureResult>
|
||||
/** Subscribe to streamed prose tokens; returns an unsubscribe fn. */
|
||||
onToken(cb: (delta: string) => void): () => void
|
||||
}
|
||||
|
||||
/** The result of reading the directive ledger. */
|
||||
|
||||
@@ -12,13 +12,18 @@ export type BacklogState =
|
||||
milestones: GiteaMilestone[]
|
||||
deps: DependencyEdge[]
|
||||
timelines: Record<number, LifecycleEvent[]>
|
||||
/** true while showing the persisted snapshot (instant boot / offline). */
|
||||
stale: boolean
|
||||
/** ISO time the shown snapshot was reconciled, when stale. */
|
||||
savedAt?: string
|
||||
}
|
||||
|
||||
/**
|
||||
* Reconcile the managed repo through the main-process bridge. Runs once on
|
||||
* mount; the returned `refetch` re-reconciles after a write so the board and
|
||||
* forecast reflect the change. `unconfigured` means no token — the UI falls
|
||||
* back to demo fixtures. Errors (network, bad token) surface as `error`.
|
||||
* Reconcile the managed repo through the main-process bridge, stale-while-
|
||||
* revalidate: on mount it shows the persisted snapshot instantly (marked stale),
|
||||
* then a fresh reconcile supersedes it. If gitea is unreachable, the fresh
|
||||
* reconcile falls back to the persisted snapshot (offline). `refetch` re-syncs
|
||||
* after a write. `unconfigured` means no token — the UI uses demo fixtures.
|
||||
*/
|
||||
export function useBacklog(): [BacklogState, () => void] {
|
||||
const [state, setState] = useState<BacklogState>({ status: 'loading' })
|
||||
@@ -27,6 +32,30 @@ export function useBacklog(): [BacklogState, () => void] {
|
||||
|
||||
useEffect(() => {
|
||||
let alive = true
|
||||
|
||||
// instant boot from the persisted snapshot (only on first mount, not refetch)
|
||||
if (nonce === 0) {
|
||||
window.commitea.gitea
|
||||
.boot()
|
||||
.then((b) => {
|
||||
if (!alive || !('cached' in b) || !b.cached) return
|
||||
setState((prev) =>
|
||||
prev.status === 'ready' && !prev.stale
|
||||
? prev // a fresh reconcile already won the race
|
||||
: {
|
||||
status: 'ready',
|
||||
issues: b.issues,
|
||||
milestones: b.milestones,
|
||||
deps: b.deps,
|
||||
timelines: b.timelines,
|
||||
stale: true,
|
||||
savedAt: b.savedAt,
|
||||
},
|
||||
)
|
||||
})
|
||||
.catch(() => {})
|
||||
}
|
||||
|
||||
window.commitea.gitea
|
||||
.reconcile()
|
||||
.then((r) => {
|
||||
@@ -39,12 +68,19 @@ export function useBacklog(): [BacklogState, () => void] {
|
||||
milestones: r.milestones,
|
||||
deps: r.deps,
|
||||
timelines: r.timelines,
|
||||
stale: r.stale ?? false,
|
||||
savedAt: r.savedAt,
|
||||
}
|
||||
: { status: 'unconfigured' },
|
||||
)
|
||||
})
|
||||
.catch((e: unknown) => {
|
||||
if (alive) setState({ status: 'error', message: e instanceof Error ? e.message : String(e) })
|
||||
if (alive) {
|
||||
// keep a shown boot snapshot rather than clobbering it with an error
|
||||
setState((prev) =>
|
||||
prev.status === 'ready' ? prev : { status: 'error', message: e instanceof Error ? e.message : String(e) },
|
||||
)
|
||||
}
|
||||
})
|
||||
return () => {
|
||||
alive = false
|
||||
|
||||
@@ -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<string | null>(null)
|
||||
const [steps, setSteps] = useState<string[]>([])
|
||||
const [proposals, setProposals] = useState<ChangeProposal[]>([])
|
||||
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 }
|
||||
}
|
||||
|
||||
@@ -31,18 +31,20 @@ function stringify(result: unknown): string {
|
||||
}
|
||||
|
||||
export async function runAgentTurn(opts: {
|
||||
complete: (messages: ChatMessage[], tools?: ToolDecl[]) => Promise<CompletionResult>
|
||||
complete: (messages: ChatMessage[], tools?: ToolDecl[], onToken?: (delta: string) => void) => Promise<CompletionResult>
|
||||
messages: ChatMessage[]
|
||||
tools: ToolDecl[]
|
||||
execute: ToolExecutor
|
||||
maxSteps?: number
|
||||
/** Streams content deltas as the model produces prose (final-answer streaming). */
|
||||
onToken?: (delta: string) => void
|
||||
}): Promise<AgentTurn> {
|
||||
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 }
|
||||
}
|
||||
|
||||
@@ -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<Uint8Array>({
|
||||
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)
|
||||
|
||||
@@ -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<CompletionResult>
|
||||
/** When `onToken` is given, the response streams (SSE) and each content delta is emitted. */
|
||||
complete(messages: ChatMessage[], tools?: ToolDecl[], onToken?: OnToken): Promise<CompletionResult>
|
||||
}
|
||||
|
||||
/** 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<string, unknown> = {
|
||||
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<Uint8Array>, onToken: OnToken): Promise<CompletionResult> {
|
||||
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 })) }
|
||||
}
|
||||
|
||||
@@ -31,6 +31,8 @@ export interface GiteaHttpResponse {
|
||||
status: number
|
||||
json(): Promise<unknown>
|
||||
text(): Promise<string>
|
||||
/** Present on the real fetch Response; used for SSE streaming (chat). */
|
||||
body?: ReadableStream<Uint8Array> | null
|
||||
}
|
||||
|
||||
export type FetchLike = (url: string, init?: GiteaRequestInit) => Promise<GiteaHttpResponse>
|
||||
|
||||
@@ -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'
|
||||
|
||||
Reference in New Issue
Block a user