Something went wrong. Try again.
opencode v2 mirror
Something went wrong. Try again.
123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321// Serial prompt queue for direct interactive mode.//// Prompts arrive from the footer (user types and hits enter) and local// operations drain one at a time. Ordinary prompts submitted during an active// ordinary turn are admitted immediately to the server's durable queue.//// The queue also handles local session commands, empty-prompt rejection,// and tracks per-turn wall-clock duration for the footer status line.//// Resolves when the footer closes and all in-flight work finishes.import { SessionMessage } from "@opencode/schema/session-message"import { Locale } from "../util/locale"import { isCompactCommand, isExitCommand, isNewCommand } from "./prompt.shared"import type { FooterApi, FooterEvent, RunDelivery, RunPrompt } from "./types"
type Trace = { write(type: string, data?: unknown): void}
export type QueueInput = { footer: FooterApi initialInput?: string trace?: Trace onSend?: (prompt: RunPrompt, emittedUser: boolean) => void onAdmissionError?: (prompt: RunPrompt, error: unknown) => void | Promise<void> onNewSession?: () => void | Promise<void> onCompact?: () => void | Promise<void> admit: (prompt: RunPrompt, delivery: RunDelivery, signal: AbortSignal) => Promise<void> settle: () => Promise<void> run: (prompt: RunPrompt, signal: AbortSignal, admitted: () => void) => Promise<void>}
type State = { queue: RunPrompt[] active?: RunPrompt admission?: Promise<void> ctrl?: AbortController closed: boolean}
// Runs the prompt queue until the footer closes.//// Subscribes to footer prompt events and drains operations through input.run().// Ordinary prompts submitted during an ordinary active turn are admitted as// durable queued work instead of remaining editable process-local state.export async function runPromptQueue(input: QueueInput): Promise<void> { const stop = Promise.withResolvers<{ type: "closed" }>() const done = Promise.withResolvers<void>() const state: State = { queue: [], closed: input.footer.isClosed, } let draining: Promise<void> | undefined let admissions = Promise.resolve() let admissionVersion = 0 const admissionController = new AbortController()
const emit = (next: FooterEvent, row: Record<string, unknown>) => { input.trace?.write("ui.patch", row) input.footer.event(next) }
const finish = () => { if (!state.closed || draining) { return }
done.resolve() }
const close = () => { if (state.closed) { return }
state.closed = true state.queue.length = 0 // Ordinary turn signals map to session.interrupt; exiting should only detach the TUI. if (state.active?.mode === "shell") state.ctrl?.abort() admissionController.abort() stop.resolve({ type: "closed" }) finish() }
const drain = () => { if (draining || state.closed || state.queue.length === 0) { return }
draining = (async () => { try { while (!state.closed && state.queue.length > 0) { const prompt = state.queue.shift()!
if (prompt.mode !== "shell" && isNewCommand(prompt.text)) { if (!input.onNewSession) { emit( { type: "stream.patch", patch: { status: "new sessions unavailable", }, }, { status: "new sessions unavailable", }, ) continue }
emit( { type: "stream.patch", patch: { phase: "running", status: "starting new session", }, }, { phase: "running", status: "starting new session", }, ) await input.onNewSession() continue }
if (prompt.mode !== "shell" && isCompactCommand(prompt.text)) { emit( { type: "stream.patch", patch: { phase: "running", status: "compacting session", }, }, { phase: "running", status: "compacting session", }, ) await input.onCompact?.() continue }
const sent = prompt.mode === "shell" ? prompt : { ...prompt, messageID: prompt.messageID ?? SessionMessage.ID.create(), } state.active = sent
emit( { type: "turn.send" }, { phase: "running", status: "sending prompt", }, ) const start = Date.now() const ctrl = new AbortController() const admission = Promise.withResolvers<void>() const version = admissionVersion state.ctrl = ctrl state.admission = admission.promise
try { await input.footer.idle() if (state.closed) { break }
input.onSend?.(sent, false)
if (state.closed) { break }
const task = input.run(sent, ctrl.signal, admission.resolve).then( () => ({ type: "done" as const }), (error) => ({ type: "error" as const, error }), )
const next = await Promise.race([task, stop.promise]) if (next.type === "closed") { break }
if (next.type === "error") { throw next.error } if (sent.mode !== "shell" && admissionVersion !== version) { do { const current = admissionVersion await admissions if (state.closed) break await input.settle() if (current === admissionVersion) break } while (!state.closed) } } finally { admission.resolve() if (state.ctrl === ctrl) { state.ctrl = undefined } if (state.admission === admission.promise) state.admission = undefined
if (sent.mode !== "shell") { const duration = Locale.duration(Math.max(0, Date.now() - start)) emit( { type: "turn.duration", duration, }, { duration, }, ) } state.active = undefined } } } catch (error) { done.reject(error) return } finally { draining = undefined emit( { type: "turn.idle" }, { phase: "idle", status: "", }, ) }
finish() })() }
const submit = (prompt: RunPrompt) => { if ( state.closed || (!prompt.text.trim() && (prompt.mode === "shell" || !prompt.parts.some((part) => part.type === "file"))) ) { return }
if (prompt.mode !== "shell" && isExitCommand(prompt.text)) { input.footer.close() return }
const active = state.active if ( active && active.mode !== "shell" && prompt.mode !== "shell" && prompt.command?.source !== "skill" && !isNewCommand(prompt.text) && !isCompactCommand(prompt.text) && !state.queue.some((item) => item.mode !== "shell" && (isNewCommand(item.text) || isCompactCommand(item.text))) ) { const sent = { ...prompt, messageID: SessionMessage.ID.create() } const admission = state.admission admissionVersion += 1 const delivery = prompt.delivery ?? "queue" input.onSend?.(sent, false) admissions = admissions .then(() => admission) .then(() => input.admit(sent, delivery, admissionController.signal)) .catch((error) => (state.closed ? undefined : input.onAdmissionError?.(sent, error))) return }
state.queue.push(prompt) if (prompt.mode !== "shell" && (isNewCommand(prompt.text) || isCompactCommand(prompt.text))) { drain() return }
emit( { type: "first", first: false, }, { first: false, }, ) drain() }
const offPrompt = input.footer.onPrompt((prompt) => { submit(prompt) }) const offClose = input.footer.onClose(() => { close() }) try { if (state.closed) { return }
submit({ text: input.initialInput ?? "", parts: [], }) finish() await done.promise } finally { offPrompt() offClose() close() await draining?.catch(() => {}) await admissions }}