diff --git a/packages/app/src/context/server-session-v2-reducer.test.ts b/packages/app/src/context/server-session-v2-reducer.test.ts index 96100d636759..863872261751 100644 --- a/packages/app/src/context/server-session-v2-reducer.test.ts +++ b/packages/app/src/context/server-session-v2-reducer.test.ts @@ -153,4 +153,88 @@ describe("v2 session reducer", () => { expect(result).toMatchObject({ sessionID: "ses_1", missing: "msg_user", touched: [] }) }) + + test("removes cancelled input from the pending promotion fold", () => { + const reducer = createV2SessionReducer() + reducer.reduce( + [], + event({ + ...base, + id: "evt_admitted", + type: "session.input.admitted", + data: { + sessionID: "ses_1", + inputID: "msg_user", + input: { type: "user", delivery: "queue", data: { text: "cancel me" } }, + }, + }), + ) + reducer.reduce( + [], + event({ + ...base, + id: "evt_cancelled", + type: "session.input.cancelled", + data: { sessionID: "ses_1", inputID: "msg_user" }, + }), + ) + const result = reducer.reduce( + [], + event({ + ...base, + id: "evt_promoted", + type: "session.input.promoted", + data: { sessionID: "ses_1", inputID: "msg_user" }, + }), + ) + + expect(result).toMatchObject({ missing: "msg_user" }) + }) + + test("keeps steered input available to the promotion fold", () => { + const reducer = createV2SessionReducer() + reducer.reduce( + [], + event({ + ...base, + id: "evt_admitted", + type: "session.input.admitted", + data: { + sessionID: "ses_1", + inputID: "msg_user", + input: { type: "user", delivery: "queue", data: { text: "steer me" } }, + }, + }), + ) + reducer.reduce( + [], + event({ + ...base, + id: "evt_steered", + type: "session.input.steered", + data: { sessionID: "ses_1", inputID: "msg_user" }, + }), + ) + reducer.reduce( + [], + event({ + ...base, + id: "evt_queued", + type: "session.input.queued", + data: { sessionID: "ses_1", inputID: "msg_user" }, + }), + ) + + const result = reducer.reduce( + [], + event({ + ...base, + id: "evt_promoted", + type: "session.input.promoted", + data: { sessionID: "ses_1", inputID: "msg_user" }, + }), + ) + + expect(result?.messages).toMatchObject([{ id: "msg_user", type: "user", text: "steer me" }]) + }) }) diff --git a/packages/app/src/context/server-session-v2-reducer.ts b/packages/app/src/context/server-session-v2-reducer.ts index 12f792d2125e..9c4540d53f09 100644 --- a/packages/app/src/context/server-session-v2-reducer.ts +++ b/packages/app/src/context/server-session-v2-reducer.ts @@ -29,6 +29,9 @@ export function createV2SessionReducer() { case "session.input.admitted": pending.set(key(sessionID, event.data.inputID), event.data.input) return result([...source]) + case "session.input.cancelled": + pending.delete(key(sessionID, event.data.inputID)) + return case "session.input.promoted": { const input = pending.get(key(sessionID, event.data.inputID)) pending.delete(key(sessionID, event.data.inputID)) diff --git a/packages/client/src/effect/api/api.ts b/packages/client/src/effect/api/api.ts index a1b457a7e860..551719ec99d3 100644 --- a/packages/client/src/effect/api/api.ts +++ b/packages/client/src/effect/api/api.ts @@ -263,38 +263,52 @@ export type Endpoint5_23Input = { readonly sessionID: Session.ID } export type Endpoint5_23Output = ReadonlyArray export type SessionPendingListOperation = (input: Endpoint5_23Input) => Effect.Effect -export type Endpoint5_24Input = { readonly sessionID: Session.ID } -export type Endpoint5_24Output = ReadonlyArray -export type SessionInstructionsEntryListOperation = ( +export type Endpoint5_24Input = { readonly sessionID: Session.ID; readonly inputID: SessionMessage.ID } +export type Endpoint5_24Output = void +export type SessionPendingCancelOperation = ( input: Endpoint5_24Input, ) => Effect.Effect -export type Endpoint5_25Input = { +export type Endpoint5_25Input = { readonly sessionID: Session.ID; readonly inputID: SessionMessage.ID } +export type Endpoint5_25Output = void +export type SessionPendingSteerOperation = (input: Endpoint5_25Input) => Effect.Effect + +export type Endpoint5_26Input = { readonly sessionID: Session.ID; readonly inputID: SessionMessage.ID } +export type Endpoint5_26Output = void +export type SessionPendingQueueOperation = (input: Endpoint5_26Input) => Effect.Effect + +export type Endpoint5_27Input = { readonly sessionID: Session.ID } +export type Endpoint5_27Output = ReadonlyArray +export type SessionInstructionsEntryListOperation = ( + input: Endpoint5_27Input, +) => Effect.Effect + +export type Endpoint5_28Input = { readonly sessionID: Session.ID readonly key: InstructionEntry.Key readonly value: Schema.Json } -export type Endpoint5_25Output = void +export type Endpoint5_28Output = void export type SessionInstructionsEntryPutOperation = ( - input: Endpoint5_25Input, -) => Effect.Effect + input: Endpoint5_28Input, +) => Effect.Effect -export type Endpoint5_26Input = { readonly sessionID: Session.ID; readonly key: InstructionEntry.Key } -export type Endpoint5_26Output = void +export type Endpoint5_29Input = { readonly sessionID: Session.ID; readonly key: InstructionEntry.Key } +export type Endpoint5_29Output = void export type SessionInstructionsEntryRemoveOperation = ( - input: Endpoint5_26Input, -) => Effect.Effect + input: Endpoint5_29Input, +) => Effect.Effect -export type Endpoint5_27Input = { readonly sessionID: Session.ID; readonly prompt: string } -export type Endpoint5_27Output = { readonly text: string } -export type SessionGenerateOperation = (input: Endpoint5_27Input) => Effect.Effect +export type Endpoint5_30Input = { readonly sessionID: Session.ID; readonly prompt: string } +export type Endpoint5_30Output = { readonly text: string } +export type SessionGenerateOperation = (input: Endpoint5_30Input) => Effect.Effect -export type Endpoint5_28Input = { +export type Endpoint5_31Input = { readonly sessionID: Session.ID readonly after?: Event.Seq | undefined readonly follow?: boolean | undefined } -export type Endpoint5_28Output = +export type Endpoint5_31Output = | ( | { readonly id: Event.ID @@ -404,6 +418,33 @@ export type Endpoint5_28Output = readonly input: SessionPending.Message } } + | { + readonly id: Event.ID + readonly created: DateTime.Utc + readonly metadata?: { readonly [x: string]: unknown } | undefined + readonly type: "session.input.cancelled" + readonly durable: { readonly aggregateID: string; readonly seq: Event.Seq; readonly version: Event.Version } + readonly location?: Location.Ref | undefined + readonly data: { readonly sessionID: Session.ID; readonly inputID: SessionMessage.ID } + } + | { + readonly id: Event.ID + readonly created: DateTime.Utc + readonly metadata?: { readonly [x: string]: unknown } | undefined + readonly type: "session.input.steered" + readonly durable: { readonly aggregateID: string; readonly seq: Event.Seq; readonly version: Event.Version } + readonly location?: Location.Ref | undefined + readonly data: { readonly sessionID: Session.ID; readonly inputID: SessionMessage.ID } + } + | { + readonly id: Event.ID + readonly created: DateTime.Utc + readonly metadata?: { readonly [x: string]: unknown } | undefined + readonly type: "session.input.queued" + readonly durable: { readonly aggregateID: string; readonly seq: Event.Seq; readonly version: Event.Version } + readonly location?: Location.Ref | undefined + readonly data: { readonly sessionID: Session.ID; readonly inputID: SessionMessage.ID } + } | { readonly id: Event.ID readonly created: DateTime.Utc @@ -862,19 +903,19 @@ export type Endpoint5_28Output = } ) | EventLog.Synced -export type SessionLogOperation = (input: Endpoint5_28Input) => Stream.Stream +export type SessionLogOperation = (input: Endpoint5_31Input) => Stream.Stream -export type Endpoint5_29Input = { readonly sessionID: Session.ID } -export type Endpoint5_29Output = void -export type SessionInterruptOperation = (input: Endpoint5_29Input) => Effect.Effect +export type Endpoint5_32Input = { readonly sessionID: Session.ID } +export type Endpoint5_32Output = void +export type SessionInterruptOperation = (input: Endpoint5_32Input) => Effect.Effect -export type Endpoint5_30Input = { readonly sessionID: Session.ID } -export type Endpoint5_30Output = void -export type SessionBackgroundOperation = (input: Endpoint5_30Input) => Effect.Effect +export type Endpoint5_33Input = { readonly sessionID: Session.ID } +export type Endpoint5_33Output = void +export type SessionBackgroundOperation = (input: Endpoint5_33Input) => Effect.Effect -export type Endpoint5_31Input = { readonly sessionID: Session.ID; readonly messageID: SessionMessage.ID } -export type Endpoint5_31Output = SessionMessage.Info -export type SessionMessageOperation = (input: Endpoint5_31Input) => Effect.Effect +export type Endpoint5_34Input = { readonly sessionID: Session.ID; readonly messageID: SessionMessage.ID } +export type Endpoint5_34Output = SessionMessage.Info +export type SessionMessageOperation = (input: Endpoint5_34Input) => Effect.Effect export interface SessionApi { readonly list: SessionListOperation @@ -902,7 +943,12 @@ export interface SessionApi { readonly commit: SessionRevertCommitOperation } readonly context: SessionContextOperation - readonly pending: { readonly list: SessionPendingListOperation } + readonly pending: { + readonly list: SessionPendingListOperation + readonly cancel: SessionPendingCancelOperation + readonly steer: SessionPendingSteerOperation + readonly queue: SessionPendingQueueOperation + } readonly instructions: { readonly entry: { readonly list: SessionInstructionsEntryListOperation diff --git a/packages/client/src/effect/generated/client.ts b/packages/client/src/effect/generated/client.ts index 1b58cec68330..1b1e6f59e6f5 100644 --- a/packages/client/src/effect/generated/client.ts +++ b/packages/client/src/effect/generated/client.ts @@ -80,6 +80,12 @@ import type { Endpoint5_30Output, Endpoint5_31Input, Endpoint5_31Output, + Endpoint5_32Input, + Endpoint5_32Output, + Endpoint5_33Input, + Endpoint5_33Output, + Endpoint5_34Input, + Endpoint5_34Output, Endpoint6_0Input, Endpoint6_0Output, Endpoint7_0Input, @@ -523,37 +529,58 @@ const Endpoint5_23 = (raw: RawClient["server.session"]) => (input: Endpoint5_23I const Endpoint5_24 = (raw: RawClient["server.session"]) => (input: Endpoint5_24Input) => preserveEffect()( - raw["session.instructions.entry.list"]({ params: { sessionID: input["sessionID"] } }).pipe( + raw["session.pending.cancel"]({ params: { sessionID: input["sessionID"], inputID: input["inputID"] } }).pipe( Effect.mapError(mapClientError), - Effect.map((value) => value.data), ), ) const Endpoint5_25 = (raw: RawClient["server.session"]) => (input: Endpoint5_25Input) => preserveEffect()( + raw["session.pending.steer"]({ params: { sessionID: input["sessionID"], inputID: input["inputID"] } }).pipe( + Effect.mapError(mapClientError), + ), + ) + +const Endpoint5_26 = (raw: RawClient["server.session"]) => (input: Endpoint5_26Input) => + preserveEffect()( + raw["session.pending.queue"]({ params: { sessionID: input["sessionID"], inputID: input["inputID"] } }).pipe( + Effect.mapError(mapClientError), + ), + ) + +const Endpoint5_27 = (raw: RawClient["server.session"]) => (input: Endpoint5_27Input) => + preserveEffect()( + raw["session.instructions.entry.list"]({ params: { sessionID: input["sessionID"] } }).pipe( + Effect.mapError(mapClientError), + Effect.map((value) => value.data), + ), + ) + +const Endpoint5_28 = (raw: RawClient["server.session"]) => (input: Endpoint5_28Input) => + preserveEffect()( raw["session.instructions.entry.put"]({ params: { sessionID: input["sessionID"], key: input["key"] }, payload: { value: input["value"] }, }).pipe(Effect.mapError(mapClientError)), ) -const Endpoint5_26 = (raw: RawClient["server.session"]) => (input: Endpoint5_26Input) => - preserveEffect()( +const Endpoint5_29 = (raw: RawClient["server.session"]) => (input: Endpoint5_29Input) => + preserveEffect()( raw["session.instructions.entry.remove"]({ params: { sessionID: input["sessionID"], key: input["key"] } }).pipe( Effect.mapError(mapClientError), ), ) -const Endpoint5_27 = (raw: RawClient["server.session"]) => (input: Endpoint5_27Input) => - preserveEffect()( +const Endpoint5_30 = (raw: RawClient["server.session"]) => (input: Endpoint5_30Input) => + preserveEffect()( raw["session.generate"]({ params: { sessionID: input["sessionID"] }, payload: { prompt: input["prompt"] } }).pipe( Effect.mapError(mapClientError), Effect.map((value) => value.data), ), ) -const Endpoint5_28 = (raw: RawClient["server.session"]) => (input: Endpoint5_28Input) => - preserveStream()( +const Endpoint5_31 = (raw: RawClient["server.session"]) => (input: Endpoint5_31Input) => + preserveStream()( Stream.unwrap( raw["session.log"]({ params: { sessionID: input["sessionID"] }, @@ -565,18 +592,18 @@ const Endpoint5_28 = (raw: RawClient["server.session"]) => (input: Endpoint5_28I ), ) -const Endpoint5_29 = (raw: RawClient["server.session"]) => (input: Endpoint5_29Input) => - preserveEffect()( +const Endpoint5_32 = (raw: RawClient["server.session"]) => (input: Endpoint5_32Input) => + preserveEffect()( raw["session.interrupt"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)), ) -const Endpoint5_30 = (raw: RawClient["server.session"]) => (input: Endpoint5_30Input) => - preserveEffect()( +const Endpoint5_33 = (raw: RawClient["server.session"]) => (input: Endpoint5_33Input) => + preserveEffect()( raw["session.background"]({ params: { sessionID: input["sessionID"] } }).pipe(Effect.mapError(mapClientError)), ) -const Endpoint5_31 = (raw: RawClient["server.session"]) => (input: Endpoint5_31Input) => - preserveEffect()( +const Endpoint5_34 = (raw: RawClient["server.session"]) => (input: Endpoint5_34Input) => + preserveEffect()( raw["session.message"]({ params: { sessionID: input["sessionID"], messageID: input["messageID"] } }).pipe( Effect.mapError(mapClientError), Effect.map((value) => value.data), @@ -605,13 +632,13 @@ const adaptGroup5 = (raw: RawClient["server.session"]) => ({ wait: Endpoint5_18(raw), revert: { stage: Endpoint5_19(raw), clear: Endpoint5_20(raw), commit: Endpoint5_21(raw) }, context: Endpoint5_22(raw), - pending: { list: Endpoint5_23(raw) }, - instructions: { entry: { list: Endpoint5_24(raw), put: Endpoint5_25(raw), remove: Endpoint5_26(raw) } }, - generate: Endpoint5_27(raw), - log: Endpoint5_28(raw), - interrupt: Endpoint5_29(raw), - background: Endpoint5_30(raw), - message: Endpoint5_31(raw), + pending: { list: Endpoint5_23(raw), cancel: Endpoint5_24(raw), steer: Endpoint5_25(raw), queue: Endpoint5_26(raw) }, + instructions: { entry: { list: Endpoint5_27(raw), put: Endpoint5_28(raw), remove: Endpoint5_29(raw) } }, + generate: Endpoint5_30(raw), + log: Endpoint5_31(raw), + interrupt: Endpoint5_32(raw), + background: Endpoint5_33(raw), + message: Endpoint5_34(raw), }) const Endpoint6_0 = (raw: RawClient["server.message"]) => (input: Endpoint6_0Input) => diff --git a/packages/client/src/promise/generated/client.ts b/packages/client/src/promise/generated/client.ts index 6702d36a93f6..2cf0a1a8c789 100644 --- a/packages/client/src/promise/generated/client.ts +++ b/packages/client/src/promise/generated/client.ts @@ -58,6 +58,12 @@ import type { SessionContextOutput, SessionPendingListInput, SessionPendingListOutput, + SessionPendingCancelInput, + SessionPendingCancelOutput, + SessionPendingSteerInput, + SessionPendingSteerOutput, + SessionPendingQueueInput, + SessionPendingQueueOutput, SessionInstructionsEntryListInput, SessionInstructionsEntryListOutput, SessionInstructionsEntryPutInput, @@ -766,6 +772,39 @@ export function make(options: ClientOptions) { }, requestOptions, ).then((value) => value.data), + cancel: (input: SessionPendingCancelInput, requestOptions?: RequestOptions) => + request( + { + method: "DELETE", + path: `/api/session/${encodeURIComponent(input.sessionID)}/pending/${encodeURIComponent(input.inputID)}`, + successStatus: 204, + declaredStatuses: [409, 404, 401, 400], + empty: true, + }, + requestOptions, + ), + steer: (input: SessionPendingSteerInput, requestOptions?: RequestOptions) => + request( + { + method: "POST", + path: `/api/session/${encodeURIComponent(input.sessionID)}/pending/${encodeURIComponent(input.inputID)}/steer`, + successStatus: 204, + declaredStatuses: [409, 404, 401, 400], + empty: true, + }, + requestOptions, + ), + queue: (input: SessionPendingQueueInput, requestOptions?: RequestOptions) => + request( + { + method: "POST", + path: `/api/session/${encodeURIComponent(input.sessionID)}/pending/${encodeURIComponent(input.inputID)}/queue`, + successStatus: 204, + declaredStatuses: [409, 404, 401, 400], + empty: true, + }, + requestOptions, + ), }, instructions: { entry: { diff --git a/packages/client/src/promise/generated/types.ts b/packages/client/src/promise/generated/types.ts index 6263d57a98f4..adaac3e31dfa 100644 --- a/packages/client/src/promise/generated/types.ts +++ b/packages/client/src/promise/generated/types.ts @@ -502,6 +502,36 @@ export type SessionInputPromoted = { data: { sessionID: string; inputID: string } } +export type SessionInputCancelled = { + id: string + created: number + metadata?: { [x: string]: any } + type: "session.input.cancelled" + durable: { aggregateID: string; seq: number; version: 1 } + location?: LocationRef + data: { sessionID: string; inputID: string } +} + +export type SessionInputSteered = { + id: string + created: number + metadata?: { [x: string]: any } + type: "session.input.steered" + durable: { aggregateID: string; seq: number; version: 1 } + location?: LocationRef + data: { sessionID: string; inputID: string } +} + +export type SessionInputQueued = { + id: string + created: number + metadata?: { [x: string]: any } + type: "session.input.queued" + durable: { aggregateID: string; seq: number; version: 1 } + location?: LocationRef + data: { sessionID: string; inputID: string } +} + export type SessionExecutionStarted = { id: string created: number @@ -1970,6 +2000,9 @@ export type SessionEventDurable = | SessionForked | SessionInputPromoted | SessionInputAdmitted + | SessionInputCancelled + | SessionInputSteered + | SessionInputQueued | SessionExecutionStarted | SessionExecutionSucceeded | SessionExecutionFailed @@ -2024,6 +2057,9 @@ export type V2Event = | SessionForked | SessionInputPromoted | SessionInputAdmitted + | SessionInputCancelled + | SessionInputSteered + | SessionInputQueued | SessionExecutionStarted | SessionExecutionSucceeded | SessionExecutionFailed @@ -3689,6 +3725,27 @@ export type SessionPendingListInput = { readonly sessionID: { readonly sessionID export type SessionPendingListOutput = { data: Array }["data"] +export type SessionPendingCancelInput = { + readonly sessionID: { readonly sessionID: string; readonly inputID: string }["sessionID"] + readonly inputID: { readonly sessionID: string; readonly inputID: string }["inputID"] +} + +export type SessionPendingCancelOutput = void + +export type SessionPendingSteerInput = { + readonly sessionID: { readonly sessionID: string; readonly inputID: string }["sessionID"] + readonly inputID: { readonly sessionID: string; readonly inputID: string }["inputID"] +} + +export type SessionPendingSteerOutput = void + +export type SessionPendingQueueInput = { + readonly sessionID: { readonly sessionID: string; readonly inputID: string }["sessionID"] + readonly inputID: { readonly sessionID: string; readonly inputID: string }["inputID"] +} + +export type SessionPendingQueueOutput = void + export type SessionInstructionsEntryListInput = { readonly sessionID: { readonly sessionID: string }["sessionID"] } export type SessionInstructionsEntryListOutput = { data: Array }["data"] diff --git a/packages/client/test/contract-identity.test.ts b/packages/client/test/contract-identity.test.ts index 257f5124f5bc..677cf4d8a065 100644 --- a/packages/client/test/contract-identity.test.ts +++ b/packages/client/test/contract-identity.test.ts @@ -19,7 +19,7 @@ test("effect entrypoint exposes canonical Schema contracts", () => { test("generated Effect API names canonical and composed outputs", async () => { const source = await Bun.file(new URL("../src/effect/api/api.ts", import.meta.url)).text() - expect(source).toContain("export type Endpoint5_3Output = Session.Info") + expect(source).toContain("export type Endpoint5_5Output = Session.Info") expect(source).toContain("export type Endpoint19_0Output = OpenCodeEvent") expect(source).not.toContain("HttpApiClient.ForApi") }) diff --git a/packages/client/test/promise.test.ts b/packages/client/test/promise.test.ts index 14a495cb543f..45dd27b54d1c 100644 --- a/packages/client/test/promise.test.ts +++ b/packages/client/test/promise.test.ts @@ -32,6 +32,7 @@ test("exposes every standard HTTP API group", () => { "projectCopy", "vcs", "debug", + "migration", "websearch", "config", ]) @@ -356,6 +357,28 @@ test("session.pending.list uses the public HTTP contract", async () => { expect(requests).toEqual([{ method: "GET", url: "http://localhost:3000/api/session/ses_test/pending" }]) }) +test("session.pending mutations use the public HTTP contract", async () => { + const requests: Array<{ method: string; url: string }> = [] + const client = OpenCode.make({ + baseUrl: "http://localhost:3000", + fetch: async (input, init) => { + const request = input instanceof Request ? input : new Request(input, init) + requests.push({ method: request.method, url: request.url }) + return new Response(null, { status: 204 }) + }, + }) + + await client.session.pending.cancel({ sessionID: "ses_test", inputID: "msg_cancel" }) + await client.session.pending.steer({ sessionID: "ses_test", inputID: "msg_steer" }) + await client.session.pending.queue({ sessionID: "ses_test", inputID: "msg_queue" }) + + expect(requests).toEqual([ + { method: "DELETE", url: "http://localhost:3000/api/session/ses_test/pending/msg_cancel" }, + { method: "POST", url: "http://localhost:3000/api/session/ses_test/pending/msg_steer/steer" }, + { method: "POST", url: "http://localhost:3000/api/session/ses_test/pending/msg_queue/queue" }, + ]) +}) + test("event.subscribe exposes the Promise event stream wire projection", async () => { const client = OpenCode.make({ baseUrl: "http://localhost:3000", diff --git a/packages/core/src/session.ts b/packages/core/src/session.ts index fc6fc41e0e7f..2fcde1c67715 100644 --- a/packages/core/src/session.ts +++ b/packages/core/src/session.ts @@ -133,6 +133,14 @@ export class CompactionConflictError extends Schema.TaggedErrorClass()("Session.BusyError", { sessionID: SessionSchema.ID, }) {} +export class PendingInputConflictError extends Schema.TaggedErrorClass()( + "Session.PendingInputConflictError", + { + sessionID: SessionSchema.ID, + inputID: SessionMessage.ID, + }, +) {} +type PendingInputRef = { readonly sessionID: SessionSchema.ID; readonly inputID: SessionMessage.ID } export class SkillNotFoundError extends Schema.TaggedErrorClass()("Session.SkillNotFoundError", { skill: Skill.ID, }) {} @@ -181,6 +189,9 @@ export interface Interface { * unhandled compaction barriers. */ readonly pending: (sessionID: SessionSchema.ID) => Effect.Effect + readonly cancelPending: (input: PendingInputRef) => Effect.Effect + readonly steerPending: (input: PendingInputRef) => Effect.Effect + readonly queuePending: (input: PendingInputRef) => Effect.Effect /** * Durable, ordered session log read. Replays durable session bus after * the exclusive `after` cursor, emits a `Synced` marker at the captured @@ -318,6 +329,31 @@ const layer = Layer.effect( ), ) + const pendingConflict = Effect.fn("Session.pendingConflict")(function* (input: PendingInputRef) { + yield* result.get(input.sessionID) + return yield* new PendingInputConflictError(input) + }) + const mutatePending = ( + input: PendingInputRef, + mutation: ( + bus: Bus.Interface, + input: { readonly id: SessionMessage.ID; readonly sessionID: SessionSchema.ID }, + ) => Effect.Effect, + wake = false, + ) => + Effect.uninterruptible( + Effect.gen(function* () { + yield* mutation(bus, { sessionID: input.sessionID, id: input.inputID }).pipe( + Effect.catchDefect((defect) => + defect instanceof SessionPending.LifecycleConflict + ? pendingConflict(input) + : Effect.die(defect), + ), + ) + if (wake) yield* execution.wake(input.sessionID) + }), + ) + const result = Service.of({ create: Effect.fn("Session.create")(function* (input) { const sessionID = input.id ?? SessionSchema.ID.create() @@ -507,6 +543,9 @@ const layer = Layer.effect( yield* result.get(sessionID) return yield* SessionPending.list(db, sessionID) }), + cancelPending: Effect.fn("Session.cancelPending")((input) => mutatePending(input, SessionPending.cancel)), + steerPending: Effect.fn("Session.steerPending")((input) => mutatePending(input, SessionPending.steer, true)), + queuePending: Effect.fn("Session.queuePending")((input) => mutatePending(input, SessionPending.queue)), log: (input) => Stream.unwrap( result diff --git a/packages/core/src/session/message-updater.ts b/packages/core/src/session/message-updater.ts index 51ff426bc46f..d4fbe6ab3eb1 100644 --- a/packages/core/src/session/message-updater.ts +++ b/packages/core/src/session/message-updater.ts @@ -90,6 +90,9 @@ export function update(adapter: Adapter, event: SessionEvent.DurableEvent) { "session.forked": () => Effect.void, "session.input.promoted": () => Effect.void, "session.input.admitted": () => Effect.void, + "session.input.cancelled": () => Effect.void, + "session.input.steered": () => Effect.void, + "session.input.queued": () => Effect.void, "session.execution.started": () => Effect.void, "session.execution.succeeded": () => clearCurrentRetry, "session.execution.failed": () => clearCurrentRetry, diff --git a/packages/core/src/session/pending.ts b/packages/core/src/session/pending.ts index 2b9bdac8be0d..d656564b5152 100644 --- a/packages/core/src/session/pending.ts +++ b/packages/core/src/session/pending.ts @@ -37,6 +37,7 @@ const decodeSynthetic = Schema.decodeUnknownSync(SyntheticData) const encodeSynthetic = Schema.encodeSync(SyntheticData) const decodeMessage = Schema.decodeUnknownSync(SessionMessage.Info) const inboxLocks = KeyedMutex.makeUnsafe() +type PendingRef = { readonly id: SessionMessage.ID; readonly sessionID: SessionSchema.ID } export class LifecycleConflict extends Schema.TaggedErrorClass()( "SessionPending.LifecycleConflict", @@ -294,10 +295,7 @@ export const projectCompactionAdmitted = Effect.fn("SessionPending.projectCompac */ export const projectPromoted = Effect.fn("SessionPending.projectPromoted")(function* ( db: DatabaseService, - input: { - readonly id: SessionMessage.ID - readonly sessionID: SessionSchema.ID - }, + input: PendingRef, ) { if (yield* compaction(db, input.sessionID)) return yield* Effect.die(new LifecycleConflict({ id: input.id })) const deleted = yield* db @@ -312,6 +310,55 @@ export const projectPromoted = Effect.fn("SessionPending.projectPromoted")(funct return stored }) +export const projectCancelled = Effect.fn("SessionPending.projectCancelled")(function* ( + db: DatabaseService, + input: PendingRef, +) { + const deleted = yield* db + .delete(SessionPendingTable) + .where( + and( + eq(SessionPendingTable.id, input.id), + eq(SessionPendingTable.session_id, input.sessionID), + or(eq(SessionPendingTable.delivery, "queue"), eq(SessionPendingTable.delivery, "steer")), + ), + ) + .returning({ id: SessionPendingTable.id }) + .get() + .pipe(Effect.orDie) + if (!deleted) return yield* Effect.die(new LifecycleConflict({ id: input.id })) +}) + +const projectDelivery = Effect.fn("SessionPending.projectDelivery")(function* ( + db: DatabaseService, + input: PendingRef & { readonly from: Delivery; readonly to: Delivery }, +) { + const updated = yield* db + .update(SessionPendingTable) + .set({ delivery: input.to }) + .where( + and( + eq(SessionPendingTable.id, input.id), + eq(SessionPendingTable.session_id, input.sessionID), + eq(SessionPendingTable.delivery, input.from), + ), + ) + .returning({ id: SessionPendingTable.id }) + .get() + .pipe(Effect.orDie) + if (!updated) return yield* Effect.die(new LifecycleConflict({ id: input.id })) +}) + +export const projectSteered = Effect.fn("SessionPending.projectSteered")( + (db: DatabaseService, input: PendingRef) => + projectDelivery(db, { ...input, from: "queue", to: "steer" }), +) + +export const projectQueued = Effect.fn("SessionPending.projectQueued")( + (db: DatabaseService, input: PendingRef) => + projectDelivery(db, { ...input, from: "steer", to: "queue" }), +) + export const settleCompaction = Effect.fn("SessionPending.settleCompaction")(function* ( db: DatabaseService, input: { readonly sessionID: SessionSchema.ID }, @@ -389,6 +436,39 @@ export const equivalent = ( return false } +const publishMutation = (input: PendingRef, effect: Effect.Effect) => + inboxLocks.withLock(input.sessionID)(effect).pipe(Effect.asVoid) + +export const cancel = Effect.fn("SessionPending.cancel")((bus: Bus.Interface, input: PendingRef) => + publishMutation( + input, + bus.publish(SessionEvent.InputCancelled, { + sessionID: input.sessionID, + inputID: input.id, + }), + ), +) + +export const steer = Effect.fn("SessionPending.steer")((bus: Bus.Interface, input: PendingRef) => + publishMutation( + input, + bus.publish(SessionEvent.InputSteered, { + sessionID: input.sessionID, + inputID: input.id, + }), + ), +) + +export const queue = Effect.fn("SessionPending.queue")((bus: Bus.Interface, input: PendingRef) => + publishMutation( + input, + bus.publish(SessionEvent.InputQueued, { + sessionID: input.sessionID, + inputID: input.id, + }), + ), +) + const publish = Effect.fn("SessionPending.publish")(function* ( db: DatabaseService, bus: Bus.Interface, diff --git a/packages/core/src/session/projector.ts b/packages/core/src/session/projector.ts index 5e12004d3e22..7e7fd4a29aa0 100644 --- a/packages/core/src/session/projector.ts +++ b/packages/core/src/session/projector.ts @@ -485,6 +485,24 @@ const layer = Layer.effectDiscard( .pipe(Effect.orDie) }), ) + yield* bus.project(SessionEvent.InputCancelled, (event) => + SessionPending.projectCancelled(db, { + id: event.data.inputID, + sessionID: event.data.sessionID, + }), + ) + yield* bus.project(SessionEvent.InputSteered, (event) => + SessionPending.projectSteered(db, { + id: event.data.inputID, + sessionID: event.data.sessionID, + }), + ) + yield* bus.project(SessionEvent.InputQueued, (event) => + SessionPending.projectQueued(db, { + id: event.data.inputID, + sessionID: event.data.sessionID, + }), + ) yield* bus.project(SessionEvent.Compaction.Admitted, (event) => Effect.gen(function* () { if (event.durable === undefined) diff --git a/packages/core/test/session-prompt.test.ts b/packages/core/test/session-prompt.test.ts index 457175c35c40..40c50c80f37d 100644 --- a/packages/core/test/session-prompt.test.ts +++ b/packages/core/test/session-prompt.test.ts @@ -1086,4 +1086,78 @@ describe("Session.pending", () => { expect(yield* session.pending(sessionID)).toEqual([]) }), ) + + it.effect("cancels pending input and allows its ID to be admitted again", () => + Effect.gen(function* () { + yield* setup + const session = yield* Session.Service + const inputID = SessionMessage.ID.make("msg_cancelled_queue") + yield* session.prompt({ + id: inputID, + sessionID, + text: "Queue this", + delivery: "queue", + resume: false, + }) + + yield* session.cancelPending({ sessionID, inputID }) + + expect(yield* session.pending(sessionID)).toEqual([]) + expect(yield* eventCount(Bus.versionedType(SessionEvent.InputCancelled.type, 1))).toBe(1) + expect( + yield* session.cancelPending({ sessionID, inputID }).pipe(Effect.flip), + ).toMatchObject({ _tag: "Session.PendingInputConflictError", sessionID, inputID }) + expect(yield* eventCount(Bus.versionedType(SessionEvent.InputCancelled.type, 1))).toBe(1) + + const retried = yield* session.prompt({ + id: inputID, + sessionID, + text: "Queue this", + delivery: "queue", + resume: false, + }) + expect(retried).toMatchObject({ id: inputID, delivery: "queue" }) + }), + ) + + it.effect("moves pending input between steer and queue delivery", () => + Effect.gen(function* () { + yield* setup + const session = yield* Session.Service + const queued = yield* session.synthetic({ + sessionID, + text: "Steer this", + delivery: "queue", + resume: false, + }) + const alreadySteered = yield* session.prompt({ sessionID, text: "Already steer", resume: false }) + wakeCalls.length = 0 + + yield* session.steerPending({ sessionID, inputID: queued.id }) + + expect(yield* session.pending(sessionID)).toMatchObject([ + { id: queued.id, delivery: "steer" }, + { id: alreadySteered.id, delivery: "steer" }, + ]) + expect(wakeCalls).toEqual([sessionID]) + expect(yield* eventCount(Bus.versionedType(SessionEvent.InputSteered.type, 1))).toBe(1) + + wakeCalls.length = 0 + yield* session.queuePending({ sessionID, inputID: queued.id }) + expect(yield* session.pending(sessionID)).toMatchObject([ + { id: queued.id, delivery: "queue" }, + { id: alreadySteered.id, delivery: "steer" }, + ]) + expect(wakeCalls).toEqual([]) + expect(yield* eventCount(Bus.versionedType(SessionEvent.InputQueued.type, 1))).toBe(1) + + expect( + yield* session.steerPending({ sessionID, inputID: alreadySteered.id }).pipe(Effect.flip), + ).toMatchObject({ _tag: "Session.PendingInputConflictError", sessionID, inputID: alreadySteered.id }) + yield* session.cancelPending({ sessionID, inputID: alreadySteered.id }) + expect(wakeCalls).toEqual([]) + expect(yield* eventCount(Bus.versionedType(SessionEvent.InputSteered.type, 1))).toBe(1) + expect(yield* eventCount(Bus.versionedType(SessionEvent.InputCancelled.type, 1))).toBe(1) + }), + ) }) diff --git a/packages/protocol/src/groups/session.ts b/packages/protocol/src/groups/session.ts index ec83fa3fdd1d..2380a25f9fcb 100644 --- a/packages/protocol/src/groups/session.ts +++ b/packages/protocol/src/groups/session.ts @@ -522,6 +522,45 @@ export const makeSessionGroup = (sessionLo }), ), ) + .add( + HttpApiEndpoint.delete("session.pending.cancel", "/api/session/:sessionID/pending/:inputID", { + params: { sessionID: Session.ID, inputID: SessionMessage.ID }, + success: HttpApiSchema.NoContent, + error: [ConflictError, SessionNotFoundError], + }).annotateMerge( + OpenApi.annotations({ + identifier: "v2.session.pending.cancel", + summary: "Cancel pending input", + description: "Cancel an input that has not yet been promoted into session history.", + }), + ), + ) + .add( + HttpApiEndpoint.post("session.pending.steer", "/api/session/:sessionID/pending/:inputID/steer", { + params: { sessionID: Session.ID, inputID: SessionMessage.ID }, + success: HttpApiSchema.NoContent, + error: [ConflictError, SessionNotFoundError], + }).annotateMerge( + OpenApi.annotations({ + identifier: "v2.session.pending.steer", + summary: "Steer queued input", + description: "Change a queued input to steer delivery and wake session execution.", + }), + ), + ) + .add( + HttpApiEndpoint.post("session.pending.queue", "/api/session/:sessionID/pending/:inputID/queue", { + params: { sessionID: Session.ID, inputID: SessionMessage.ID }, + success: HttpApiSchema.NoContent, + error: [ConflictError, SessionNotFoundError], + }).annotateMerge( + OpenApi.annotations({ + identifier: "v2.session.pending.queue", + summary: "Queue pending steer", + description: "Change a pending steer to queued delivery.", + }), + ), + ) .add( HttpApiEndpoint.get("session.instructions.entry.list", "/api/session/:sessionID/instructions/entries", { params: { sessionID: Session.ID }, diff --git a/packages/schema/src/session-event.ts b/packages/schema/src/session-event.ts index a47be3194159..2ae3f7c4059c 100644 --- a/packages/schema/src/session-event.ts +++ b/packages/schema/src/session-event.ts @@ -152,13 +152,15 @@ export const Forked = Event.durable({ }) export type Forked = typeof Forked.Type +const InputRef = { + ...Base, + inputID: SessionMessage.ID, +} + export const InputPromoted = Event.durable({ type: "session.input.promoted", ...options, - schema: { - sessionID: SessionID, - inputID: SessionMessage.ID, - }, + schema: InputRef, }) export type InputPromoted = typeof InputPromoted.Type @@ -166,13 +168,33 @@ export const InputAdmitted = Event.durable({ type: "session.input.admitted", ...options, schema: { - ...Base, - inputID: SessionMessage.ID, + ...InputRef, input: SessionPending.Message, }, }) export type InputAdmitted = typeof InputAdmitted.Type +export const InputCancelled = Event.durable({ + type: "session.input.cancelled", + ...options, + schema: InputRef, +}) +export type InputCancelled = typeof InputCancelled.Type + +export const InputSteered = Event.durable({ + type: "session.input.steered", + ...options, + schema: InputRef, +}) +export type InputSteered = typeof InputSteered.Type + +export const InputQueued = Event.durable({ + type: "session.input.queued", + ...options, + schema: InputRef, +}) +export type InputQueued = typeof InputQueued.Type + export namespace Execution { export const Started = Event.durable({ type: "session.execution.started", ...options, schema: Base }) export type Started = typeof Started.Type @@ -580,6 +602,9 @@ export const Definitions = Event.inventory( Forked, InputPromoted, InputAdmitted, + InputCancelled, + InputSteered, + InputQueued, Execution.Started, Execution.Succeeded, Execution.Failed, @@ -621,13 +646,16 @@ export const DurableDefinitions = Event.inventory( ...Definitions.filter((definition) => definition.durability === "durable"), UsageRecorded, ) +export const EphemeralDefinitions = Event.inventory( + ...Definitions.filter((definition) => definition.durability === "ephemeral"), +) export const Durable = Schema.Union(DurableDefinitions, { mode: "oneOf" }) .pipe(Schema.toTaggedUnion("type")) .annotate({ identifier: "Session.Event.Durable" }) export type DurableEvent = typeof Durable.Type -export const All = Schema.Union(Event.inventory(...Definitions, UsageRecorded), { mode: "oneOf" }).pipe( +export const All = Schema.Union([Durable, ...EphemeralDefinitions], { mode: "oneOf" }).pipe( Schema.toTaggedUnion("type"), ) export type Event = typeof All.Type diff --git a/packages/schema/test/event-manifest.test.ts b/packages/schema/test/event-manifest.test.ts index 5f2872f381e5..7896bdc0d671 100644 --- a/packages/schema/test/event-manifest.test.ts +++ b/packages/schema/test/event-manifest.test.ts @@ -84,6 +84,9 @@ describe("public event manifest", () => { "session.forked.2", "session.input.promoted.1", "session.input.admitted.1", + "session.input.cancelled.1", + "session.input.steered.1", + "session.input.queued.1", "session.execution.started.1", "session.execution.succeeded.1", "session.execution.failed.1", diff --git a/packages/server/src/handlers/session.ts b/packages/server/src/handlers/session.ts index f17f9a160e27..ccd145e1036f 100644 --- a/packages/server/src/handlers/session.ts +++ b/packages/server/src/handlers/session.ts @@ -26,6 +26,22 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl Effect.gen(function* () { const session = yield* Session.Service const transfer = yield* SessionTransfer.Service + const pendingMutation = (effect: ReturnType, conflict: string) => + effect.pipe( + Effect.catchTag( + "Session.NotFoundError", + (error) => + new SessionNotFoundError({ + sessionID: error.sessionID, + message: `Session not found: ${error.sessionID}`, + }), + ), + Effect.catchTag( + "Session.PendingInputConflictError", + (error) => new ConflictError({ resource: error.inputID, message: `${conflict}: ${error.inputID}` }), + ), + Effect.as(HttpApiSchema.NoContent.make()), + ) return handlers .handle( @@ -661,6 +677,33 @@ export const SessionHandler = HttpApiBuilder.group(Api, "server.session", (handl } }), ) + .handle( + "session.pending.cancel", + Effect.fn(function* (ctx) { + return yield* pendingMutation( + session.cancelPending({ sessionID: ctx.params.sessionID, inputID: ctx.params.inputID }), + "Pending input can no longer be cancelled", + ) + }), + ) + .handle( + "session.pending.steer", + Effect.fn(function* (ctx) { + return yield* pendingMutation( + session.steerPending({ sessionID: ctx.params.sessionID, inputID: ctx.params.inputID }), + "Pending input is no longer queued", + ) + }), + ) + .handle( + "session.pending.queue", + Effect.fn(function* (ctx) { + return yield* pendingMutation( + session.queuePending({ sessionID: ctx.params.sessionID, inputID: ctx.params.inputID }), + "Pending input is no longer a steer", + ) + }), + ) .handle( "session.instructions.entry.list", Effect.fn(function* (ctx) { diff --git a/packages/tui/src/component/prompt/index.tsx b/packages/tui/src/component/prompt/index.tsx index ae30cad21102..8b0fad398338 100644 --- a/packages/tui/src/component/prompt/index.tsx +++ b/packages/tui/src/component/prompt/index.tsx @@ -53,12 +53,14 @@ import { useLocation } from "../../context/location" import { Keymap, type KeymapCommand } from "../../context/keymap" import { abbreviateHome } from "../../runtime" import { PluginSlot } from "../../plugin/render" +import type { SessionPending } from "@opencode-ai/schema/session-pending" export type PromptProps = { sessionID?: string visible?: boolean disabled?: boolean onSubmit?: () => void + onEmptySubmit?: () => boolean | Promise ref?: (ref: PromptRef | undefined) => void hint?: JSX.Element right?: JSX.Element @@ -357,6 +359,20 @@ export function Prompt(props: PromptProps) { dialog.clear() }, }, + { + title: "Queue prompt", + name: "prompt.queue", + category: "Prompt", + palette: undefined, + run: async (_input: string | undefined, event?: KeyEvent) => { + event?.preventDefault() + event?.stopPropagation() + if (!input.focused) return + const handled = await submit("queue") + if (!handled) return + dialog.clear() + }, + }, { title: "Remove editor context", name: "prompt.editor_context.clear", @@ -515,6 +531,11 @@ export function Prompt(props: PromptProps) { commands: promptCommands(), })) + Keymap.createLayer(() => ({ + priority: 1, + bindings: ["prompt.queue"], + })) + Keymap.createLayer(() => ({ bindings: [ "prompt.submit", @@ -900,7 +921,7 @@ export function Prompt(props: PromptProps) { }) let submitting = false - async function submit() { + async function submit(delivery: SessionPending.Delivery = "steer") { // Prevent overlapping invocations (e.g. a double-pressed Enter, or the // input's native onSubmit racing another dispatch). Without this guard, // a second call slips past the empty-input check before the first call @@ -910,13 +931,13 @@ export function Prompt(props: PromptProps) { if (submitting) return false submitting = true try { - return await submitInner() + return await submitInner(delivery) } finally { submitting = false } } - async function submitInner() { + async function submitInner(delivery: SessionPending.Delivery) { // IME: double-defer may fire before onContentChange flushes the last // composed character (e.g. Korean hangul) to the store, so read // plainText directly and sync before any downstream reads. @@ -927,14 +948,25 @@ export function Prompt(props: PromptProps) { if (props.disabled) return false if (move.creating()) return false if (auto()?.visible) return false - if (!store.prompt.text) return false const trimmed = store.prompt.text.trim() + if (!trimmed) return delivery === "steer" ? (await props.onEmptySubmit?.()) === true : false + if ( + delivery === "queue" && + (store.mode === "shell" || trimmed === "exit" || trimmed === "quit" || trimmed === ":q") + ) { + toast.show({ message: "This prompt cannot be queued", variant: "warning" }) + return false + } if (trimmed === "exit" || trimmed === "quit" || trimmed === ":q") { void exit() return true } const slash = argumentSlash(store.prompt.text, keymapCommands()) if (slash) { + if (delivery === "queue") { + toast.show({ message: "This prompt cannot be queued", variant: "warning" }) + return false + } clearPrompt() await slash.command.run(slash.input) return true @@ -958,6 +990,16 @@ export function Prompt(props: PromptProps) { const isCommand = slashHead !== undefined && (data.location.command.list(currentLocation.ref) ?? []).some((command) => command.name === slashHead.name) + if (delivery === "queue" && isSkill) { + toast.show({ message: "Skills cannot be queued", variant: "warning" }) + return false + } + const editorSelection = editorContext() + const pendingEditorSelection = editorSelection && editor.labelState() === "pending" ? editorSelection : undefined + if (delivery === "queue" && pendingEditorSelection) { + toast.show({ message: "Editor context cannot be queued", variant: "warning" }) + return false + } const agent = local.agent.current() if (!agent) return false const selection = local.model.selection() @@ -1016,8 +1058,6 @@ export function Prompt(props: PromptProps) { // Capture mode before it gets reset const currentMode = store.mode - const editorSelection = editorContext() - const pendingEditorSelection = editorSelection && editor.labelState() === "pending" ? editorSelection : undefined if (store.mode === "shell") { move.startSubmit() @@ -1040,6 +1080,7 @@ export function Prompt(props: PromptProps) { model, files: store.prompt.files, agents: store.prompt.agents, + delivery, }) .catch((error) => { cancelCommit() @@ -1049,7 +1090,7 @@ export function Prompt(props: PromptProps) { move.startSubmit() void client.api.session.skill({ sessionID, - skill: slashHead!.name, + skill: slashHead.name, }) } else { move.startSubmit() @@ -1105,6 +1146,7 @@ export function Prompt(props: PromptProps) { text: inputText, files: store.prompt.files, agents: store.prompt.agents, + delivery, }) .then( () => undefined, diff --git a/packages/tui/src/config/v1/keybind.ts b/packages/tui/src/config/v1/keybind.ts index 0b4ecd94a062..947cd5fee1cc 100644 --- a/packages/tui/src/config/v1/keybind.ts +++ b/packages/tui/src/config/v1/keybind.ts @@ -103,7 +103,8 @@ export const Definitions = { session_interrupt: keybind("escape", "Interrupt current session"), session_background: keybind("ctrl+b", "Background blocking session tools"), session_compact: keybind("c", "Compact the session"), - session_queued_prompts: keybind("q", "View pending work"), + session_queued_prompts: keybind("q", "View queued prompts"), + queued_prompt_delete: keybind("ctrl+d", "Delete queued prompt"), session_child_first: keybind("down", "Toggle subagent picker"), session_parent: keybind("up", "Go to parent session"), session_pin_toggle: keybind("ctrl+f", "Pin or unpin session in the session list"), @@ -161,6 +162,7 @@ export const Definitions = { display_thinking: keybind("none", "Toggle thinking blocks visibility"), prompt_submit: keybind("none", "Submit prompt"), + prompt_queue: keybind("alt+return", "Queue prompt"), prompt_editor_context_clear: keybind("none", "Clear editor context"), prompt_skills: keybind("none", "Open skill selector"), prompt_stash: keybind("none", "Stash prompt"), @@ -170,7 +172,7 @@ export const Definitions = { input_clear: keybind("ctrl+c", "Clear input field"), input_paste: keybind({ key: "ctrl+v", preventDefault: false }, "Paste from clipboard"), input_submit: keybind("return", "Submit input"), - input_newline: keybind("shift+return,ctrl+return,alt+return,ctrl+j", "Insert newline in input"), + input_newline: keybind("shift+return,ctrl+return,ctrl+j", "Insert newline in input"), input_move_left: keybind("left,ctrl+b", "Move cursor left in input"), input_move_right: keybind("right,ctrl+f", "Move cursor right in input"), input_move_up: keybind("up", "Move cursor up in input"), @@ -305,6 +307,7 @@ export const CommandMap = { session_background: "session.background", session_compact: "session.compact", session_queued_prompts: "session.queued_prompts", + queued_prompt_delete: "queued_prompt.delete", session_child_first: "session.child.first", session_parent: "session.parent", session_pin_toggle: "session.pin.toggle", @@ -359,6 +362,7 @@ export const CommandMap = { messages_redo: "session.redo", display_thinking: "session.toggle.thinking", prompt_submit: "prompt.submit", + prompt_queue: "prompt.queue", prompt_editor_context_clear: "prompt.editor_context.clear", prompt_skills: "prompt.skills", prompt_stash: "prompt.stash", diff --git a/packages/tui/src/context/data.tsx b/packages/tui/src/context/data.tsx index c4dcf58a8a7e..51efc9d2a7ab 100644 --- a/packages/tui/src/context/data.tsx +++ b/packages/tui/src/context/data.tsx @@ -38,6 +38,7 @@ import { createStore, produce, reconcile } from "solid-js/store" import { createSimpleContext } from "./helper" import { useClient } from "./client" import { nonEmptyToolContent } from "../util/tool-display" +import type { SessionPending } from "@opencode-ai/schema/session-pending" import { createEffect, createSignal, onCleanup } from "solid-js" export type DataSessionStatus = "idle" | "running" @@ -170,12 +171,20 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ function removePending(sessionID: string, inputID?: string) { if (!inputID) return - setStore( - "session", - "pending", - sessionID, - (store.session.pending[sessionID] ?? []).filter((item) => item.id !== inputID), - ) + if (store.session.pending[sessionID]?.some((item) => item.id === inputID)) + setStore( + "session", + "pending", + sessionID, + (store.session.pending[sessionID] ?? []).filter((item) => item.id !== inputID), + ) + if (store.session.input[sessionID]?.includes(inputID)) + setStore( + "session", + "input", + sessionID, + (store.session.input[sessionID] ?? []).filter((id) => id !== inputID), + ) } function removePermission(sessionID: string, requestID: string) { @@ -189,6 +198,13 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ ) } + function updatePending(sessionID: string, inputID: string, delivery: SessionPending.Delivery) { + const index = store.session.pending[sessionID]?.findIndex((item) => item.id === inputID) ?? -1 + const item = store.session.pending[sessionID]?.[index] + if (index < 0 || !item || item.type === "compaction" || item.delivery === delivery) return + setStore("session", "pending", sessionID, index, { ...item, delivery }) + } + const message = { update(sessionID: string, fn: (messages: SessionMessageInfo[], index: Map) => void) { setStore( @@ -235,6 +251,12 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ (item): item is SessionMessageAssistantReasoning => item.type === "reasoning" && !item.time?.completed, ) }, + reindex(messages: SessionMessageInfo[], index: Map, start: number) { + for (let position = start; position < messages.length; position++) { + const item = messages[position] + if (item) index.set(item.id, position) + } + }, } function index(sessionID: string) { @@ -416,24 +438,36 @@ export const { use: useData, provider: DataProvider } = createSimpleContext({ } break case "session.input.promoted": { + const admitted = store.session.input[event.data.sessionID]?.includes(event.data.inputID) ?? false removePending(event.data.sessionID, event.data.inputID) message.update(event.data.sessionID, (draft, index) => { const position = index.get(event.data.inputID) if (position === undefined) return const existing = draft[position] - if (!existing || !store.session.input[event.data.sessionID]?.includes(event.data.inputID)) return + if (!existing || !admitted) return existing.time.created = event.created draft.splice(position, 1) draft.push(existing) - index.clear() - draft.forEach((message, indexValue) => index.set(message.id, indexValue)) + message.reindex(draft, index, position) }) - setStore( - "session", - "input", - event.data.sessionID, - (store.session.input[event.data.sessionID] ?? []).filter((id) => id !== event.data.inputID), - ) + break + } + case "session.input.steered": + updatePending(event.data.sessionID, event.data.inputID, "steer") + break + case "session.input.queued": + updatePending(event.data.sessionID, event.data.inputID, "queue") + break + case "session.input.cancelled": { + removePending(event.data.sessionID, event.data.inputID) + if (messageIndex.get(event.data.sessionID)?.has(event.data.inputID)) + message.update(event.data.sessionID, (draft, index) => { + const position = index.get(event.data.inputID) + if (position === undefined) return + draft.splice(position, 1) + index.delete(event.data.inputID) + message.reindex(draft, index, position) + }) break } case "session.input.admitted": diff --git a/packages/tui/src/mini/footer.command.tsx b/packages/tui/src/mini/footer.command.tsx index 8564fbc08265..3861fd587dc9 100644 --- a/packages/tui/src/mini/footer.command.tsx +++ b/packages/tui/src/mini/footer.command.tsx @@ -3,7 +3,9 @@ import { TextAttributes, type InputRenderable, type KeyEvent } from "@opentui/co import { useKeyboard, type JSX } from "@opentui/solid" import fuzzysort from "fuzzysort" import { createEffect, createMemo, createSignal, type Accessor } from "solid-js" +import { Keymap } from "../context/keymap" import { RunFooterMenu, createFooterMenuState, type RunFooterMenuItem } from "./footer.menu" +import { monoShortcut } from "./mono" import type { RunFooterTheme } from "./theme" import type { FooterQueuedPrompt, @@ -56,6 +58,10 @@ type SkillEntry = PanelEntry & { name: string } +type QueuedPromptEntry = PanelEntry & { + prompt: FooterQueuedPrompt +} + type SubagentEntry = PanelEntry & { sessionID: string current: boolean @@ -442,7 +448,7 @@ export function RunCommandMenuBody(props: { { action: "queued" as const, category: "Agent", - display: "View pending work", + display: "View queued prompts", footer: `${props.queued().length} pending`, keywords: props .queued() @@ -837,28 +843,48 @@ export function RunQueuedPromptSelectBody(props: { theme: Accessor prompts: Accessor onClose: () => void + onSteer: (prompt: FooterQueuedPrompt) => void + onDelete: (prompt: FooterQueuedPrompt) => void onRows?: (rows: number) => void mono?: boolean }) { - const entries = createMemo(() => + const entries = createMemo(() => props.prompts().map((prompt) => ({ category: "", display: prompt.prompt.text.replaceAll("\n", " "), - footer: prompt.delivery, + footer: "queued", keywords: prompt.prompt.text, + prompt, })), ) const controller = createSearchablePanelController({ entries, limit: SUBAGENT_LIST_ROWS, onClose: props.onClose, - onSelect: props.onClose, + onSelect: (item) => props.onSteer(item.prompt), onRows: props.onRows, }) + const shortcuts = Keymap.useShortcuts() + const deleteShortcut = () => monoShortcut(shortcuts.get("queued_prompt.delete") ?? "", props.mono ?? false) + Keymap.createLayer(() => ({ + priority: 1, + commands: [ + { + id: "queued_prompt.delete", + title: "Delete queued prompt", + group: "Prompt", + run() { + const item = controller.items()[controller.menu.selected()] + if (!item) return false + props.onDelete(item.prompt) + }, + }, + ], + })) return ( mono: Accessor history?: Accessor + queuedPrompts: Accessor + onQueuedPromptSteer: (inputID: string) => Promise onSubmit: (input: RunPrompt) => boolean | Promise onCycle: () => void onInterrupt: () => boolean @@ -980,8 +992,18 @@ export function createPromptState(input: PromptInput): PromptState { })) Keymap.createLayer(() => ({ + priority: 1, enabled: input.prompt() && !visible(), commands: [ + { + id: "prompt.queue", + title: "Queue prompt", + group: "Prompt", + run() { + syncDraft() + submitPrompt(promptCopy(draft), "queue") + }, + }, { id: "prompt.editor", title: "Open editor", @@ -1116,7 +1138,8 @@ export function createPromptState(input: PromptInput): PromptState { } } - const submitPrompt = (next: RunPrompt) => { + let submitting = false + const submitPrompt = (next: RunPrompt, delivery: RunDelivery = "steer") => { if (!area || area.isDestroyed) { draft = promptCopy(next) } @@ -1130,12 +1153,34 @@ export function createPromptState(input: PromptInput): PromptState { hide() } + if (submitting) return + if (!next.text.trim()) { + const queued = delivery === "steer" ? input.queuedPrompts()[0] : undefined + if (queued) { + submitting = true + void input.onQueuedPromptSteer(queued.messageID).finally(() => { + submitting = false + }) + return + } input.onStatus(input.state().phase === "running" ? "waiting for current response" : "empty prompt ignored") return } const command = next.mode === "shell" ? undefined : selectedCommand(next.text, next.command) + if ( + delivery === "queue" && + (next.mode === "shell" || + command?.source === "skill" || + isNewCommand(next.text) || + isCompactCommand(next.text) || + isExitCommand(next.text) || + next.text.trim().toLowerCase() === "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/settings") + ) { + input.onStatus("this prompt cannot be queued") + return + } if (!command && next.mode !== "shell" && isExitCommand(next.text)) { input.onExit() return @@ -1157,24 +1202,28 @@ export function createPromptState(input: PromptInput): PromptState { } const submit = command - ? { ...next, command } + ? { ...next, command, delivery } : parsed?.type === "command" - ? { ...next, command: parsed.command } - : next + ? { ...next, command: parsed.command, delivery } + : { ...next, delivery } const shellMode = next.mode === "shell" + submitting = true resetDraft() queueMicrotask(async () => { - if (await input.onSubmit(submit)) { - push(next) - if (shellMode) { - setShellMode(false) - draft = emptyPrompt(false) + try { + if (await input.onSubmit(submit)) { + push(next) + if (shellMode) { + setShellMode(false) + draft = emptyPrompt(false) + } + return } - return + restore(next) + } finally { + submitting = false } - - restore(next) }) } diff --git a/packages/tui/src/mini/footer.ts b/packages/tui/src/mini/footer.ts index 99c86dd769e6..b9d087146668 100644 --- a/packages/tui/src/mini/footer.ts +++ b/packages/tui/src/mini/footer.ts @@ -51,6 +51,7 @@ import type { MiniSettingChange, MiniSettings, PermissionReply, + QueuedPromptAction, RunAgent, RunCommand, RunInput, @@ -96,6 +97,7 @@ type RunFooterOptions = { onVariantSelect?: (variant: string | undefined) => CycleResult | void | Promise onInterrupt?: () => void onBackground?: () => void + onQueuedPromptAction?: (action: QueuedPromptAction, inputID: string) => Promise onEditorOpen: (input: { value: string }) => Promise onSubagentSelect?: (sessionID: string | undefined) => void onSubagentInterrupt?: (sessionID: string) => void @@ -343,6 +345,7 @@ export class RunFooter implements FooterApi { onCycle: footer.handleCycle, onInterrupt: footer.handleInterrupt, onBackground: options.onBackground, + onQueuedPromptAction: options.onQueuedPromptAction, onEditorOpen: options.onEditorOpen, onInputClear: footer.handleInputClear, onExitRequest: footer.handleExit, diff --git a/packages/tui/src/mini/footer.view.tsx b/packages/tui/src/mini/footer.view.tsx index 0656a83eacc8..2af744f47b75 100644 --- a/packages/tui/src/mini/footer.view.tsx +++ b/packages/tui/src/mini/footer.view.tsx @@ -34,6 +34,8 @@ import { Keymap } from "../context/keymap" import { modelInfo } from "./variant.shared" import { monoShortcut } from "./mono" import { stringWidth } from "../util/string-width" +import { errorMessage } from "../util/error" +import { createSingleFlight } from "../util/single-flight" import type { FooterPromptRoute, @@ -46,6 +48,7 @@ import type { MiniSettingChange, MiniSettings, PermissionReply, + QueuedPromptAction, RunAgent, RunCommand, RunInput, @@ -92,13 +95,14 @@ type RunFooterViewProps = { mono: boolean miniSettings: () => MiniSettings history?: () => RunPrompt[] - onSubmit: (input: RunPrompt) => boolean + onSubmit: (input: RunPrompt) => boolean | Promise onPermissionReply: (input: PermissionReply) => void | Promise onFormReply: (input: FormReply) => void | Promise onFormCancel: (input: FormCancel) => void | Promise onCycle: () => void onInterrupt: () => boolean onBackground?: () => void + onQueuedPromptAction?: (action: QueuedPromptAction, inputID: string) => Promise onEditorOpen: (input: { value: string }) => Promise onInputClear: () => void onExitRequest?: () => boolean @@ -132,6 +136,7 @@ export function RunFooterView(props: RunFooterViewProps) { const [route, setRoute] = createSignal({ type: "composer" }) const [subagentMenuRows, setSubagentMenuRows] = createSignal(RUN_SUBAGENT_PANEL_ROWS) const queuedPrompts = createMemo(() => props.queuedPrompts?.() ?? []) + const queue = createMemo(() => queuedPrompts().filter((item) => item.delivery === "queue")) const skills = createMemo(() => (props.commands() ?? []).filter((item) => item.source === "skill")) const prompt = createMemo(() => active().type === "prompt" && route().type === "composer") const selectingSubagent = createMemo(() => active().type === "prompt" && route().type === "subagent-menu") @@ -229,7 +234,7 @@ export function RunFooterView(props: RunFooterViewProps) { const details = [busy() ? "running" : "idle", `agent ${props.currentAgent()}`] if (current) details.push(variant ? `${current} ${variant}` : current) if (usage()) details.push(props.mono ? usage().replaceAll(" · ", " - ") : usage()) - if (queuedPrompts().length > 0) details.push(`${queuedPrompts().length} pending`) + if (queue().length > 0) details.push(`${queue().length} queued`) if (activeTabs().length > 0) details.push(`${activeTabs().length} subagent${activeTabs().length === 1 ? "" : "s"}`) return details.join(props.mono ? " - " : " · ") }) @@ -309,7 +314,7 @@ export function RunFooterView(props: RunFooterViewProps) { } const openQueuedMenu = () => { - if (queuedPrompts().length === 0) return + if (queue().length === 0) return setRoute({ type: "queued-menu" }) props.onSubagentSelect?.(undefined) } @@ -318,6 +323,22 @@ export function RunFooterView(props: RunFooterViewProps) { setRoute({ type: "composer" }) } + const runQueuedAction = createSingleFlight() + const queuedPromptAction = async (action: QueuedPromptAction, inputID: string) => { + const run = props.onQueuedPromptAction + if (!run) return false + const result = await runQueuedAction(inputID, async () => { + const error = await run(action, inputID).then( + () => undefined, + (error) => error, + ) + if (!error) return true + props.onStatus(`failed to ${action === "cancel" ? "delete" : action} queued prompt: ${errorMessage(error)}`) + return false + }) + return result ?? false + } + const openTab = (sessionID: string) => { setRoute({ type: "subagent", sessionID }) props.onSubagentSelect?.(sessionID) @@ -357,6 +378,8 @@ export function RunFooterView(props: RunFooterViewProps) { theme, mono: () => props.mono, history: props.history, + queuedPrompts: queue, + onQueuedPromptSteer: (inputID) => queuedPromptAction("steer", inputID), onSubmit: props.onSubmit, onCycle: props.onCycle, onInterrupt: props.onInterrupt, @@ -451,13 +474,12 @@ export function RunFooterView(props: RunFooterViewProps) { if (foregroundSubagents() && backgroundShortcut()) { items.push({ key: backgroundShortcut(), label: "background" }) } - if (queuedPrompts().length > 0 && queuedShortcut()) { - items.push({ key: queuedShortcut(), label: `${queuedPrompts().length} pending` }) + if (queue().length > 0 && queuedShortcut()) { + items.push({ key: queuedShortcut(), label: `${queue().length} queued` }) } if (activeTabs().length > 0 && subagentShortcut()) { items.push({ key: subagentShortcut(), label: "subagents" }) } - return items }) const commandHint = createMemo(() => { @@ -568,11 +590,11 @@ export function RunFooterView(props: RunFooterViewProps) { })) Keymap.createLayer(() => ({ - enabled: active().type === "prompt" && route().type === "composer" && queuedPrompts().length > 0, + enabled: active().type === "prompt" && route().type === "composer" && queue().length > 0, commands: [ { id: "session.queued_prompts", - title: "View pending work", + title: "View queued prompts", group: "Session", run: openQueuedMenu, }, @@ -630,7 +652,7 @@ export function RunFooterView(props: RunFooterViewProps) { }) createEffect(() => { - if (route().type !== "queued-menu" || queuedPrompts().length > 0) return + if (route().type !== "queued-menu" || queue().length > 0) return closePanel() }) @@ -734,8 +756,16 @@ export function RunFooterView(props: RunFooterViewProps) { { + void queuedPromptAction("steer", item.messageID).then((steered) => { + if (steered) closePanel() + }) + }} + onDelete={(item) => { + void queuedPromptAction("cancel", item.messageID) + }} onRows={setSubagentMenuRows} mono={props.mono} /> @@ -745,7 +775,7 @@ export function RunFooterView(props: RunFooterViewProps) { theme={theme} commands={props.commands} subagents={tabs} - queued={queuedPrompts} + queued={queue} variants={props.variants} variantCycle={variantCycle()} onClose={closePanel} diff --git a/packages/tui/src/mini/runtime.lifecycle.ts b/packages/tui/src/mini/runtime.lifecycle.ts index d8ef63eb3384..44abfa9c1244 100644 --- a/packages/tui/src/mini/runtime.lifecycle.ts +++ b/packages/tui/src/mini/runtime.lifecycle.ts @@ -22,6 +22,7 @@ import type { MiniSettings, MiniHost, PermissionReply, + QueuedPromptAction, RunAgent, RunInput, RunPrompt, @@ -70,6 +71,7 @@ export type LifecycleInput = { onVariantSelect?: (variant: string | undefined) => CycleResult | void | Promise onInterrupt?: () => void onBackground?: () => void + onQueuedPromptAction?: (action: QueuedPromptAction, inputID: string) => Promise onSubagentSelect?: (sessionID: string | undefined) => void onSubagentInterrupt?: (sessionID: string) => void } @@ -243,6 +245,7 @@ export async function createRuntimeLifecycle(input: LifecycleInput): Promise { if (closed || renderer.isDestroyed) { return diff --git a/packages/tui/src/mini/runtime.queue.ts b/packages/tui/src/mini/runtime.queue.ts index 75894237e3e8..8ff308b7e1f8 100644 --- a/packages/tui/src/mini/runtime.queue.ts +++ b/packages/tui/src/mini/runtime.queue.ts @@ -11,7 +11,7 @@ import { SessionMessage } from "@opencode-ai/schema/session-message" import { Locale } from "../util/locale" import { isCompactCommand, isExitCommand, isNewCommand } from "./prompt.shared" -import type { FooterApi, FooterEvent, RunPrompt } from "./types" +import type { FooterApi, FooterEvent, RunDelivery, RunPrompt } from "./types" type Trace = { write(type: string, data?: unknown): void @@ -21,11 +21,11 @@ export type QueueInput = { footer: FooterApi initialInput?: string trace?: Trace - onSend?: (prompt: RunPrompt, delivery: "steer" | "queue") => void + onSend?: (prompt: RunPrompt, delivery: RunDelivery) => void onAdmissionError?: (prompt: RunPrompt, error: unknown) => void | Promise onNewSession?: () => void | Promise onCompact?: () => void | Promise - admit: (prompt: RunPrompt, signal: AbortSignal) => Promise + admit: (prompt: RunPrompt, delivery: RunDelivery, signal: AbortSignal) => Promise settle: () => Promise run: (prompt: RunPrompt, signal: AbortSignal, admitted: () => void) => Promise } @@ -183,7 +183,7 @@ export async function runPromptQueue(input: QueueInput): Promise { input.trace?.write("ui.commit", commit) input.footer.append(commit) } - input.onSend?.(sent, "steer") + input.onSend?.(sent, sent.delivery ?? "steer") if (state.closed) { break @@ -276,10 +276,11 @@ export async function runPromptQueue(input: QueueInput): Promise { const sent = { ...prompt, messageID: SessionMessage.ID.create() } const admission = state.admission admissionVersion += 1 - input.onSend?.(sent, "queue") + const delivery = prompt.delivery ?? "queue" + input.onSend?.(sent, delivery) admissions = admissions .then(() => admission) - .then(() => input.admit(sent, admissionController.signal)) + .then(() => input.admit(sent, delivery, admissionController.signal)) .catch((error) => (state.closed ? undefined : input.onAdmissionError?.(sent, error))) return } diff --git a/packages/tui/src/mini/runtime.ts b/packages/tui/src/mini/runtime.ts index 041d3bc28cc6..b72c0e0457e4 100644 --- a/packages/tui/src/mini/runtime.ts +++ b/packages/tui/src/mini/runtime.ts @@ -390,6 +390,15 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep log?.write("send.background", { sessionID: state.sessionID }) void state.sdk.session.background({ sessionID: state.sessionID }).catch(() => {}) }, + onQueuedPromptAction: async (action, inputID) => { + if (!state.sessionID) return + log?.write(`send.pending.${action}`, { sessionID: state.sessionID, inputID }) + if (action === "steer") { + await state.sdk.session.pending.steer({ sessionID: state.sessionID, inputID }) + return + } + await state.sdk.session.pending.cancel({ sessionID: state.sessionID, inputID }) + }, onSubagentInterrupt: (sessionID) => { log?.write("send.subagent.interrupt", { sessionID }) void state.sdk.session.interrupt({ sessionID }).catch(() => {}) @@ -892,7 +901,7 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep trace: log, onSend: (prompt, delivery) => { state.shown = true - state.history.push(prompt) + state.history.push({ ...prompt, delivery: undefined }) if (prompt.mode !== "shell" && delivery === "steer") { rememberLocal({ kind: "user", @@ -903,18 +912,21 @@ async function runInteractiveRuntime(input: RunRuntimeInput, deps: RunRuntimeDep }) } }, - admit: async (prompt, signal) => { + admit: async (prompt, delivery, signal) => { await state.switching?.catch(() => {}) const next = await ensureStream() - await next.handle.queuePromptTurn({ - agent: state.agent, - model: state.model, - variant: state.activeVariant, - prompt, - files: input.files, - includeFiles: false, - signal, - }) + await next.handle.admitPromptTurn( + { + agent: state.agent, + model: state.model, + variant: state.activeVariant, + prompt, + files: input.files, + includeFiles: false, + signal, + }, + delivery, + ) }, onAdmissionError: renderPromptError, onCompact: async () => { diff --git a/packages/tui/src/mini/stream-v2.subagent.ts b/packages/tui/src/mini/stream-v2.subagent.ts index d1ff8556d01b..2dc2c76d8b91 100644 --- a/packages/tui/src/mini/stream-v2.subagent.ts +++ b/packages/tui/src/mini/stream-v2.subagent.ts @@ -653,6 +653,10 @@ export function createSubagentTracker(input: SubagentTrackerInput): SubagentTrac } return } + if (event.type === "session.input.cancelled") { + child.prompts.delete(event.data.inputID) + return + } if (event.type === "session.step.started") { touch(child, event.created) if (child.label === FALLBACK_LABEL && event.data.agent) child.label = Locale.titlecase(event.data.agent) diff --git a/packages/tui/src/mini/stream-v2.transport.ts b/packages/tui/src/mini/stream-v2.transport.ts index ad249c6eb5e4..ea4a15e978e1 100644 --- a/packages/tui/src/mini/stream-v2.transport.ts +++ b/packages/tui/src/mini/stream-v2.transport.ts @@ -26,6 +26,7 @@ import type { FooterQueuedPrompt, RunFilePart, RunInput, + RunDelivery, RunPrompt, RunPromptPart, StreamCommit, @@ -71,7 +72,7 @@ export type SessionResizeReplayInput = { export type SessionTransport = { runPromptTurn(input: SessionTurnInput, admitted?: () => void): Promise - queuePromptTurn(input: SessionTurnInput): Promise + admitPromptTurn(input: SessionTurnInput, delivery: RunDelivery): Promise waitForIdle(): Promise interruptActiveTurn(): Promise selectSubagent(sessionID: string | undefined): void @@ -515,8 +516,12 @@ export async function createSessionTransport(input: StreamInput): Promise { - const prompts = [...state.pending.values()] + const prompts = [...state.pending.values()].filter((item) => item.delivery === "queue") + const ids = prompts.map((item) => item.messageID) + if (syncedPending?.length === ids.length && syncedPending.every((id, index) => id === ids[index])) return + syncedPending = ids input.trace?.write("ui.patch", { pending: prompts.length }) input.footer.event({ type: "queued.prompts", prompts }) } @@ -934,6 +939,36 @@ export async function createSessionTransport(input: StreamInput): Promise | undefined - const admitPrompt = async (next: SessionTurnInput, client: OpenCodeClient, delivery: "steer" | "queue") => { + const admitPrompt = async (next: SessionTurnInput, client: OpenCodeClient, delivery: RunDelivery) => { const messageID = next.prompt.messageID if (!messageID) throw new Error("Prompt message ID is required") const command = next.prompt.command @@ -1643,14 +1678,20 @@ export async function createSessionTransport(input: StreamInput): Promise admitPrompt(next, client, "steer"), admitted) + await runTurnWait(next, messageID, client, () => admitPrompt(next, client, next.prompt.delivery ?? "steer"), admitted) return } @@ -1700,7 +1741,7 @@ export async function createSessionTransport(input: StreamInput): Promise admitPrompt(next, client, "steer"), admitted) + await runTurnWait(next, messageID, client, () => admitPrompt(next, client, next.prompt.delivery ?? "steer"), admitted) }, async interruptActiveTurn() { // A running shell holds no drain, so session.interrupt cannot reach it; diff --git a/packages/tui/src/mini/types.ts b/packages/tui/src/mini/types.ts index 001ec9260524..17dcc601fe08 100644 --- a/packages/tui/src/mini/types.ts +++ b/packages/tui/src/mini/types.ts @@ -23,6 +23,7 @@ import type { } from "@opencode-ai/client/promise" import type { Config } from "../config" import type { CliRenderer } from "@opentui/core" +import type { SessionPending } from "@opencode-ai/schema/session-pending" export type RunFilePart = { type: "file" @@ -71,10 +72,13 @@ export type RunProvider = { models: Record } +export type RunDelivery = SessionPending.Delivery + export type RunPrompt = { messageID?: string text: string parts: RunPromptPart[] + delivery?: RunDelivery mode?: "shell" command?: { name: string @@ -87,9 +91,11 @@ export type RunPrompt = { export type FooterQueuedPrompt = { messageID: string prompt: RunPrompt - delivery: "steer" | "queue" + delivery: RunDelivery } +export type QueuedPromptAction = "steer" | "cancel" + export type RunAgent = { id: string name: string diff --git a/packages/tui/src/routes/session/index.tsx b/packages/tui/src/routes/session/index.tsx index 52c790e8db4f..ade1ac548003 100644 --- a/packages/tui/src/routes/session/index.tsx +++ b/packages/tui/src/routes/session/index.tsx @@ -52,6 +52,7 @@ import { useClient } from "../../context/client" import { useEditorContext } from "../../context/editor" import { openEditor } from "../../editor" import { useDialog } from "../../ui/dialog" +import { DialogSelect } from "../../ui/dialog-select" import { DialogSessionRename } from "../../component/dialog-session-rename" import { DialogMessage } from "./dialog-message" import { DialogFork } from "./dialog-fork" @@ -97,6 +98,8 @@ import { stringWidth } from "../../util/string-width" import { useArgs } from "../../context/args" import { withTimestampedFallback } from "@opencode-ai/util/session-title-fallback" import { useSessionTabs } from "../../context/session-tabs" +import { createSingleFlight } from "../../util/single-flight" +import type { SessionPending } from "@opencode-ai/schema/session-pending" addDefaultParsers(parsers.parsers) @@ -109,6 +112,7 @@ const NAVIGATION_SLACK_ID = "session-navigation-slack" const TRANSCRIPT_TAIL_ROWS = 40 const TRANSCRIPT_BACKFILL_CHUNK = 60 const TRANSCRIPT_BACKFILL_DELAY = 120 +type PendingAction = "steer" | "queue" | "cancel" const context = createContext<{ width: number @@ -120,6 +124,8 @@ const context = createContext<{ diffWrapMode: () => "word" | "none" models: () => ModelInfo[] config: ReturnType["data"] + mutatePending: (action: PendingAction, inputID: string) => Promise + pendingDelivery: (inputID: string) => SessionPending.Delivery | undefined }>() function use() { @@ -175,6 +181,13 @@ export function Session() { .flatMap((sessionID) => data.session.form.list(sessionID) ?? []) .concat(global) }) + const pendingUsers = createMemo(() => + data.session.pending.list(route.sessionID).flatMap((item) => (item.type === "user" ? [item] : [])), + ) + const pendingDeliveries = createMemo(() => new Map(pendingUsers().map((item) => [item.id, item.delivery]))) + const queuedPrompts = createMemo(() => + pendingUsers().flatMap((item) => (item.delivery === "queue" ? [{ id: item.id, text: item.data.text }] : [])), + ) const [composer, setComposer] = createStore({ open: false, tab: undefined as string | undefined, @@ -369,6 +382,55 @@ export function Session() { }) const dialog = useDialog() const renderer = useRenderer() + const runPendingAction = createSingleFlight() + const mutatePending = async (action: PendingAction, inputID: string) => { + const result = await runPendingAction(inputID, async () => { + const request = + action === "steer" + ? client.api.session.pending.steer({ sessionID: route.sessionID, inputID }) + : action === "queue" + ? client.api.session.pending.queue({ sessionID: route.sessionID, inputID }) + : client.api.session.pending.cancel({ sessionID: route.sessionID, inputID }) + const error = await request.then( + () => undefined, + (error) => error, + ) + if (!error) return true + const label = action === "cancel" ? "delete" : action + toast.show({ title: `Failed to ${label} pending prompt`, message: errorMessage(error), variant: "error" }) + return false + }) + return result ?? false + } + const openQueuedPrompts = () => + dialog.replace(() => ( + ({ + title: prompt.text, + value: prompt.id, + footer: `${index + 1} of ${queuedPrompts().length}`, + }))} + onSelect={(option) => { + void mutatePending("steer", option.value).then((steered) => { + if (steered) dialog.clear() + }) + }} + actions={[ + { + command: "queued_prompt.delete", + title: "delete", + onTrigger: (option) => { + const last = queuedPrompts().length === 1 + void mutatePending("cancel", option.value).then((cancelled) => { + if (cancelled && last) dialog.clear() + }) + }, + }, + ]} + footerHints={[{ title: "steer", label: "enter" }]} + /> + )) const unavailable = (feature: string) => { toast.show({ message: `${feature} is not implemented for V2 sessions yet`, variant: "error", duration: 5000 }) dialog.clear() @@ -871,6 +933,13 @@ export function Session() { dialog.clear() }, }, + { + title: "View queued prompts", + id: "session.queued_prompts", + group: "Session", + enabled: queuedPrompts().length > 0, + run: openQueuedPrompts, + }, { title: "Go to parent session", id: "session.parent", @@ -942,6 +1011,8 @@ export function Session() { diffWrapMode, models, config, + mutatePending, + pendingDelivery: (inputID) => pendingDeliveries().get(inputID), }} > @@ -997,6 +1068,9 @@ export function Session() { + 0}> + + { toBottom() }} + onEmptySubmit={async () => { + const next = queuedPrompts()[0] + if (!next) return false + return mutatePending("steer", next.id) + }} sessionID={route.sessionID} /> @@ -1813,6 +1892,7 @@ function ShellMessage(props: { message: Extract local.agent.color(data.session.get(ctx.sessionID)?.agent ?? "build")) - const queued = createMemo( - () => data.session.status(ctx.sessionID) === "running" && data.session.input.has(ctx.sessionID, props.message.id), - ) + const delivery = createMemo(() => ctx.pendingDelivery(props.message.id)) const dialog = useDialog() const renderer = useRenderer() const promptRef = usePromptRef() + const updatePendingSteer = async (action: "queue" | "cancel") => { + if (await ctx.mutatePending(action, props.message.id)) dialog.clear() + } + return ( { if (renderer.getSelection()?.getSelectedText()) return + if (delivery() === "steer") { + dialog.replace(() => ( + { + void updatePendingSteer(option.value) + }} + /> + )) + return + } dialog.replace(() => ( void }) { + const theme = useTheme("elevated") + const next = createMemo(() => props.prompts[0]?.text) + + return ( + + + + {props.prompts.length} queued + {(text) => <> · {text()}} + + + + ) +} + function AssistantRetry(props: { retry: SessionMessageAssistant["retry"] }) { const theme = useTheme() return ( diff --git a/packages/tui/src/routes/session/rows.ts b/packages/tui/src/routes/session/rows.ts index 14ffe29d21dc..bc39f1c00f14 100644 --- a/packages/tui/src/routes/session/rows.ts +++ b/packages/tui/src/routes/session/rows.ts @@ -46,9 +46,14 @@ export function createSessionRows(sessionID: Accessor) { function reduce() { const messages = data.session.message.list(sessionID()) const inputs = new Set(data.session.input.list(sessionID())) + const pending = data.session.pending.list(sessionID()) + const queued = new Set( + pending.flatMap((item) => (item.type === "user" && item.delivery === "queue" ? [item.id] : [])), + ) + const visible = queued.size === 0 ? messages : messages.filter((message) => !queued.has(message.id)) const boundary = revertBoundary() const rows = reduceSessionRows( - boundary ? messages.filter((message) => message.id < boundary) : messages, + boundary ? visible.filter((message) => message.id < boundary) : visible, inputs, turnTokens(), ) @@ -57,8 +62,7 @@ export function createSessionRows(sessionID: Accessor) { rows.splice( position === -1 ? rows.length : position, 0, - ...data.session.pending - .list(sessionID()) + ...pending .filter((item) => item.type === "compaction") .map((item): SessionRow => ({ type: "compaction-queued", inputID: item.id })), ) @@ -112,10 +116,11 @@ export function createSessionRows(sessionID: Accessor) { createEffect( on( () => - data.session.pending - .list(sessionID()) - .filter((item) => item.type === "compaction") - .map((item) => item.id), + data.session.pending.list(sessionID()).flatMap((item) => { + if (item.type === "compaction") return [`${item.id}:compaction`] + if (item.type === "user" && item.delivery === "queue") return [`${item.id}:queue`] + return [] + }), () => setRows(reconcile(reduce())), { defer: true }, ), @@ -196,7 +201,9 @@ export function createSessionRows(sessionID: Accessor) { const queuedStart = (rows: SessionRow[]) => { const index = rows.findIndex( - (row) => row.type === "compaction-queued" || (row.type === "message" && isPending(row.messageID)), + (row) => + row.type === "compaction-queued" || + (row.type === "message" && isPending(row.messageID)), ) return index === -1 ? rows.length : index } diff --git a/packages/tui/src/util/single-flight.ts b/packages/tui/src/util/single-flight.ts new file mode 100644 index 000000000000..09a56178f6b0 --- /dev/null +++ b/packages/tui/src/util/single-flight.ts @@ -0,0 +1,12 @@ +export function createSingleFlight() { + const pending = new Set() + return async (key: Key, run: () => Promise) => { + if (pending.has(key)) return + pending.add(key) + try { + return await run() + } finally { + pending.delete(key) + } + } +} diff --git a/packages/tui/test/cli/tui/data.test.tsx b/packages/tui/test/cli/tui/data.test.tsx index 5278ae4c5f1f..f6d1451d5ac5 100644 --- a/packages/tui/test/cli/tui/data.test.tsx +++ b/packages/tui/test/cli/tui/data.test.tsx @@ -914,6 +914,106 @@ test("completes exploration when a queued prompt is promoted", async () => { } }) +test("updates and removes queued inputs from durable lifecycle events", async () => { + const events = createEventStream() + const sessionID = "session-queue-management" + const calls = createFetch((url) => { + if (url.pathname === `/api/session/${sessionID}/message`) return json({ data: [], cursor: {} }) + }, events) + let data!: ReturnType + let rows!: ReturnType + let client!: ReturnType + + function Probe() { + client = useClient() + data = useData() + rows = createSessionRows(() => sessionID) + return + } + + const app = await testRender(() => ( + + + + + + + + + + )) + + try { + await wait(() => client.connection.status() === "connected") + emitEvent(events, { + id: "evt_queue_admitted", + created: 1, + type: "session.input.admitted", + durable: durable(sessionID), + data: { + sessionID, + inputID: "message-queued", + input: { type: "user", data: { text: "Steer me" }, delivery: "queue" }, + }, + }) + await wait(() => data.session.pending.list(sessionID).length === 1) + expect(rows).not.toContainEqual({ type: "message", messageID: "message-queued" }) + + emitEvent(events, { + id: "evt_queue_steered", + created: 2, + type: "session.input.steered", + durable: durable(sessionID, 1), + data: { sessionID, inputID: "message-queued" }, + }) + await wait(() => + data.session.pending + .list(sessionID) + .some((item) => item.id === "message-queued" && item.type !== "compaction" && item.delivery === "steer"), + ) + expect(rows).toContainEqual({ type: "message", messageID: "message-queued" }) + + emitEvent(events, { + id: "evt_queue_restored", + created: 3, + type: "session.input.queued", + durable: durable(sessionID, 2), + data: { sessionID, inputID: "message-queued" }, + }) + await wait(() => + data.session.pending + .list(sessionID) + .some((item) => item.id === "message-queued" && item.type !== "compaction" && item.delivery === "queue"), + ) + expect(rows).not.toContainEqual({ type: "message", messageID: "message-queued" }) + + emitEvent(events, { + id: "evt_cancel_admitted", + created: 4, + type: "session.input.admitted", + durable: durable(sessionID, 3), + data: { + sessionID, + inputID: "message-cancelled", + input: { type: "user", data: { text: "Delete me" }, delivery: "queue" }, + }, + }) + await wait(() => data.session.pending.list(sessionID).length === 2) + emitEvent(events, { + id: "evt_queue_cancelled", + created: 5, + type: "session.input.cancelled", + durable: durable(sessionID, 4), + data: { sessionID, inputID: "message-cancelled" }, + }) + await wait(() => !data.session.input.has(sessionID, "message-cancelled")) + expect(data.session.pending.list(sessionID).map((item) => item.id)).toEqual(["message-queued"]) + expect(data.session.message.get(sessionID, "message-cancelled")).toBeUndefined() + } finally { + app.renderer.destroy() + } +}) + test("classifies live tool rows independently of their call ID", async () => { const events = createEventStream() const sessionID = "session-tool-call-id" diff --git a/packages/tui/test/mini/fixture/footer-api.ts b/packages/tui/test/mini/fixture/footer-api.ts index 83c06ca50fce..61e3ca960490 100644 --- a/packages/tui/test/mini/fixture/footer-api.ts +++ b/packages/tui/test/mini/fixture/footer-api.ts @@ -56,9 +56,9 @@ export function createFooterApiFixture(input: { events?: FooterEvent[]; commits? commits, calls, promptReady, - submit(text: string, mode?: RunPrompt["mode"]) { + submit(text: string, mode?: RunPrompt["mode"], delivery?: RunPrompt["delivery"]) { if (prompts.size === 0) return false - const prompt: RunPrompt = mode ? { text, parts: [], mode } : { text, parts: [] } + const prompt: RunPrompt = { text, parts: [], ...(mode ? { mode } : {}), ...(delivery ? { delivery } : {}) } for (const fn of [...prompts]) fn(prompt) return true }, diff --git a/packages/tui/test/mini/footer.view.test.tsx b/packages/tui/test/mini/footer.view.test.tsx index 50e81d168d64..efff1391cb52 100644 --- a/packages/tui/test/mini/footer.view.test.tsx +++ b/packages/tui/test/mini/footer.view.test.tsx @@ -21,6 +21,7 @@ import { RunFooterView } from "../../src/mini/footer.view" import { RunEntryContent } from "../../src/mini/scrollback.writer" import { RUN_THEME_FALLBACK, type RunTheme } from "../../src/mini/theme" import type { + FooterQueuedPrompt, FooterState, FooterSubagentState, FooterSubagentTab, @@ -120,13 +121,15 @@ async function renderFooter( height?: number state?: Partial onCycle?: () => void - onSubmit?: (prompt: RunPrompt) => boolean + onSubmit?: (prompt: RunPrompt) => boolean | Promise view?: FooterView onFormReply?: (input: unknown) => void miniSettings?: MiniSettings mono?: boolean onStatus?: (status: string) => void onMiniSettingChange?: (change: MiniSettingChange) => void + queuedPrompts?: FooterQueuedPrompt[] + onQueuedPromptAction?: (action: "steer" | "cancel", inputID: string) => Promise } = {}, ) { const [view, setView] = createSignal(input.view ?? { type: "prompt" }) @@ -164,6 +167,7 @@ async function renderFooter( state={state} view={view} subagent={subagents} + queuedPrompts={() => input.queuedPrompts ?? []} theme={input.theme ?? (() => RUN_THEME_FALLBACK)} mono={input.mono ?? false} miniSettings={miniSettings} @@ -173,6 +177,7 @@ async function renderFooter( onFormCancel={() => {}} onCycle={input.onCycle ?? (() => {})} onInterrupt={() => false} + onQueuedPromptAction={input.onQueuedPromptAction} onEditorOpen={async () => undefined} onInputClear={() => {}} onExit={() => {}} @@ -913,7 +918,7 @@ test("direct subagent panel closes when moving up from the first item", async () } }) -test("direct pending panel shows durable delivery without edit actions", async () => { +test("direct queued panel steers and deletes selected prompts", async () => { const [prompts] = createSignal([ { messageID: "m-1", @@ -921,16 +926,22 @@ test("direct pending panel shows durable delivery without edit actions", async ( delivery: "queue" as const, }, ]) + const steered: string[] = [] + const deleted: string[] = [] const app = await testRender( () => ( - - RUN_THEME_FALLBACK.footer} - prompts={prompts} - onClose={() => {}} - /> - + + + RUN_THEME_FALLBACK.footer} + prompts={prompts} + onClose={() => {}} + onSteer={(prompt) => steered.push(prompt.messageID)} + onDelete={(prompt) => deleted.push(prompt.messageID)} + /> + + ), { width: 100, height: RUN_SUBAGENT_PANEL_ROWS }, ) @@ -940,19 +951,98 @@ test("direct pending panel shows durable delivery without edit actions", async ( const frame = app.captureCharFrame() const list = panelMenu(app.renderer.root) - expect(frame).toContain("Pending work") + expect(frame).toContain("Queued prompts") expect(frame).toContain("fix the auth test") - expect(frame).toContain("queue") + expect(frame).toContain("queued") + expect(frame).toContain("enter steer · ctrl+d delete") expect(frame).not.toContain("┌") expect(frame).not.toContain("┃") expectPaletteList(list, 0) - expect(frame).not.toContain("edit") - expect(frame).not.toContain("remove") + app.mockInput.pressEnter() + app.mockInput.pressKey("d", { ctrl: true }) + expect(steered).toEqual(["m-1"]) + expect(deleted).toEqual(["m-1"]) } finally { app.renderer.destroy() } }) +test("direct footer steers the oldest queued prompt from an empty composer", async () => { + const steered: string[] = [] + const app = await renderFooter({ + queuedPrompts: [ + { messageID: "m-1", prompt: { text: "first", parts: [] }, delivery: "queue" }, + { messageID: "m-2", prompt: { text: "second", parts: [] }, delivery: "queue" }, + ], + onQueuedPromptAction: async (action, inputID) => { + if (action === "steer") steered.push(inputID) + }, + }) + + try { + await app.renderOnce() + app.mockInput.pressEnter({ meta: true }) + await Bun.sleep(0) + expect(steered).toEqual([]) + app.mockInput.pressEnter() + await Bun.sleep(0) + expect(steered).toEqual(["m-1"]) + } finally { + app.cleanup() + } +}) + +test("direct footer does not steer queued work on a double submit", async () => { + const submitted: RunPrompt[] = [] + const steered: string[] = [] + const app = await renderFooter({ + queuedPrompts: [{ messageID: "m-1", prompt: { text: "queued", parts: [] }, delivery: "queue" }], + onSubmit: async (prompt) => { + submitted.push(prompt) + await Bun.sleep(10) + return true + }, + onQueuedPromptAction: async (action, inputID) => { + if (action === "steer") steered.push(inputID) + }, + }) + + try { + await app.renderOnce() + await app.mockInput.typeText("send once") + app.mockInput.pressEnter() + app.mockInput.pressEnter() + await Bun.sleep(20) + expect(submitted).toHaveLength(1) + expect(steered).toEqual([]) + } finally { + app.cleanup() + } +}) + +test("direct footer rejects local commands submitted with the queue shortcut", async () => { + const submitted: RunPrompt[] = [] + const statuses: string[] = [] + const app = await renderFooter({ + onSubmit: (prompt) => { + submitted.push(prompt) + return true + }, + onStatus: (status) => statuses.push(status), + }) + + try { + await app.renderOnce() + await app.mockInput.typeText("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/settings ") + app.mockInput.pressEnter({ meta: true }) + await Bun.sleep(0) + expect(submitted).toEqual([]) + expect(statuses).toContain("this prompt cannot be queued") + } finally { + app.cleanup() + } +}) + // OpenTUI currently crashes Bun in the full `test/cli/run` directory run here. // Re-enable after the upstream OpenTUI fix lands in this repo. test.skip("direct footer recreates the frame across command panel transitions", async () => { @@ -1068,11 +1158,11 @@ test("direct footer submits slash autocomplete selections without dispatching sh await app.renderOnce() expect(submits).toEqual([ - { text: "/review ", parts: [], command: { name: "review", arguments: "" } }, - { text: "/review ", parts: [], command: { name: "review", arguments: "" } }, - { text: "/review branch", parts: [], command: { name: "review", arguments: "branch" } }, - { text: "/new ", parts: [] }, - { text: "/new ", parts: [] }, + { text: "/review ", parts: [], command: { name: "review", arguments: "" }, delivery: "steer" }, + { text: "/review ", parts: [], command: { name: "review", arguments: "" }, delivery: "steer" }, + { text: "/review branch", parts: [], command: { name: "review", arguments: "branch" }, delivery: "steer" }, + { text: "/new ", parts: [], delivery: "steer" }, + { text: "/new ", parts: [], delivery: "steer" }, ]) expect(app.renderer.currentFocusedEditor?.plainText).toBe("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/settings ") } finally { @@ -1100,7 +1190,9 @@ test("direct footer slash autocomplete keeps a real skills command", async () => app.mockInput.pressEnter() await app.renderOnce() - expect(submits).toEqual([{ text: "/skills ", parts: [], command: { name: "skills", arguments: "" } }]) + expect(submits).toEqual([ + { text: "/skills ", parts: [], command: { name: "skills", arguments: "" }, delivery: "steer" }, + ]) expect(app.captureCharFrame()).not.toContain("Apply formatter fixes") } finally { app.cleanup() @@ -1158,7 +1250,12 @@ test("direct footer tags skill slash submissions with their catalog source", asy await app.renderOnce() expect(submits).toEqual([ - { text: "/formatter src", parts: [], command: { name: "formatter", arguments: "src", source: "skill" } }, + { + text: "/formatter src", + parts: [], + command: { name: "formatter", arguments: "src", source: "skill" }, + delivery: "steer", + }, ]) } finally { app.cleanup() @@ -1238,7 +1335,7 @@ test.skip("direct footer clears the synthetic skills draft when the panel closes } }) -test("direct footer shows authoritative pending work while running", async () => { +test("direct footer shows authoritative queued work while running", async () => { const [state] = createSignal({ phase: "running", status: "", @@ -1342,9 +1439,9 @@ test("direct footer shows authoritative pending work while running", async () => const hint = statusItems.at(-1)! expect(spinner).toBeDefined() - expect(frame).toContain("1 pending") + expect(frame).toContain("1 queued") expect(frame).toContain("ctrl+b background") - expect(frame).toContain("ctrl+x q 1 pending") + expect(frame).toContain("ctrl+x q 1 queued") expect(frame).toContain("↓ subagents") expect(frame).toContain("ctrl+p cmd") expect(frame).toContain("subagents · ctrl+p cmd") diff --git a/packages/tui/test/mini/runtime.boot.test.ts b/packages/tui/test/mini/runtime.boot.test.ts index 734d2a49233b..067a4d701125 100644 --- a/packages/tui/test/mini/runtime.boot.test.ts +++ b/packages/tui/test/mini/runtime.boot.test.ts @@ -82,7 +82,8 @@ describe("run runtime boot", () => { expect(result.keybinds.get("prompt.history.next")?.[0]?.key).toBe("down") expect(result.keybinds.get("prompt.clear")?.[0]?.key).toBe("ctrl+c") expect(result.keybinds.get("input.submit")?.[0]?.key).toBe("return") - expect(result.keybinds.get("input.newline")?.[0]?.key).toBe("shift+return,ctrl+return,alt+return,ctrl+j") + expect(result.keybinds.get("input.newline")?.[0]?.key).toBe("shift+return,ctrl+return,ctrl+j") + expect(result.keybinds.get("prompt.queue")?.[0]?.key).toBe("alt+return") }) test("preserves disabled leader from resolved tui config", async () => { diff --git a/packages/tui/test/mini/runtime.queue.test.ts b/packages/tui/test/mini/runtime.queue.test.ts index 1d024e433784..c9a73e5adcf0 100644 --- a/packages/tui/test/mini/runtime.queue.test.ts +++ b/packages/tui/test/mini/runtime.queue.test.ts @@ -265,6 +265,33 @@ describe("run runtime queue", () => { await task }) + test("preserves explicit steer and queue delivery for in-flight prompts", async () => { + const ui = createFooterApiFixture() + const admitted: string[] = [] + const gate = Promise.withResolvers() + + const task = runPromptQueue({ + footer: ui.api, + run: async (_input, _signal, onAdmitted) => { + onAdmitted() + await gate.promise + }, + admit: async (input, delivery) => { + admitted.push(`${input.text}:${delivery}`) + }, + settle: async () => ui.api.close(), + }) + + ui.submit("one") + ui.submit("two", undefined, "steer") + ui.submit("three", undefined, "queue") + while (admitted.length < 2) await Bun.sleep(0) + expect(admitted).toEqual(["two:steer", "three:queue"]) + + gate.resolve() + await task + }) + test("continues durable admission after one fails", async () => { const ui = createFooterApiFixture() const admitted: string[] = [] @@ -308,7 +335,7 @@ describe("run runtime queue", () => { admitted() await new Promise((resolve) => signal.addEventListener("abort", () => resolve(), { once: true })) }, - admit: async (_prompt, signal) => { + admit: async (_prompt, _delivery, signal) => { admissionStarted.resolve() await new Promise((resolve) => { if (signal.aborted) { diff --git a/packages/tui/test/mini/runtime.test.ts b/packages/tui/test/mini/runtime.test.ts index d912e6000812..ff661d666779 100644 --- a/packages/tui/test/mini/runtime.test.ts +++ b/packages/tui/test/mini/runtime.test.ts @@ -126,7 +126,7 @@ describe("run interactive runtime", () => { turnStarted.resolve() api.close() }, - queuePromptTurn: async () => {}, + admitPromptTurn: async () => {}, waitForIdle: async () => {}, interruptActiveTurn: async () => {}, selectSubagent: () => {}, @@ -209,7 +209,7 @@ describe("run interactive runtime", () => { streamStarted.resolve() return { runPromptTurn: async () => {}, - queuePromptTurn: async () => {}, + admitPromptTurn: async () => {}, waitForIdle: async () => {}, interruptActiveTurn: async () => {}, selectSubagent: () => {}, @@ -556,7 +556,7 @@ describe("run interactive runtime", () => { setTimeout(() => input.footer.close(), 0) return { runPromptTurn: async () => {}, - queuePromptTurn: async () => {}, + admitPromptTurn: async () => {}, waitForIdle: async () => {}, interruptActiveTurn: async () => {}, selectSubagent: () => {}, diff --git a/packages/tui/test/mini/stream-v2.transport.test.ts b/packages/tui/test/mini/stream-v2.transport.test.ts index 7dcb2772f2b3..b39d2a24a086 100644 --- a/packages/tui/test/mini/stream-v2.transport.test.ts +++ b/packages/tui/test/mini/stream-v2.transport.test.ts @@ -669,6 +669,14 @@ describe("V2 mini transport", () => { data: { text: "follow up" }, delivery: "queue", }, + { + id: "msg_cancelled", + sessionID: "ses_1", + timeCreated: 2, + type: "user", + data: { text: "remove me" }, + delivery: "queue", + }, ], }, }) @@ -684,11 +692,14 @@ describe("V2 mini transport", () => { .findLast((item) => item.type === "queued.prompts") ?.prompts.map((item) => [item.messageID, item.delivery]) - expect(pending()).toEqual([["msg_queued", "queue"]]) + expect(pending()).toEqual([ + ["msg_queued", "queue"], + ["msg_cancelled", "queue"], + ]) events.push({ - id: "evt_promoted", - created: 2, - type: "session.input.promoted", + id: "evt_steered", + created: 3, + type: "session.input.steered", durable: durable("ses_1", 2), data: { sessionID: "ses_1", inputID: "msg_queued" }, }) @@ -697,19 +708,53 @@ describe("V2 mini transport", () => { expect(ui.commits).toContainEqual( expect.objectContaining({ kind: "user", messageID: "msg_queued", text: "follow up" }), ) - expect(pending()).toEqual([]) + expect(pending()).toEqual([["msg_cancelled", "queue"]]) + events.push({ + id: "evt_queued", + created: 4, + type: "session.input.queued", + durable: durable("ses_1", 3), + data: { sessionID: "ses_1", inputID: "msg_queued" }, + }) + while (pending()?.length !== 2) await Bun.sleep(0) + expect(pending()).toEqual([ + ["msg_queued", "queue"], + ["msg_cancelled", "queue"], + ]) + events.push({ + id: "evt_cancelled", + created: 5, + type: "session.input.cancelled", + durable: durable("ses_1", 4), + data: { sessionID: "ses_1", inputID: "msg_cancelled" }, + }) + while (pending()?.length !== 1) await Bun.sleep(0) + expect(pending()).toEqual([["msg_queued", "queue"]]) + events.push({ + id: "evt_promoted", + created: 6, + type: "session.input.promoted", + durable: durable("ses_1", 5), + data: { sessionID: "ses_1", inputID: "msg_queued" }, + }) + while (pending()?.length !== 0) await Bun.sleep(0) + expect(ui.commits.filter((item) => item.messageID === "msg_queued")).toHaveLength(1) const prompt = spyOn(client.session, "prompt").mockImplementation( (request) => ok(promptAdmission(request)) as never, ) - await transport.queuePromptTurn({ + await transport.admitPromptTurn({ agent: "review", - model: undefined, - variant: undefined, + model: { providerID: "test", modelID: "next" }, + variant: "high", prompt: { messageID: "msg_next", text: "another", parts: [] }, files: [], includeFiles: false, - }) + }, "queue") expect(client.session.switchAgent).toHaveBeenCalledWith({ sessionID: "ses_1", agent: "review" }, expect.anything()) + expect(client.session.switchModel).toHaveBeenCalledWith( + { sessionID: "ses_1", model: { providerID: "test", id: "next", variant: "high" } }, + expect.anything(), + ) expect(prompt).toHaveBeenCalledWith(expect.objectContaining({ delivery: "queue" }), expect.anything()) events.push({ id: "evt_earlier_admission", @@ -722,15 +767,8 @@ describe("V2 mini transport", () => { input: { type: "user", data: { text: "earlier" }, delivery: "steer" }, }, }) - while (true) { - const pending = ui.events.findLast((item) => item.type === "queued.prompts") - if (pending?.type === "queued.prompts" && pending.prompts.length >= 2) break - await Bun.sleep(0) - } - expect(pending()).toEqual([ - ["msg_next", "queue"], - ["msg_earlier", "steer"], - ]) + await Bun.sleep(10) + expect(pending()).toEqual([["msg_next", "queue"]]) await transport.close() }) @@ -813,14 +851,14 @@ describe("V2 mini transport", () => { durable: durable("ses_1", 2), data: { sessionID: "ses_1", inputID: "msg_prompt" }, }) - await transport.queuePromptTurn({ + await transport.admitPromptTurn({ agent: undefined, model: undefined, variant: undefined, prompt: { messageID: "msg_queued", text: "follow up", parts: [] }, files: [], includeFiles: false, - }) + }, "queue") events.push({ id: "evt_queued_promoted", created: 3,