feat(queue): add live steer and follow-up session input queues

.github/workflows/go-ci.yml

@@ -174,3 +174,5 @@ jobs:
             -run '^TestPostgresRuntimeFence' ./internal/chat/message
           TEST_POSTGRES_BOOTSTRAP_SCHEMA=0 go test -v -race -count=1 -timeout 90s \
             -run '^TestPostgresRuntimeFence' ./internal/agent/runtime/acp
+          TEST_POSTGRES_BOOTSTRAP_SCHEMA=0 go test -v -race -count=1 -timeout 90s \
+            ./internal/agent/runtime/session/ledger

apps/web/src/composables/api/useChat.chat-api.ts

@@ -17,8 +17,23 @@ import {
   patchBotsByBotIdSessionsBySessionIdAcpRuntimeMode,
   patchBotsByBotIdSessionsBySessionIdAcpRuntimeModel,
   patchBotsByBotIdSessionsBySessionIdAcpRuntimeReasoning,
+  getBotsByBotIdSessionsBySessionIdQueue,
+  postBotsByBotIdSessionsBySessionIdSteerQueue,
+  postBotsByBotIdSessionsBySessionIdFollowUpQueue,
+  postBotsByBotIdSessionsBySessionIdFollowUpQueueByItemIdSteer,
+  putBotsByBotIdSessionsBySessionIdSteerQueueReorder,
+  putBotsByBotIdSessionsBySessionIdFollowUpQueueReorder,
+  patchBotsByBotIdSessionsBySessionIdSteerQueueByItemId,
+  patchBotsByBotIdSessionsBySessionIdFollowUpQueueByItemId,
+  deleteBotsByBotIdSessionsBySessionIdSteerQueueByItemId,
+  deleteBotsByBotIdSessionsBySessionIdFollowUpQueueByItemId,
+} from '@memohai/sdk'
+import type {
+  AcpagentRuntimeStatus,
+  HandlersFollowUpQueueItemResponse,
+  HandlersSessionQueueResponse,
+  HandlersSteerQueueItemResponse,
 } from '@memohai/sdk'
