diff --git a/.changeset/queue-changed-snapshot.md b/.changeset/queue-changed-snapshot.md new file mode 100644 index 00000000000..94f785ae3de --- /dev/null +++ b/.changeset/queue-changed-snapshot.md @@ -0,0 +1,5 @@ +--- +"kilo-code": minor +--- + +Emit session queue state so remote clients can show queued messages. diff --git a/packages/opencode/src/kilo-sessions/remote-sender.ts b/packages/opencode/src/kilo-sessions/remote-sender.ts index 84a747560a8..267677b9962 100644 --- a/packages/opencode/src/kilo-sessions/remote-sender.ts +++ b/packages/opencode/src/kilo-sessions/remote-sender.ts @@ -9,6 +9,7 @@ import type { MessageV2 } from "@/session/message-v2" import { SessionPrompt } from "@/session/prompt" import { Question } from "@/question" import { Suggestion } from "@/kilocode/suggestion" // kilocode_change +import { KiloSessionPromptQueue } from "@/kilocode/session/prompt-queue" import { Permission } from "@/permission" import { PermissionV1 } from "@opencode-ai/core/v1/permission" import { SessionID } from "@/session/schema" @@ -344,6 +345,19 @@ export namespace RemoteSender { data: p, }) } + // Always send the current queue snapshot, including + // empty, so a resubscribing client can reconcile stale "Queued" badges. + // Uses send() directly (not publishQueueChanged) to avoid re-broadcasting + // to every other subscriber. The forwarder already routes live + // session.queue.changed events to subscribed clients via extractSessionId. + const queued = KiloSessionPromptQueue.snapshot(SessionID.make(sessionId)) + options.conn.send({ + type: "event", + sessionId, + ...(root ? { parentSessionId: root } : {}), + event: "session.queue.changed", + data: { sessionID: sessionId, queued }, + }) } async function backfillPendingState(sessionId: string) { diff --git a/packages/opencode/src/kilocode/session/event.ts b/packages/opencode/src/kilocode/session/event.ts index c142ff8211f..47025588a81 100644 --- a/packages/opencode/src/kilocode/session/event.ts +++ b/packages/opencode/src/kilocode/session/event.ts @@ -1,5 +1,5 @@ import { BusEvent } from "@/bus/bus-event" -import { SessionID } from "@/session/schema" +import { MessageID, SessionID } from "@/session/schema" import { Schema } from "effect" const CloseReason = Schema.Literals(["completed", "error", "interrupted"]) @@ -19,6 +19,16 @@ export const KiloSessionEvent = { reason: CloseReason, }), ), + // FIFO snapshot of queued (waiting, not-yet-running) + // user message IDs per session, for remote clients (mobile) to show + // "Queued" badges. The currently-running turn's own message is not included. + QueueChanged: BusEvent.define( + "session.queue.changed", + Schema.Struct({ + sessionID: SessionID, + queued: Schema.Array(MessageID), + }), + ), } export type KiloSessionCloseReason = Schema.Schema.Type diff --git a/packages/opencode/src/kilocode/session/index.ts b/packages/opencode/src/kilocode/session/index.ts index 41a97ad90b6..3810eafaf36 100644 --- a/packages/opencode/src/kilocode/session/index.ts +++ b/packages/opencode/src/kilocode/session/index.ts @@ -2,7 +2,7 @@ import { prepareForkedPart as _prepareForkedPart } from "./fork" import z from "zod" import { Cause, Effect, Schema } from "effect" import { Bus } from "@/bus" -import { Instance } from "@/kilocode/instance" +import { Instance, type InstanceContext } from "@/kilocode/instance" import { EffectBridge } from "@/effect/bridge" import { Session } from "@/session/session" import { MessageID, SessionID } from "@/session/schema" @@ -18,6 +18,7 @@ import type { Provider } from "@/provider/provider" import { ENV_FEATURE } from "@kilocode/kilo-gateway" import { existsSync } from "fs" import path from "path" +import { iife } from "@/util/iife" import { KiloSessionEvent, type KiloSessionCloseReason } from "./event" export namespace KiloSession { @@ -38,6 +39,28 @@ export namespace KiloSession { export const publishTurnClose = (input: { sessionID: SessionID; parentID?: SessionID; reason: CloseReason }) => Effect.promise(() => Bus.publish(Instance.current, Event.TurnClose, input)) + // FIFO snapshot of the per-session waiting list. + // Emitted by KiloSessionPromptQueue on every transition that changes the set + // of queued (not-yet-running) user messages. + export const publishQueueChanged = (input: { sessionID: SessionID; queued: MessageID[] }) => + Effect.promise(() => Bus.publish(Instance.current, Event.QueueChanged, input)) + + // Synchronous, fire-and-forget variant for callers that run outside an Effect + // context (e.g. KiloSessionPromptQueue transitions, which fire from inside + // Effect.sync blocks). Swallows errors so a transient context loss never + // breaks the queue. + export function publishQueueChangedAsync(input: { sessionID: SessionID; queued: MessageID[] }) { + const ctx = iife((): InstanceContext | undefined => { + try { + return Instance.current + } catch { + return undefined + } + }) + if (!ctx) return + Bus.publish(ctx, Event.QueueChanged, input).catch(err => log.warn("queue changed publish failed", { err })) + } + // --------------------------------------------------------------------------- // Per-session platform override (telemetry attribution) // --------------------------------------------------------------------------- diff --git a/packages/opencode/src/kilocode/session/prompt-queue.ts b/packages/opencode/src/kilocode/session/prompt-queue.ts index 419c7e1379e..75e1774cd37 100644 --- a/packages/opencode/src/kilocode/session/prompt-queue.ts +++ b/packages/opencode/src/kilocode/session/prompt-queue.ts @@ -1,6 +1,7 @@ import { Effect } from "effect" import { MessageV2 } from "@/session/message-v2" import { MessageID, SessionID } from "@/session/schema" +import { KiloSession } from "@/kilocode/session" type Slot = { readonly seq: number @@ -25,11 +26,22 @@ export namespace KiloSessionPromptQueue { // a newer slot was enqueued after the active one began running. const latest = new Map() const activeSince = new Map() + // FIFO waiting list of user message IDs that have been + // enqueued but have not yet started running. The currently-running slot's + // own message is never in this list. Published via session.queue.changed so + // remote clients can reconcile "Queued" badges. + const waiting = new Map() let seq = 0 /** @internal - test-only helper */ export function _hasInternalState(sessionID: SessionID): boolean { - return versions.has(sessionID) || targets.has(sessionID) || latest.has(sessionID) || activeSince.has(sessionID) + return ( + versions.has(sessionID) || + targets.has(sessionID) || + latest.has(sessionID) || + activeSince.has(sessionID) || + waiting.has(sessionID) + ) } const version = (sessionID: SessionID) => versions.get(sessionID) ?? 0 @@ -39,6 +51,25 @@ export namespace KiloSessionPromptQueue { () => undefined, ) + // Read-only FIFO snapshot of the per-session waiting + // list. Used by replay-on-subscribe in remote-sender.ts to always emit the + // current queue state (including empty) to a resubscribing client. + export function snapshot(sessionID: SessionID): MessageID[] { + return [...(waiting.get(sessionID) ?? [])] + } + + // Emit session.queue.changed when the waiting set + // actually changes; suppress the redundant empty→empty transition to keep + // the bus quiet. Replay uses snapshot() directly so it is never affected. + const publishIfChanged = (sessionID: SessionID, next: MessageID[]) => { + const prev = waiting.get(sessionID) ?? [] + if (prev.length === 0 && next.length === 0) return + if (prev.length === next.length && prev.every((id, i) => id === next[i])) return + if (next.length === 0) waiting.delete(sessionID) + else waiting.set(sessionID, next) + KiloSession.publishQueueChangedAsync({ sessionID, queued: snapshot(sessionID) }) + } + export function cancel(sessionID: SessionID) { return Effect.sync(() => { if (!tails.has(sessionID)) { @@ -46,8 +77,14 @@ export namespace KiloSessionPromptQueue { targets.delete(sessionID) latest.delete(sessionID) activeSince.delete(sessionID) + // Cancel on an idle session still drops any + // lingering waiting entry, then publishes an empty snapshot. + publishIfChanged(sessionID, []) return } + // Active turn: bump version invalidates the + // queued slots, then drop the waiting list and publish empty. + publishIfChanged(sessionID, []) versions.set(sessionID, version(sessionID) + 1) }) } @@ -128,11 +165,20 @@ export namespace KiloSessionPromptQueue { Effect.sync(() => { const mine = ++seq latest.set(sessionID, mine) + // Record whether this slot starts immediately + // (no existing tail) so we can publish the queue change exactly once + // on the transition that actually mutates the waiting list. + const startsImmediately = !tails.has(sessionID) const previous = tails.get(sessionID) ?? Promise.resolve() const done = Promise.withResolvers() // Keep later queued prompts moving; each caller still observes its own failure. const tail = settle(previous).then(() => done.promise) tails.set(sessionID, tail) + if (!startsImmediately) { + // Another slot is still running; this prompt joins the waiting FIFO. + const list = waiting.get(sessionID) ?? [] + publishIfChanged(sessionID, [...list, target]) + } return { seq: mine, version: version(sessionID), previous, done, tail } satisfies Slot }), (slot) => @@ -143,6 +189,17 @@ export namespace KiloSessionPromptQueue { // running. hasFollowup compares against this value so the slot only // breaks when something newer than itself arrives. activeSince.set(sessionID, latest.get(sessionID) ?? slot.seq) + // This slot is taking over, so drop its + // own message ID from the head of the waiting list (if present) + // and publish the updated snapshot. Cancelled slots never reach + // this branch, so the waiting list retains only truly-pending IDs. + const list = waiting.get(sessionID) + if (list && list.length > 0) { + const head = list[0] + if (head === target) { + publishIfChanged(sessionID, list.slice(1)) + } + } return Effect.acquireUseRelease( Effect.sync(() => { targets.set(sessionID, { base: target, extras: new Set() }) @@ -164,6 +221,9 @@ export namespace KiloSessionPromptQueue { targets.delete(sessionID) latest.delete(sessionID) activeSince.delete(sessionID) + // Last slot of the session finished cleanly; + // drop any lingering waiting entry and clear internal state. + waiting.delete(sessionID) }), ) } diff --git a/packages/opencode/test/kilocode/session-prompt-queue.test.ts b/packages/opencode/test/kilocode/session-prompt-queue.test.ts index 5ec0eb666e7..93438932e2b 100644 --- a/packages/opencode/test/kilocode/session-prompt-queue.test.ts +++ b/packages/opencode/test/kilocode/session-prompt-queue.test.ts @@ -8,6 +8,7 @@ import { AppRuntime } from "../../src/effect/app-runtime" import { InstanceRef } from "../../src/effect/instance-ref" import { KiloSessionCompaction } from "@/kilocode/session/compaction" import { KiloSessionPromptQueue } from "@/kilocode/session/prompt-queue" +import { KiloSession } from "@/kilocode/session" import { Suggestion } from "../../src/kilocode/suggestion" import { ProviderV2 } from "@opencode-ai/core/provider" import { ModelV2 } from "@opencode-ai/core/model" @@ -854,4 +855,297 @@ describe("session prompt queue", () => { }, }) }) + + // session.queue.changed event surface + snapshot accessor + describe("session.queue.changed", () => { + test("snapshot() returns an empty list for an unknown session", () => { + expect(KiloSessionPromptQueue.snapshot(SessionID.make("session_unknown"))).toEqual([]) + }) + + test("enqueueing on an idle session does not transiently publish a non-empty snapshot", async () => { + // A prompt enqueued into an idle session starts almost immediately, so + // it must never appear in the waiting list (and must not emit any + // session.queue.changed event whose queued list is non-empty). + await using tmp = await tmpdir({ git: true }) + await provideTestInstance({ + directory: tmp.path, + fn: async () => { + const sessionID = SessionID.make("session_queue_idle") + const events: Array<{ type: string; queued: string[] }> = [] + const off = Bus.subscribe(KiloSession.Event.QueueChanged, (event) => { + events.push({ type: event.type, queued: [...(event.properties.queued as readonly string[])] }) + }) + try { + await Effect.runPromise( + KiloSessionPromptQueue.enqueue( + sessionID, + MessageID.make("msg_idle_1"), + Effect.succeed("done"), + Effect.succeed("cancelled"), + ), + ) + expect(KiloSessionPromptQueue.snapshot(sessionID)).toEqual([]) + expect(events).toEqual([]) + } finally { + off() + } + }, + }) + }) + + test("enqueueing while busy appends to the FIFO snapshot and emits the event", async () => { + await using tmp = await tmpdir({ git: true }) + await provideTestInstance({ + directory: tmp.path, + fn: async () => { + const sessionID = SessionID.make("session_queue_busy") + const events: Array<{ sessionID: string; queued: string[] }> = [] + const off = Bus.subscribe(KiloSession.Event.QueueChanged, (event) => { + if (event.properties.sessionID === sessionID) { + events.push({ + sessionID: event.properties.sessionID as string, + queued: [...(event.properties.queued as readonly string[])], + }) + } + }) + + const firstStarted = Promise.withResolvers() + const firstRelease = Promise.withResolvers() + const m1 = MessageID.make("msg_busy_1") + const m2 = MessageID.make("msg_busy_2") + const m3 = MessageID.make("msg_busy_3") + + try { + const first = Effect.runPromise( + KiloSessionPromptQueue.enqueue( + sessionID, + m1, + Effect.gen(function* () { + firstStarted.resolve() + yield* Effect.promise(() => firstRelease.promise) + return "first" as const + }), + Effect.succeed("first-cancelled" as const), + ), + ) + await firstStarted.promise + + // Idle-start for slot 1 must not have emitted anything. + expect(KiloSessionPromptQueue.snapshot(sessionID)).toEqual([]) + expect(events).toEqual([]) + + const second = Effect.runPromise( + KiloSessionPromptQueue.enqueue( + sessionID, + m2, + Effect.succeed("second" as const), + Effect.succeed("second-cancelled" as const), + ), + ) + const third = Effect.runPromise( + KiloSessionPromptQueue.enqueue( + sessionID, + m3, + Effect.succeed("third" as const), + Effect.succeed("third-cancelled" as const), + ), + ) + + // FIFO order is preserved: msg_busy_2 then msg_busy_3. + expect(KiloSessionPromptQueue.snapshot(sessionID)).toEqual([m2, m3]) + // Publishes are fire-and-forget microtasks; let them flush. + await Bun.sleep(10) + expect(events).toEqual([ + { sessionID, queued: [m2] }, + { sessionID, queued: [m2, m3] }, + ]) + + firstRelease.resolve() + expect(await first).toBe("first") + expect(await second).toBe("second") + expect(await third).toBe("third") + } finally { + off() + } + }, + }) + }) + + test("a waiting slot starting running shrinks the snapshot and emits the event", async () => { + await using tmp = await tmpdir({ git: true }) + await provideTestInstance({ + directory: tmp.path, + fn: async () => { + const sessionID = SessionID.make("session_queue_start") + const events: Array<{ queued: string[] }> = [] + const off = Bus.subscribe(KiloSession.Event.QueueChanged, (event) => { + if (event.properties.sessionID === sessionID) { + events.push({ queued: [...(event.properties.queued as readonly string[])] }) + } + }) + + const firstStarted = Promise.withResolvers() + const firstRelease = Promise.withResolvers() + const secondStarted = Promise.withResolvers() + const secondRelease = Promise.withResolvers() + const m1 = MessageID.make("msg_start_1") + const m2 = MessageID.make("msg_start_2") + + try { + const first = Effect.runPromise( + KiloSessionPromptQueue.enqueue( + sessionID, + m1, + Effect.gen(function* () { + firstStarted.resolve() + yield* Effect.promise(() => firstRelease.promise) + return "first" as const + }), + Effect.succeed("first-cancelled" as const), + ), + ) + await firstStarted.promise + + const second = Effect.runPromise( + KiloSessionPromptQueue.enqueue( + sessionID, + m2, + Effect.gen(function* () { + secondStarted.resolve() + yield* Effect.promise(() => secondRelease.promise) + return "second" as const + }), + Effect.succeed("second-cancelled" as const), + ), + ) + + // msg2 is waiting behind msg1. + expect(KiloSessionPromptQueue.snapshot(sessionID)).toEqual([m2]) + await Bun.sleep(10) + expect(events.map((e) => e.queued)).toEqual([[m2]]) + + // Release msg1; msg2 takes over and the waiting list drops to empty. + firstRelease.resolve() + await secondStarted.promise + + expect(KiloSessionPromptQueue.snapshot(sessionID)).toEqual([]) + await Bun.sleep(10) + expect(events.map((e) => e.queued)).toEqual([[m2], []]) + + secondRelease.resolve() + expect(await first).toBe("first") + expect(await second).toBe("second") + } finally { + off() + } + }, + }) + }) + + test("cancel empties the snapshot and emits an empty list", async () => { + await using tmp = await tmpdir({ git: true }) + await provideTestInstance({ + directory: tmp.path, + fn: async () => { + const sessionID = SessionID.make("session_queue_cancel") + const events: Array<{ queued: string[] }> = [] + const off = Bus.subscribe(KiloSession.Event.QueueChanged, (event) => { + if (event.properties.sessionID === sessionID) { + events.push({ queued: [...(event.properties.queued as readonly string[])] }) + } + }) + + const firstStarted = Promise.withResolvers() + const firstRelease = Promise.withResolvers() + const m1 = MessageID.make("msg_cancel_1") + const m2 = MessageID.make("msg_cancel_2") + const m3 = MessageID.make("msg_cancel_3") + + try { + const first = Effect.runPromise( + KiloSessionPromptQueue.enqueue( + sessionID, + m1, + Effect.gen(function* () { + firstStarted.resolve() + yield* Effect.promise(() => firstRelease.promise) + return "first" as const + }), + Effect.succeed("first-cancelled" as const), + ), + ) + await firstStarted.promise + + const second = Effect.runPromise( + KiloSessionPromptQueue.enqueue( + sessionID, + m2, + Effect.succeed("second" as const), + Effect.succeed("second-cancelled" as const), + ), + ) + const third = Effect.runPromise( + KiloSessionPromptQueue.enqueue( + sessionID, + m3, + Effect.succeed("third" as const), + Effect.succeed("third-cancelled" as const), + ), + ) + + expect(KiloSessionPromptQueue.snapshot(sessionID)).toEqual([m2, m3]) + await Bun.sleep(10) + const beforeCancel = events.length + expect(beforeCancel).toBeGreaterThan(0) + + await Effect.runPromise(KiloSessionPromptQueue.cancel(sessionID)) + + // The most recent emission must be the empty list, and the snapshot + // must be empty for downstream replay callers. + expect(KiloSessionPromptQueue.snapshot(sessionID)).toEqual([]) + await Bun.sleep(10) + expect(events.length).toBeGreaterThan(beforeCancel) + expect(events.at(-1)?.queued).toEqual([]) + expect(events.slice(beforeCancel).every((e) => e.queued.length === 0)).toBe(true) + + firstRelease.resolve() + expect(await first).toBe("first") + // Cancel bumped the version, so the queued slots return their + // cancelled effect instead of running their work. + expect(await second).toBe("second-cancelled") + expect(await third).toBe("third-cancelled") + } finally { + off() + } + }, + }) + }) + + test("cancel on an idle session suppresses the empty→empty emission", async () => { + // Steady-state no-op empty→empty emissions are intentionally suppressed + // by the queue to keep the bus quiet. Replay uses snapshot() directly + // and is therefore never affected by this suppression. + await using tmp = await tmpdir({ git: true }) + await provideTestInstance({ + directory: tmp.path, + fn: async () => { + const sessionID = SessionID.make("session_queue_cancel_idle") + const events: Array<{ queued: string[] }> = [] + const off = Bus.subscribe(KiloSession.Event.QueueChanged, (event) => { + if (event.properties.sessionID === sessionID) { + events.push({ queued: [...(event.properties.queued as readonly string[])] }) + } + }) + + try { + await Effect.runPromise(KiloSessionPromptQueue.cancel(sessionID)) + expect(KiloSessionPromptQueue.snapshot(sessionID)).toEqual([]) + expect(events).toEqual([]) + } finally { + off() + } + }, + }) + }) + }) }) diff --git a/packages/opencode/test/kilocode/sessions/remote-sender.test.ts b/packages/opencode/test/kilocode/sessions/remote-sender.test.ts index cbd5f6a49d7..214793c00a4 100644 --- a/packages/opencode/test/kilocode/sessions/remote-sender.test.ts +++ b/packages/opencode/test/kilocode/sessions/remote-sender.test.ts @@ -16,7 +16,8 @@ import { ProviderV2 } from "@opencode-ai/core/provider" import { ModelV2 } from "@opencode-ai/core/model" import { SessionID } from "../../../src/session/schema" import { Session } from "../../../src/session/session" -import { Suggestion } from "../../../src/kilocode/suggestion" // kilocode_change +import { Suggestion } from "../../../src/kilocode/suggestion" +import { KiloSessionPromptQueue } from "../../../src/kilocode/session/prompt-queue" function fakeConn() { const sent: any[] = [] @@ -1658,6 +1659,8 @@ describe("RemoteSender", () => { spyOn(Suggestion, "list").mockResolvedValue([ { id: "sug_1", sessionID: "ses_other", text: "Review?", actions: [] } as any, ]) + // Queue snapshot is always replayed, even when empty + spyOn(KiloSessionPromptQueue, "snapshot").mockReturnValue([]) const sender = RemoteSender.create({ conn, @@ -1681,8 +1684,85 @@ describe("RemoteSender", () => { sender.handle({ type: "subscribe", sessionId: "ses_target" }) await new Promise((r) => setTimeout(r, 10)) - const events = sent.filter((m: any) => m.type === "event") - expect(events).toHaveLength(0) + // No question/permission/suggestion events for the subscribed session, but + // the queue snapshot replay always fires (here, an empty list) so a + // resubscribing client can reconcile stale "Queued" badges. + const replayed = sent.filter((m: any) => m.type === "event") + expect(replayed).toEqual([ + { + type: "event", + sessionId: "ses_target", + event: "session.queue.changed", + data: { sessionID: "ses_target", queued: [] }, + }, + ]) + }) + + // Queue snapshot replay-on-subscribe coverage + test("subscribe always replays the current queue snapshot, including empty", async () => { + // A resubscribing/reconnecting client must see the authoritative queue + // state immediately, even when the session has no queued messages. This + // is what lets mobile reconcile a stale "Queued" badge away. + const { conn, sent } = fakeConn() + const bus = fakeBus() + + spyOn(Suggestion, "list").mockResolvedValue([]) + spyOn(KiloSessionPromptQueue, "snapshot").mockReturnValue([]) + + const sender = RemoteSender.create({ + conn, + directory: "/tmp/test", + log: nolog, + subscribe: bus.subscribe, + provide: async (input: any) => input.fn(), + question: questions(), + permission: permissions(), + }) + + sender.handle({ type: "subscribe", sessionId: "ses_target" }) + await new Promise((r) => setTimeout(r, 10)) + + const queueEvents = sent.filter((m: any) => m.event === "session.queue.changed") + expect(queueEvents).toEqual([ + { + type: "event", + sessionId: "ses_target", + event: "session.queue.changed", + data: { sessionID: "ses_target", queued: [] }, + }, + ]) + expect(KiloSessionPromptQueue.snapshot).toHaveBeenCalledWith(SessionID.make("ses_target")) + }) + + test("subscribe replays a non-empty queue snapshot for the subscribed session", async () => { + const { conn, sent } = fakeConn() + const bus = fakeBus() + + spyOn(Suggestion, "list").mockResolvedValue([]) + spyOn(KiloSessionPromptQueue, "snapshot").mockReturnValue(["msg_a", "msg_b"] as any) + + const sender = RemoteSender.create({ + conn, + directory: "/tmp/test", + log: nolog, + subscribe: bus.subscribe, + provide: async (input: any) => input.fn(), + question: questions(), + permission: permissions(), + }) + + sender.handle({ type: "subscribe", sessionId: "ses_target" }) + await new Promise((r) => setTimeout(r, 10)) + + const queueEvents = sent.filter((m: any) => m.event === "session.queue.changed") + expect(queueEvents).toEqual([ + { + type: "event", + sessionId: "ses_target", + event: "session.queue.changed", + data: { sessionID: "ses_target", queued: ["msg_a", "msg_b"] }, + }, + ]) }) test("subscribe replays pending suggestion for the subscribed session", async () => {