|
| 1 | +/** |
| 2 | + * Delivery queue: mid-run queue / steer / interrupt state machine, the serial |
| 3 | + * operation chain that drains it, and the generation-gated delivery hops. |
| 4 | + * |
| 5 | + * One module: `session-queue.ts` (pure item state), `session-operation-queue.ts` |
| 6 | + * (serial promise chain), and `queued-delivery.ts` (kind routing + delivery |
| 7 | + * hops) were three slices of the same drain pipeline. No behavior change — |
| 8 | + * sections below are verbatim moves. |
| 9 | + * |
| 10 | + * Mid-run queue / steer / interrupt state machine (interaction contract §3). |
| 11 | + * Pure data — no paint, no OpenTUI. Shell + demo own delivery and UI flash. |
| 12 | + * |
| 13 | + * Product chords (CL-6290): |
| 14 | + * - Enter mid-run → kind "steer" (soft steer; drain at tool.boundary) |
| 15 | + * - Alt+Enter mid-run → kind "queue" (follow-up; drain only when run goes idle) |
| 16 | + * Internal "reinject" is a separate bridge/shell submit kind, not a QueueKind, |
| 17 | + * and no product chord wires it anymore — leave the path for tests/API only. |
| 18 | + */ |
| 19 | + |
| 20 | +import type { PendingImageAttachment } from "./image-attachments.js"; |
| 21 | +import type { AgentDeliveryResult } from "./deliver-agent-message.js"; |
| 22 | +import type { ProductHostDeliver } from "./product-host.js"; |
| 23 | +import { ASK_DIRECTOR_WAKE_PREFIX } from "../subagent/fleet-report.js"; |
| 24 | +import { MAILBOX_MAIL_WAKE_PREFIX } from "../subagent/mailbox-mail-drive.js"; |
| 25 | + |
| 26 | +export type QueueKind = "queue" | "steer"; |
| 27 | + |
| 28 | +export interface QueueItem { |
| 29 | + readonly id: string; |
| 30 | + readonly text: string; |
| 31 | + readonly kind: QueueKind; |
| 32 | + readonly enqueuedAt: number; |
| 33 | + /** Images attached to this message, delivered with it at the boundary. */ |
| 34 | + readonly attachments?: readonly PendingImageAttachment[]; |
| 35 | +} |
| 36 | + |
| 37 | +export type RunState = "idle" | "busy"; |
| 38 | + |
| 39 | +export interface SessionQueueState { |
| 40 | + readonly run: RunState; |
| 41 | + readonly items: readonly QueueItem[]; |
| 42 | + /** True after interrupt until consumer clears (status flash). */ |
| 43 | + readonly interruptFlash: boolean; |
| 44 | + /** Monotonic id seed for queue items. */ |
| 45 | + readonly nextId: number; |
| 46 | +} |
| 47 | + |
| 48 | +export function createSessionQueue(run: RunState = "idle"): SessionQueueState { |
| 49 | + return { |
| 50 | + run, |
| 51 | + items: [], |
| 52 | + interruptFlash: false, |
| 53 | + nextId: 1, |
| 54 | + }; |
| 55 | +} |
| 56 | + |
| 57 | +/** Pending badge count (queue + steer share one pool for depth totals). */ |
| 58 | +export function badgeCount(state: SessionQueueState): number { |
| 59 | + return state.items.length; |
| 60 | +} |
| 61 | + |
| 62 | +/** Soft-steer pending count (Enter mid-run). */ |
| 63 | +export function steerCount(state: SessionQueueState): number { |
| 64 | + return state.items.filter((i) => i.kind === "steer").length; |
| 65 | +} |
| 66 | + |
| 67 | +/** Follow-up pending count (Alt+Enter mid-run). */ |
| 68 | +export function queueCount(state: SessionQueueState): number { |
| 69 | + return state.items.filter((i) => i.kind === "queue").length; |
| 70 | +} |
| 71 | + |
| 72 | +export function setRunState( |
| 73 | + state: SessionQueueState, |
| 74 | + run: RunState, |
| 75 | +): SessionQueueState { |
| 76 | + if (state.run === run) return state; |
| 77 | + return { ...state, run }; |
| 78 | +} |
| 79 | + |
| 80 | +/** |
| 81 | + * Enqueue a mid-run message. Empty / whitespace-only is a no-op. |
| 82 | + * When idle, still accepts into the queue bag for tests; product shell |
| 83 | + * may route idle Enter as immediate send instead of calling this. |
| 84 | + */ |
| 85 | +export function enqueue( |
| 86 | + state: SessionQueueState, |
| 87 | + text: string, |
| 88 | + kind: QueueKind = "queue", |
| 89 | + now = Date.now(), |
| 90 | + attachments?: readonly PendingImageAttachment[], |
| 91 | +): SessionQueueState { |
| 92 | + const t = text.trim(); |
| 93 | + if ( |
| 94 | + t.length === 0 && |
| 95 | + (attachments === undefined || attachments.length === 0) |
| 96 | + ) { |
| 97 | + return state; |
| 98 | + } |
| 99 | + const item: QueueItem = { |
| 100 | + id: `q${state.nextId}`, |
| 101 | + text: t, |
| 102 | + kind, |
| 103 | + enqueuedAt: now, |
| 104 | + ...(attachments !== undefined && attachments.length > 0 |
| 105 | + ? { attachments } |
| 106 | + : {}), |
| 107 | + }; |
| 108 | + return { |
| 109 | + ...state, |
| 110 | + items: [...state.items, item], |
| 111 | + nextId: state.nextId + 1, |
| 112 | + interruptFlash: false, |
| 113 | + }; |
| 114 | +} |
| 115 | + |
| 116 | +/** Steer = priority enqueue (same badge pool). */ |
| 117 | +export function enqueueSteer( |
| 118 | + state: SessionQueueState, |
| 119 | + text: string, |
| 120 | + now = Date.now(), |
| 121 | + attachments?: readonly PendingImageAttachment[], |
| 122 | +): SessionQueueState { |
| 123 | + return enqueue(state, text, "steer", now, attachments); |
| 124 | +} |
| 125 | + |
| 126 | +/** |
| 127 | + * Hard interrupt: stop the run, keep everything the operator queued. Typing a |
| 128 | + * correction and then interrupting so it lands sooner is the common shape of |
| 129 | + * this gesture, so discarding the queue destroyed exactly the input the |
| 130 | + * operator most wanted delivered. Pending items survive to the next drain |
| 131 | + * boundary; only the run state and the flash change here. |
| 132 | + */ |
| 133 | +export function interrupt(state: SessionQueueState): SessionQueueState { |
| 134 | + return { |
| 135 | + ...state, |
| 136 | + run: "idle", |
| 137 | + interruptFlash: true, |
| 138 | + }; |
| 139 | +} |
| 140 | + |
| 141 | +export function clearInterruptFlash( |
| 142 | + state: SessionQueueState, |
| 143 | +): SessionQueueState { |
| 144 | + if (!state.interruptFlash) return state; |
| 145 | + return { ...state, interruptFlash: false }; |
| 146 | +} |
| 147 | + |
| 148 | +/** |
| 149 | + * Retract the most recently enqueued item, queue or steer alike. Last-only: |
| 150 | + * an operator who wants an earlier item gone has no path here (see |
| 151 | + * `applyShellCancelLast` for why that is the shipped scope, not an oversight). |
| 152 | + */ |
| 153 | +export function cancelLast(state: SessionQueueState): { |
| 154 | + state: SessionQueueState; |
| 155 | + item: QueueItem | null; |
| 156 | +} { |
| 157 | + const item = state.items[state.items.length - 1] ?? null; |
| 158 | + if (item === null) return { state, item: null }; |
| 159 | + return { |
| 160 | + state: { ...state, items: state.items.slice(0, -1) }, |
| 161 | + item, |
| 162 | + }; |
| 163 | +} |
| 164 | + |
| 165 | +/** |
| 166 | + * Retract a specific item by id — the pending column's per-row drop, where the |
| 167 | + * operator picked exactly which held message to kill rather than the newest. |
| 168 | + */ |
| 169 | +export function cancelItem( |
| 170 | + state: SessionQueueState, |
| 171 | + id: string, |
| 172 | +): { state: SessionQueueState; item: QueueItem | null } { |
| 173 | + const index = state.items.findIndex((item) => item.id === id); |
| 174 | + const item = state.items[index] ?? null; |
| 175 | + if (item === null) return { state, item: null }; |
| 176 | + return { |
| 177 | + state: { |
| 178 | + ...state, |
| 179 | + items: [...state.items.slice(0, index), ...state.items.slice(index + 1)], |
| 180 | + }, |
| 181 | + item, |
| 182 | + }; |
| 183 | +} |
| 184 | + |
| 185 | +/** Drain order: steers first (FIFO within class), then queue (FIFO). */ |
| 186 | +export function drainOrder(state: SessionQueueState): readonly QueueItem[] { |
| 187 | + const steers = state.items.filter((i) => i.kind === "steer"); |
| 188 | + const queues = state.items.filter((i) => i.kind === "queue"); |
| 189 | + return [...steers, ...queues]; |
| 190 | +} |
| 191 | + |
| 192 | +/** |
| 193 | + * Pop next delivery item. When `kind` is set, only that class (FIFO within |
| 194 | + * class); otherwise full `drainOrder` (steer-first, then queue). |
| 195 | + */ |
| 196 | +export function drainOne( |
| 197 | + state: SessionQueueState, |
| 198 | + kind?: QueueKind, |
| 199 | +): { state: SessionQueueState; item: QueueItem | null } { |
| 200 | + const order = |
| 201 | + kind === undefined |
| 202 | + ? drainOrder(state) |
| 203 | + : state.items.filter((i) => i.kind === kind); |
| 204 | + const item = order[0] ?? null; |
| 205 | + if (!item) return { state, item: null }; |
| 206 | + return { |
| 207 | + state: { |
| 208 | + ...state, |
| 209 | + items: state.items.filter((i) => i.id !== item.id), |
| 210 | + }, |
| 211 | + item, |
| 212 | + }; |
| 213 | +} |
| 214 | + |
| 215 | +/** Drain every pending soft-steer; leave follow-ups untouched. */ |
| 216 | +export function drainSteersOnly(state: SessionQueueState): { |
| 217 | + state: SessionQueueState; |
| 218 | + drained: readonly QueueItem[]; |
| 219 | +} { |
| 220 | + const drained: QueueItem[] = []; |
| 221 | + let current = state; |
| 222 | + for (;;) { |
| 223 | + const next = drainOne(current, "steer"); |
| 224 | + if (!next.item) break; |
| 225 | + drained.push(next.item); |
| 226 | + current = next.state; |
| 227 | + } |
| 228 | + return { state: current, drained }; |
| 229 | +} |
| 230 | + |
| 231 | +// Serial promise chain for session-scoped operations (reload, interrupt, deliver). |
| 232 | +// Each task runs after the previous one settles; failures do not block the tail. |
| 233 | + |
| 234 | +export interface SessionOperationQueue { |
| 235 | + /** Enqueue an async operation; returns a promise for this operation's settlement. */ |
| 236 | + enqueue: (op: () => Promise<void>) => Promise<void>; |
| 237 | + /** Await the tail of the queue (all prior operations finished or failed). */ |
| 238 | + awaitTail: () => Promise<void>; |
| 239 | +} |
| 240 | + |
| 241 | +export function createSessionOperationQueue(): SessionOperationQueue { |
| 242 | + let tail: Promise<void> = Promise.resolve(); |
| 243 | + |
| 244 | + const enqueue = (op: () => Promise<void>): Promise<void> => { |
| 245 | + tail = tail.then(op, op); |
| 246 | + return tail; |
| 247 | + }; |
| 248 | + |
| 249 | + return { |
| 250 | + enqueue, |
| 251 | + awaitTail: () => tail.catch(() => undefined), |
| 252 | + }; |
| 253 | +} |
| 254 | + |
1 | 255 | /** |
2 | 256 | * Kind routing for drained queue items, plus a generation token so a |
3 | 257 | * /clear|/new rotation can drop in-flight delivers that belonged to the |
|
7 | 261 | * parent tool.boundary. Leftover steers at idle, idle-with-fleet, or |
8 | 262 | * post-interrupt share the send path (sendQueue, inFlight, token refresh). |
9 | 263 | */ |
10 | | -import type { PendingImageAttachment } from "./image-attachments.js"; |
11 | | -import type { AgentDeliveryResult } from "./deliver-agent-message.js"; |
12 | | -import type { ProductHostDeliver } from "./product-host.js"; |
13 | | -import { ASK_DIRECTOR_WAKE_PREFIX } from "../subagent/fleet-report.js"; |
14 | | -import { MAILBOX_MAIL_WAKE_PREFIX } from "../subagent/mailbox-mail-drive.js"; |
15 | 264 |
|
16 | 265 | export type DeliverySettle = (result: AgentDeliveryResult) => void; |
17 | 266 | type MaybeAsyncDeliveryResult = |
|
0 commit comments