-import type { AcpagentRuntimeStatus } from '@memohai/sdk'
 import type { Bot, SessionSummary } from './useChat.types'
 
 export interface CreateSessionOptions {
@@ -51,6 +66,99 @@ export interface CreateACPRuntimeOptions {
   projectPath?: string
 }
 
+/**
+ * The two queues share one wire shape for the fields the composer renders.
+ * The server keeps them as separate response types; the union here is only a
+ * read-side convenience and never crosses back into a request.
+ */
+export type SessionQueueItem = Pick<
+  HandlersSteerQueueItemResponse & HandlersFollowUpQueueItemResponse,
+  'item_id' | 'status' | 'position' | 'text'
+>
+
+export type SessionQueuesResponse = HandlersSessionQueueResponse
+
+export function queueItemText(item: SessionQueueItem): string {
+  return item.text ?? ''
+}
+
+const queuePath = (botId: string, sessionId: string) => ({ bot_id: botId.trim(), session_id: sessionId.trim() })
+
+export async function enqueueSteerQueue(botId: string, sessionId: string, text: string, invocationId = crypto.randomUUID()): Promise<SessionQueueItem> {
+  const { data } = await postBotsByBotIdSessionsBySessionIdSteerQueue({
+    path: queuePath(botId, sessionId),
+    body: { invocation_id: invocationId, text },
+    throwOnError: true,
+  })
+  return data
+}
+
+export async function fetchSessionQueues(botId: string, sessionId: string): Promise<SessionQueuesResponse> {
+  const { data } = await getBotsByBotIdSessionsBySessionIdQueue({ path: queuePath(botId, sessionId), throwOnError: true })
+  return data ?? {}
+}
+
+export async function enqueueFollowUpQueue(botId: string, sessionId: string, text: string, invocationId = crypto.randomUUID()): Promise<SessionQueueItem> {
+  const { data } = await postBotsByBotIdSessionsBySessionIdFollowUpQueue({
+    path: queuePath(botId, sessionId),
+    body: { invocation_id: invocationId, text },
+    throwOnError: true,
+  })
+  return data
+}
+
+export async function promoteFollowUpQueueItemToSteer(botId: string, sessionId: string, itemId: string): Promise<SessionQueueItem> {
+  const { data } = await postBotsByBotIdSessionsBySessionIdFollowUpQueueByItemIdSteer({
+    path: { ...queuePath(botId, sessionId), item_id: itemId.trim() },
+    throwOnError: true,
+  })
+  return data
+}
+
+export async function updateSteerQueueItem(botId: string, sessionId: string, itemId: string, text: string): Promise<SessionQueueItem> {
+  const { data } = await patchBotsByBotIdSessionsBySessionIdSteerQueueByItemId({
+    path: { ...queuePath(botId, sessionId), item_id: itemId.trim() },
+    body: { text },
+    throwOnError: true,
+  })
+  return data
+}
+
+export async function updateFollowUpQueueItem(botId: string, sessionId: string, itemId: string, text: string): Promise<SessionQueueItem> {
+  const { data } = await patchBotsByBotIdSessionsBySessionIdFollowUpQueueByItemId({
+    path: { ...queuePath(botId, sessionId), item_id: itemId.trim() },
+    body: { text },
+    throwOnError: true,
+  })
+  return data
+}
+
+export async function deleteSteerQueueItem(botId: string, sessionId: string, itemId: string): Promise<void> {
+  await deleteBotsByBotIdSessionsBySessionIdSteerQueueByItemId({ path: { ...queuePath(botId, sessionId), item_id: itemId.trim() }, throwOnError: true })
+}
+
+export async function deleteFollowUpQueueItem(botId: string, sessionId: string, itemId: string): Promise<void> {
+  await deleteBotsByBotIdSessionsBySessionIdFollowUpQueueByItemId({ path: { ...queuePath(botId, sessionId), item_id: itemId.trim() }, throwOnError: true })
+}
+
+export async function reorderSteerQueue(botId: string, sessionId: string, itemId: string, beforeId: string): Promise<SessionQueueItem[]> {
+  const { data } = await putBotsByBotIdSessionsBySessionIdSteerQueueReorder({
+    path: queuePath(botId, sessionId),
+    body: { item: { item_id: itemId }, before: { item_id: beforeId } },
+    throwOnError: true,
+  })
+  return data?.items ?? []
+}
+
+export async function reorderFollowUpQueue(botId: string, sessionId: string, itemId: string, beforeId: string): Promise<SessionQueueItem[]> {
+  const { data } = await putBotsByBotIdSessionsBySessionIdFollowUpQueueReorder({
+    path: queuePath(botId, sessionId),
+    body: { item: { item_id: itemId }, before: { item_id: beforeId } },
+    throwOnError: true,
+  })
+  return data?.items ?? []
+}
+
 export async function fetchBots(): Promise<Bot[]> {
   const { data } = await getBots({ throwOnError: true })
   return data?.items ?? []

apps/web/src/composables/api/useChat.types.ts

@@ -409,14 +409,6 @@ export interface RuntimeCursor {
   seq: number
 }
 
-export interface RuntimeSteerState {
-  id: string
-  status: string
-  text?: string
-  error?: string
-  created_at: string
-  updated_at: string
-}
 
 export interface RuntimeRunOperation {
   kind: 'retry' | 'edit'
@@ -438,14 +430,29 @@ export interface RuntimeCurrentRunView {
   updated_at: string
   messages: UIMessage[]
   request_user_turn?: UIUserTurn
+  // Ordered inputs already admitted into this run. The first entry is the
+  // request turn when present; later entries are applied steers.
+  user_turns?: UIUserTurn[]
+  // Live steer claims projected at their exact assistant-message
+  // boundary. Claimed entries are provisional; applied entries reference the
+  // settled history turn that replaces them.
+  steer_turns?: RuntimeSteerTurnView[]
   error_code?: string
   error?: string
   proposed_terminal_status?: RuntimeRunStatus
   finish_proposed_at?: string
-  steer?: RuntimeSteerState
   operation?: RuntimeRunOperation
 }
 
+export interface RuntimeSteerTurnView {
+  item_id: string
+  status: 'claimed' | 'applied'
+  text: string
+  turn_id?: string
+  after_message_id: number
+  timestamp: string
+}
+
 export interface RuntimeSnapshot {
   bot_id: string
   session_id: string
@@ -460,7 +467,6 @@ export interface RuntimeCurrentRunPatch {
   status?: RuntimeRunStatus
   error_code?: string
   error?: string
-  steer?: RuntimeSteerState
   updated_at?: string
   owner_lease_expires_at?: string
 }
@@ -480,6 +486,9 @@ export interface RuntimeProgressAppend {
 export interface RuntimeDelta {
   current_run_view?: RuntimeCurrentRunView
   run?: RuntimeCurrentRunPatch
+  user_turn_upserts?: UIUserTurn[]
+  steer_turn_upserts?: RuntimeSteerTurnView[]
+  steer_turn_removals?: string[]
   message_appends?: RuntimeMessageAppend[]
   progress_appends?: RuntimeProgressAppend[]
   message_upserts?: UIMessage[]

apps/web/src/i18n/locales/en.json

@@ -112,6 +112,9 @@
     "finish": "Done"
   },
   "errors": {
+    "queue": {
+      "steer_unsupported": "This run cannot accept steer input. Wait for it to finish and send a new message."
+    },
     "bot": {
       "name_taken": "This name is already taken."
     },
@@ -141,6 +144,10 @@
       "response_timeout": "The model did not respond in time. Please try again.",
       "response_interrupted": "The model response was interrupted. Please try again."
     },
+    "queue_no_active_run": "The current response has ended. The message is still queued.",
+    "queue_admission_overloaded": "The queue is busy. Please retry shortly.",
+    "queue_admission_unavailable": "Queue admission is temporarily unavailable. Please retry shortly.",
+    "queue_item_not_pending": "This message is no longer pending in the queue.",
     "profile": {
       "request_invalid": "The profile update request is invalid.",
       "title_model_invalid": "The selected title model is unavailable or is not a chat model.",
@@ -695,6 +702,23 @@
       "fragmentSeparator": ", "
     },
     "currentBot": "Current Bot",
+    "queue": {
+      "steer": "Current response",
+      "followUp": "Up next",
+      "mode": "Send as",
+      "steerDescription": "Add to the current response at the next safe step",
+      "followUpDescription": "Send automatically after the current response",
+      "steerPlaceholder": "Add something to the current response…",
+      "followUpPlaceholder": "Write the next message…",
+      "enqueueSteer": "Add to current response",
+      "enqueueFollowUp": "Add to up next",
+      "steerQueued": "Added to the current response queue",
+      "steerFailed": "Could not steer the current response. The message is still queued.",
+      "reorder": "Reorder",
+      "remove": "Remove",
+      "switchToFollowUp": "Queue after this run",
+      "switchToSteer": "Steer the current run"
+    },
     "inputPlaceholder": "Ask anything",
     "readonlyHint": "This chat is read-only",
     "readonlyPlaceholder": "This chat is read-only. Sending messages is disabled.",

apps/web/src/i18n/locales/ja.json

@@ -109,6 +109,9 @@
     "finish": "完了"
   },
   "errors": {
+    "queue": {
+      "steer_unsupported": "現在の実行にはメッセージを追加できません。終了してから新しいメッセージを送信してください。"
+    },
     "bot": {
       "name_taken": "この名前はすでに使用されています。"
     },
@@ -138,6 +141,10 @@
       "response_timeout": "モデルから時間内に応答がありませんでした。もう一度お試しください。",
       "response_interrupted": "モデルの応答が中断されました。もう一度お試しください。"
     },
+    "queue_no_active_run": "現在の応答は終了しました。メッセージはキューに残っています。",
+    "queue_admission_overloaded": "キューが混雑しています。しばらくしてから再試行してください。",
+    "queue_admission_unavailable": "キューサービスは一時的に利用できません。しばらくしてから再試行してください。",
+    "queue_item_not_pending": "このメッセージは待機中のキューにありません。",
     "profile": {
       "request_invalid": "プロフィール更新リクエストが無効です。",
       "title_model_invalid": "選択したタイトルModelは利用できないか、チャットModelではありません。",
@@ -679,6 +686,23 @@
       "fragmentSeparator": "、"
     },
     "currentBot": "現在のBot",
+    "queue": {
+      "steer": "現在の応答",
+      "followUp": "次のメッセージ",
+      "mode": "送信方法",
+      "steerDescription": "次の安全なステップで現在の応答に追加",
+      "followUpDescription": "現在の応答が終わった後に自動送信",
+      "steerPlaceholder": "現在の応答に追加する内容を入力…",
+      "followUpPlaceholder": "次のメッセージを入力…",
+      "enqueueSteer": "現在の応答に追加",
+      "enqueueFollowUp": "次のメッセージに追加",
+      "steerQueued": "現在の応答キューに追加済み",
+      "steerFailed": "現在の応答に追加できませんでした。メッセージはキューに残っています。",
+      "reorder": "並べ替え",
+      "remove": "削除",
+      "switchToFollowUp": "この実行の後に追加",
+      "switchToSteer": "現在の実行に追加"
+    },
     "inputPlaceholder": "質問を入力してください",
     "readonlyHint": "このチャットは読み取り専用です",
     "readonlyPlaceholder": "このチャットは読み取り専用です。メッセージの送信は無効になっています。",

apps/web/src/i18n/locales/zh.json

@@ -112,6 +112,9 @@
     "finish": "完成"
   },
   "errors": {
+    "queue": {
+      "steer_unsupported": "当前运行不支持插入消息，请等它结束后再发送。"
+    },
     "bot": {
       "name_taken": "该名称已被占用。"
     },
@@ -141,6 +144,10 @@
       "response_timeout": "模型未能及时响应，请重试。",
       "response_interrupted": "模型响应意外中断，请重试。"
     },
+    "queue_no_active_run": "当前回复已经结束，消息仍保留在队列中。",
+    "queue_admission_overloaded": "队列当前繁忙，请稍后重试。",
+    "queue_admission_unavailable": "队列服务暂时不可用，请稍后重试。",
+    "queue_item_not_pending": "这条消息已不在待处理队列中。",
     "profile": {
       "request_invalid": "个人资料更新请求无效。",
       "title_model_invalid": "所选标题模型不可用或不是聊天模型。",
@@ -695,6 +702,23 @@
       "fragmentSeparator": "、"
     },
     "currentBot": "当前 Bot",
+    "queue": {
+      "steer": "当前回复",
+      "followUp": "接下来",
+      "mode": "发送方式",
+      "steerDescription": "在下一个安全步骤加入当前回复",
+      "followUpDescription": "当前回复结束后自动发送",
+      "steerPlaceholder": "输入要插入当前回复的内容…",
+      "followUpPlaceholder": "输入下一条消息…",
+      "enqueueSteer": "插入当前回复",
+      "enqueueFollowUp": "添加到接下来",
+      "steerQueued": "已加入当前回复队列",
+      "steerFailed": "无法插入当前回复，消息仍保留在队列中。",
+      "reorder": "调整顺序",
+      "remove": "删除",
+      "switchToFollowUp": "排到本次运行之后",
+      "switchToSteer": "插入当前运行"
+    },
     "inputPlaceholder": "问点什么",
     "readonlyHint": "该聊天为只读",
     "readonlyPlaceholder": "该聊天为只读，无法发送消息",