feat(cli): emit session queue state for remote clients

.changeset/queue-changed-snapshot.md

@@ -0,0 +1,5 @@
+---
+"kilo-code": minor
+---
+
+Emit session queue state so remote clients can show queued messages.

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) {

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<typeof CloseReason>

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)
   // ---------------------------------------------------------------------------

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<SessionID, number>()
   const activeSince = new Map<SessionID, number>()
+  // 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<SessionID, MessageID[]>()
   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,15 +51,40 @@ 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)) {
         versions.delete(sessionID)
         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<void>()
         // 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)
         }),
     )
   }