feat(chat): steer persistent queued messages into the active turn

apps/presentation/dashboard/src/data/chat.ts

@@ -965,6 +965,45 @@ export async function fetchChatSession(sessionId: string, signal?: AbortSignal)
   return requestJson<ChatSessionSnapshot>(`/api/chat/sessions/${encodeURIComponent(sessionId)}`, { signal });
 }
 
+const chatQueueSchema = z.object({
+  ok: z.literal(true), schema_version: z.literal("loopx_chat_store_v1"), session_id: z.string().min(1),
+  active_turn_id: z.string().min(1).nullable(),
+  turns: z.array(z.object({
+    turn_id: z.string().min(1), created_at: z.string(), origin: z.string().nullable(),
+    preview: z.string(), character_count: z.number().int().nonnegative(),
+    attachment_count: z.number().int().nonnegative(),
+    steering: z.object({
+      schema_version: z.literal("loopx_chat_queue_steering_v1"),
+      client_ingress_id: z.string().min(1), operation_id: z.string().min(1),
+      target_turn_id: z.string().min(1), status: z.enum(["pending", "delivered", "failed"]),
+      error_code: z.string().optional(),
+    }).nullable(),
+  })),
+});
+export type ChatQueuedTurn = z.infer<typeof chatQueueSchema>["turns"][number];
+
+/** Bounded metadata from the existing persistent queue, never a local queue. */
+export async function fetchChatQueue(sessionId: string, signal?: AbortSignal) {
+  const result = chatQueueSchema.safeParse(await requestJson<unknown>(
+    `/api/chat/sessions/${encodeURIComponent(sessionId)}/queue`, {signal}));
+  if (!result.success || result.data.session_id !== sessionId
+    || new Set(result.data.turns.map(row => row.turn_id)).size !== result.data.turns.length) {
+    throw new Error("The original Session queue could not be verified.");
+  }
+  return result.data;
+}
+
+export async function steerQueuedChatTurn(sessionId: string, queuedId: string, targetId: string, ingressId: string) {
+  const value = await requestJson<Record<string, unknown>>(
+    `/api/chat/sessions/${encodeURIComponent(sessionId)}/queue/${encodeURIComponent(queuedId)}/steer`,
+    {method: "POST", body: JSON.stringify({target_turn_id: targetId, client_ingress_id: ingressId}),
+      signal: AbortSignal.timeout(15_000)});
+  if (value.ok !== true || value.session_id !== sessionId || value.queued_turn_id !== queuedId
+    || value.turn_id !== targetId || value.client_ingress_id !== ingressId || value.status !== "delivered") {
+    throw new Error("Queue steering acknowledgement does not match the original request");
+  }
+}
+
 const chatMessagePageSchema = z.object({
   schema_version: z.literal("loopx_chat_message_page_v1"),
   offset: z.number().int().nonnegative(), total: z.number().int().nonnegative(),
@@ -1553,7 +1592,7 @@ export async function sendChatTurnStreaming(
     options.attachments,
     options.signal,
   );
-  options.onPhase?.("turn.accepted", accepted.turn_id);
+  options.onPhase?.(accepted.status === "queued" ? "turn.queued" : "turn.accepted", accepted.turn_id);
   return receiveChatTurnStreaming(
     sessionId,
     accepted.turn_id,
@@ -1574,6 +1613,7 @@ async function receiveChatTurnStreaming(
   } = {},
 ) {
   let finalResponse: unknown = null;
+  let steeredIntoTurnId: string | undefined;
   const outcome: {
     failure: Record<string, unknown> | null;
     interrupted: Record<string, unknown> | null;
@@ -1591,6 +1631,9 @@ async function receiveChatTurnStreaming(
           if (label) options.onActivity?.(label, parseTurnStep(event.payload.step));
         }
         if (event.kind === "turn.completed") {
+          if (typeof event.payload.steered_into_turn_id === "string" && event.payload.steered_into_turn_id) {
+            steeredIntoTurnId = event.payload.steered_into_turn_id;
+          }
           finalResponse = event.payload.response;
         }
         if (event.kind === "turn.failed") {
@@ -1632,6 +1675,7 @@ async function receiveChatTurnStreaming(
     response: agentResponseSchema.parse(finalResponse),
     sessionId,
     turnId,
+    steeredIntoTurnId,
   };
 }
 
@@ -1642,7 +1686,7 @@ async function receiveChatTurnStreaming(
 export async function readCompletedChatTurn(sessionId: string, turnId: string, signal: AbortSignal): Promise<AgentResponse | null> {
   let completed: AgentResponse | null = null;
   await streamChatTurn(`/api/chat/sessions/${sessionId}/turns/${turnId}/events`, (event) => {
-    if (event.kind === "turn.completed") completed = agentResponseSchema.parse(event.payload.response);
+    if (event.kind === "turn.completed" && !event.payload.steered_into_turn_id) completed = agentResponseSchema.parse(event.payload.response);
   }, signal);
   return completed;
 }

apps/presentation/dashboard/src/data/use-conversation-queue.ts

@@ -0,0 +1,41 @@
+import { useEffect, useState } from "react";
+import { fetchChatQueue, type ChatQueuedTurn } from "./chat";
+
+/** Observes one original Session; its persisted queue remains the only owner. */
+export function useConversationQueue({sessionId, scope, active, refreshKey}: {
+  sessionId?: string; scope: string; active: boolean; refreshKey: string;
+}) {
+  const key = JSON.stringify([scope, sessionId]);
+  const [attempt, setAttempt] = useState(0);
+  const [state, setState] = useState<{key: string; turns: ChatQueuedTurn[]; activeTurnId: string | null; error: Error | null} | null>(null);
+  useEffect(() => {
+    if (!sessionId) return;
+    const controller = new AbortController();
+    let timer: ReturnType<typeof setTimeout> | undefined;
+    let failures = 0, pending = false;
+    const read = async () => {
+      try {
+        const snapshot = await fetchChatQueue(sessionId, AbortSignal.any([controller.signal, AbortSignal.timeout(5000)]));
+        if (controller.signal.aborted) return;
+        failures = 0; pending = snapshot.turns.length > 0 || Boolean(snapshot.active_turn_id);
+        setState({key, turns: snapshot.turns, activeTurnId: snapshot.active_turn_id, error: null});
+      } catch (cause) {
+        if (controller.signal.aborted) return;
+        failures += 1;
+        // A failed read is unknown, including revoked scope. Never retain
+        // potentially inaccessible input or present an empty successful queue.
+        setState({key, turns: [], activeTurnId: null, error: cause instanceof Error ? cause : new Error(String(cause))});
+      }
+      // After a successful read, the original queue/Session decides whether
+      // observation continues; a cached model hint must not keep idle polling.
+      if (!controller.signal.aborted && (pending || (failures > 0 && active))) {
+        timer = setTimeout(() => void read(), failures ? Math.min(5000 * 2 ** failures, 30_000) : 5000);
+      }
+    };
+    void read();
+    return () => { controller.abort(); clearTimeout(timer); };
+  }, [key, sessionId, active, refreshKey, attempt]);
+  const current = sessionId && state?.key === key ? state : null;
+  return {turns: current?.turns ?? [], activeTurnId: current?.activeTurnId ?? null, error: current?.error ?? null,
+    retry: () => setAttempt(value => value + 1)};
+}

apps/presentation/dashboard/src/features/personal-workspace/channel-timeline.tsx

@@ -48,12 +48,14 @@ export function ChannelTimeline({
   historyPhase = "ready",
   onReadMessage,
   onRetryHistory,
+  queuedTurnKeys,
 }: {
   emptyState?: { title: string; description: string };
   messageFocus?: {sessionId: string; messageId: string};
   historyPhase?: ConversationHistoryStatus["phase"];
   onReadMessage?: () => void;
   onRetryHistory?: () => void;
+  queuedTurnKeys?: ReadonlySet<string>;
   readOnly?: boolean;
   items: WorkspaceTimelineItem[];
   onCancelPreparation?: () => void;
@@ -67,7 +69,15 @@ export function ChannelTimeline({
   onAdjustTurn?: (turnId: string) => void;
 }) {
   const { locale, t } = useWorkspaceI18n();
-  const orderedItems = conversationOrder(items);
+  const orderedItems = conversationOrder(items.filter(item => {
+    if (item.kind !== "message") return true;
+    if (item.message.queued) return false;
+    const queued = item.message.role === "user" && queuedTurnKeys?.has(
+      JSON.stringify([item.message.sourceSessionId, item.message.sourceTurnId]));
+    const focused = messageFocus?.sessionId === item.message.sourceSessionId
+      && messageFocus?.messageId === item.message.sourceMessageId;
+    return !queued || focused;
+  }));
   const messageKey = (message: Extract<WorkspaceTimelineItem, {kind: "message"}>["message"]) =>
     JSON.stringify([message.sourceSessionId, message.sourceMessageId]);
   const [historical, setHistorical] = useState<Set<string> | null>(null);
@@ -135,7 +145,7 @@ export function ChannelTimeline({
     item.kind === "run" && ["queued", "running", "completed"].includes(item.run.status));
   const scheduleItems = items.filter((item): item is Extract<WorkspaceTimelineItem, { kind: "schedule" }> => item.kind === "schedule");
   const backgroundIds = new Set([...routineRuns, ...scheduleItems].map(item => item.id));
-  const primaryItems = items.filter(item => item.kind !== "proposal" && !backgroundIds.has(item.id));
+  const primaryItems = orderedItems.filter(item => item.kind !== "proposal" && !backgroundIds.has(item.id));
   const pausedScheduleCount = scheduleItems.filter(item => item.schedule.status === "paused").length;
   const enabledScheduleCount = scheduleItems.length - pausedScheduleCount;
   const workingCount = routineRuns.filter(item => item.run.status === "running" && Boolean(item.run.sessionId) && Boolean(item.run.canInterrupt)).length;

apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.tsx

@@ -0,0 +1,94 @@
+import { useEffect, useRef, useState } from "react";
+import { CornerUpLeft, RefreshCw } from "lucide-react";
+import { ChatApiError, fetchChatSessionHistory, steerQueuedChatTurn, type ChatQueuedTurn, type ChatVisibleMessage } from "../../data/chat";
+import { composerKeyAction } from "./composer-keyboard";
+import { DocumentAttachment, isTextDocument } from "./document-attachment";
+import { ContextHelp } from "./context-help";
+import { useWorkspaceI18n } from "./i18n";
+import "./conversation-queue.css";
+
+function QueuedMessage({sessionId, turn, index, targetTurnId, onChange}: {
+  sessionId: string; turn: ChatQueuedTurn; index: number; targetTurnId?: string; onChange: () => void;
+}) {
+  const { locale } = useWorkspaceI18n();
+  const zh = locale === "zh-CN";
+  const [open, setOpen] = useState(false), [retry, setRetry] = useState(0);
+  const [message, setMessage] = useState<ChatVisibleMessage | null>(null);
+  const [failed, setFailed] = useState(false);
+  const lock = useRef(false);
+  const [busy, setBusy] = useState(false), [actionError, setActionError] = useState("");
+  const held = turn.steering?.status === "pending" ? turn.steering : null;
+  const canSteer = Boolean(held || (targetTurnId && !turn.attachment_count));
+  async function steer() {
+    if (lock.current || !canSteer) return;
+    lock.current = true; setBusy(true); setActionError("");
+    try {
+      await steerQueuedChatTurn(sessionId, turn.turn_id, held?.target_turn_id ?? targetTurnId!,
+        held?.operation_id ?? crypto.randomUUID());
+    } catch (cause) {
+      const code = cause instanceof ChatApiError ? cause.payload.error_code : "";
+      setActionError(code === "queue_steering_source_unavailable"
+        ? (zh ? "此消息须沿原协作入口处理，已保留在队列中。" : "Use this message's original collaboration entry. It is still queued.")
+        : code === "queue_turn_already_started" || code === "live_steering_turn_mismatch"
+        ? (zh ? "回合已变化，请查看当前状态。" : "The turn changed. Check its current state.")
+        : (zh ? "尚未确认插入，原消息已保留。" : "Insertion is unconfirmed. The original message is retained."));
+    } finally {lock.current = false; setBusy(false); onChange();}
+  }
+  useEffect(() => {
+    if (!open || message) return;
+    const controller = new AbortController();
+    setFailed(false);
+    void fetchChatSessionHistory(sessionId, AbortSignal.any([controller.signal, AbortSignal.timeout(5000)]), turn.turn_id)
+      .then(snapshot => {
+        const original = snapshot.messages.find(row => row.role === "user" && row.turn_id === turn.turn_id);
+        if (!original) throw new Error("Original queued input is unavailable");
+        if (!controller.signal.aborted) setMessage(original);
+      }).catch(() => { if (!controller.signal.aborted) setFailed(true); });
+    return () => controller.abort();
+  }, [open, message, sessionId, turn.turn_id, retry]);
+  return <li onKeyDown={event => {
+    if (composerKeyAction({...event, isComposing: event.nativeEvent.isComposing, keyCode: event.nativeEvent.keyCode}, false) === "steer" && canSteer) {
+      event.preventDefault(); event.stopPropagation(); void steer();
+    }
+  }}><div className="personal-queue-row"><details onToggle={event => setOpen(event.currentTarget.open)}>
+    <summary aria-label={`${zh ? "待处理消息" : "Queued message"} ${index + 1}`}>
+      <span>{turn.preview || (zh ? "附件消息" : "Attachments")}</span>
+      {turn.attachment_count ? <small>{zh ? `${turn.attachment_count} 个附件` : `${turn.attachment_count} attachments`}</small> : null}
+    </summary>
+    {open ? <div className="personal-queue-message">
+      {failed ? <p role="alert">{zh ? "完整消息读取失败" : "Full message unavailable"}<button type="button" onClick={() => setRetry(value => value + 1)}>{zh ? "重试" : "Retry"}</button></p>
+        : !message ? <p role="status">{zh ? "正在读取完整消息…" : "Reading the full message…"}</p>
+        : <><p>{message.reply_context?.source_text ?? message.text}</p>
+          {message.attachments?.map(attachment => isTextDocument(attachment.mime_type)
+            ? <DocumentAttachment key={attachment.id} dataUrl={attachment.data_url} name={attachment.name} size={attachment.size} zh={zh}/>
+            : <img key={attachment.id} src={attachment.data_url} alt={attachment.name}/>)}</>}
+    </div> : null}
+  </details><button type="button" className="personal-queue-steer" onClick={() => void steer()}
+    disabled={busy || !canSteer} aria-label={zh ? `插入本轮：待处理消息 ${index + 1}` : `Steer with queued message ${index + 1}`}
+    title={turn.attachment_count ? (zh ? "原生本轮插入仅支持文本；附件仍在队列中" : "Native steering supports text only; attachments stay queued")
+      : !targetTurnId && !held ? (zh ? "当前回合或执行器不支持插入，消息仍在队列中" : "This turn or executor cannot be steered; the message stays queued")
+        : (zh ? "插入本轮 · Command+Enter" : "Steer · Command+Enter")}>
+    {busy ? <RefreshCw size={14}/> : <CornerUpLeft size={14}/>}<span>{held ? (zh ? "核对插入" : "Check insertion") : (zh ? "插入本轮" : "Steer")}</span>
+  </button></div>
+    {held || actionError || turn.steering?.status === "failed" ? <p className="personal-queue-feedback" role={actionError ? "alert" : "status"}>
+      {actionError || (held ? (zh ? "插入尚未确认，已暂停自动出队。" : "Insertion is unconfirmed; automatic dispatch is paused.")
+        : (zh ? "上次插入未送达，消息仍在队列中。" : "The previous insertion was not delivered. The message is still queued."))}
+    </p> : null}
+  </li>;
+}
+
+export function ConversationQueue({sessionId, turns, error, onRetry, targetTurnId}: {
+  sessionId: string; turns: ChatQueuedTurn[]; error: Error | null; onRetry: () => void; targetTurnId?: string;
+}) {
+  const { locale } = useWorkspaceI18n();
+  const zh = locale === "zh-CN";
+  if (!turns.length && !error) return null;
+  return <section className="personal-conversation-queue" aria-label={zh ? "待处理消息" : "Queued messages"}>
+    <header><span>{zh ? "待处理" : "Queued"}{error ? "" : ` · ${turns.length}`}</span>
+      {error ? <><span role="alert">{zh ? "读取失败" : "Read failed"}</span><button type="button" onClick={onRetry}
+        aria-label={zh ? "重试读取待处理消息" : "Retry queued messages"}><RefreshCw size={14}/></button>
+        <ContextHelp label={zh ? "队列读取失败原因" : "Queue read error"}><p>{error.message}</p></ContextHelp></> : null}</header>
+    <ol>{turns.map((turn, index) => <QueuedMessage key={`${sessionId}:${turn.turn_id}`} sessionId={sessionId} turn={turn} index={index}
+      targetTurnId={targetTurnId} onChange={onRetry}/>)}</ol>
+  </section>;
+}

apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts

@@ -393,6 +393,8 @@ export type WorkspaceActionPreview = {
 };
 
 export type WorkspaceMessage = {
+  /** A native queue receipt, not an assistant answer or active execution. */
+  queued?: boolean;
   replyContext?: ChatReplyContext;
   createdAt?: string;
   goalDraft?: GoalDraft | null;
@@ -573,6 +575,7 @@ export type PersonalWorkspaceCallbacks = {
   onInterruptConversationTurn?: (contextId: string, turnId: string) => Promise<void>;
   onSteerConversationTurn?: (contextId: string, turnId: string, message: string, ingressId: string) => Promise<void>;
   onQueueConversationTurn?: (contextId: string, sessionId: string, message: string, ingressId: string, attachments?: WorkspaceImageAttachment[]) => Promise<void>;
+  onObserveConversationTurn?: (contextId: string, sessionId: string, turnId: string) => Promise<void>;
   onOpenGoal?: (goalId: string) => void | Promise<void>;
   onOpenGoalView?: (tab: WorkspaceGoalTab) => void;
   onOpenRunSession?: (run: WorkspaceRun) => void | Promise<void>;