From 05cd52561518aa9db7aaeeab1e234309d091b82b Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 00:51:22 +0800 Subject: [PATCH 01/11] feat(chat): steer persistent queued messages into active turns Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- apps/presentation/dashboard/src/data/chat.ts | 47 ++++- .../src/data/use-conversation-queue.ts | 39 ++++ .../personal-workspace/channel-timeline.tsx | 14 +- .../personal-workspace/conversation-queue.css | 18 ++ .../personal-workspace/conversation-queue.tsx | 94 +++++++++ .../personal-workspace-model.ts | 2 + .../personal-workspace-page.tsx | 24 ++- .../personal-workspace/personal-workspace.css | 2 + .../dashboard/src/views/dashboard-page.tsx | 15 +- .../composer-session-admission.mjs | 11 +- .../personal-workspace-browser/fixture.mjs | 6 + loopx/capabilities/native_chat/queue_api.py | 59 ++++++ loopx/chat_ingress.py | 77 ++++++- loopx/chat_runtime.py | 75 +++++-- loopx/chat_server.py | 6 + loopx/control_plane/chat_turn_acceptance.py | 35 +++ .../control_plane/effect_runtime_handlers.ts | 2 + .../turn_driver/chat_queue_steering.ts | 99 +++++++++ .../chat_queue_steering.test.ts | 49 +++++ tests/presentation/test_chat_turn_queue.py | 199 ++++++++++++++++++ 20 files changed, 829 insertions(+), 44 deletions(-) create mode 100644 apps/presentation/dashboard/src/data/use-conversation-queue.ts create mode 100644 apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css create mode 100644 apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.tsx create mode 100644 loopx/capabilities/native_chat/queue_api.py create mode 100644 loopx/control_plane/turn_driver/chat_queue_steering.ts create mode 100644 tests/control_plane_ts/chat_queue_steering.test.ts diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index 851e627cff..9901dbbfc0 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -965,6 +965,44 @@ export async function fetchChatSession(sessionId: string, signal?: AbortSignal) return requestJson(`/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), + 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["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( + `/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.turns; +} + +export async function steerQueuedChatTurn(sessionId: string, queuedId: string, targetId: string, ingressId: string) { + const value = await requestJson>( + `/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(), @@ -1574,6 +1612,7 @@ async function receiveChatTurnStreaming( } = {}, ) { let finalResponse: unknown = null; + let steeredIntoTurnId: string | undefined; const outcome: { failure: Record | null; interrupted: Record | null; @@ -1591,7 +1630,10 @@ async function receiveChatTurnStreaming( if (label) options.onActivity?.(label, parseTurnStep(event.payload.step)); } if (event.kind === "turn.completed") { - finalResponse = event.payload.response; + if (typeof event.payload.steered_into_turn_id === "string" && event.payload.steered_into_turn_id) { + steeredIntoTurnId = event.payload.steered_into_turn_id; + finalResponse = {schema_version: "loopx_chat_agent_response_v0", message: "", proposals: [], gate: null}; + } else finalResponse = event.payload.response; } if (event.kind === "turn.failed") { outcome.failure = event.payload; @@ -1632,6 +1674,7 @@ async function receiveChatTurnStreaming( response: agentResponseSchema.parse(finalResponse), sessionId, turnId, + steeredIntoTurnId, }; } @@ -1642,7 +1685,7 @@ async function receiveChatTurnStreaming( export async function readCompletedChatTurn(sessionId: string, turnId: string, signal: AbortSignal): Promise { 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; } diff --git a/apps/presentation/dashboard/src/data/use-conversation-queue.ts b/apps/presentation/dashboard/src/data/use-conversation-queue.ts new file mode 100644 index 0000000000..398d6799d7 --- /dev/null +++ b/apps/presentation/dashboard/src/data/use-conversation-queue.ts @@ -0,0 +1,39 @@ +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[]; error: Error | null} | null>(null); + useEffect(() => { + if (!sessionId) return; + const controller = new AbortController(); + let timer: ReturnType | undefined; + let failures = 0, pending = false; + const read = async () => { + try { + const turns = await fetchChatQueue(sessionId, AbortSignal.any([controller.signal, AbortSignal.timeout(5000)])); + if (controller.signal.aborted) return; + failures = 0; pending = turns.length > 0; + setState({key, turns, 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: [], error: cause instanceof Error ? cause : new Error(String(cause))}); + } + if (!controller.signal.aborted && (active || pending)) { + 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 ?? [], error: current?.error ?? null, + retry: () => setAttempt(value => value + 1)}; +} diff --git a/apps/presentation/dashboard/src/features/personal-workspace/channel-timeline.tsx b/apps/presentation/dashboard/src/features/personal-workspace/channel-timeline.tsx index ab14063ef2..ecdbbdab7d 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/channel-timeline.tsx +++ b/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; 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["message"]) => JSON.stringify([message.sourceSessionId, message.sourceMessageId]); const [historical, setHistorical] = useState | 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 => 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; diff --git a/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css b/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css new file mode 100644 index 0000000000..d097d6bcda --- /dev/null +++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css @@ -0,0 +1,18 @@ +.personal-conversation-queue { min-width: 0; margin: 0 0 8px; font-size: var(--text-caption); color: var(--pw-text); } +.personal-conversation-queue header { display: flex; align-items: center; gap: 8px; margin: 0 8px 4px; color: var(--pw-muted); } +.personal-conversation-queue ol { max-height: min(28dvh, 220px); overflow: auto; overscroll-behavior: contain; list-style: none; padding: 0; margin: 0; border: 1px solid var(--pw-line-strong); border-radius: 12px; background: #fff; } +.personal-conversation-queue li + li { border-top: 1px solid var(--pw-line-strong); } +.personal-queue-row { display: flex; align-items: start; gap: 8px; padding: 8px 10px; } +.personal-queue-row details { min-width: 0; flex: 1; } +.personal-queue-row summary { cursor: pointer; } +.personal-queue-row summary > span { display: -webkit-inline-box; max-width: calc(100% - 16px); vertical-align: top; -webkit-line-clamp: 2; -webkit-box-orient: vertical; overflow: hidden; overflow-wrap: anywhere; white-space: pre-wrap; } +.personal-queue-row summary small { display: block; margin: 4px 0 0 16px; color: var(--pw-muted); } +.personal-conversation-queue button { display: inline-flex; align-items: center; justify-content: center; gap: 4px; min-height: 32px; padding: 4px 6px; border: 0; border-radius: 6px; color: var(--pw-muted); background: transparent; cursor: pointer; } +.personal-conversation-queue button:hover { background: var(--pw-blue-soft); color: var(--pw-blue-ink); } +.personal-conversation-queue button:disabled { opacity: .45; cursor: default; } +.personal-conversation-queue button:focus-visible, .personal-queue-row summary:focus-visible { outline: 2px solid var(--pw-blue); outline-offset: 2px; } +.personal-queue-steer { flex-shrink: 0; } +.personal-queue-message { margin: 8px 0 0; } +.personal-queue-message p { white-space: pre-wrap; overflow-wrap: anywhere; } +.personal-queue-message img { display: block; max-width: 100%; max-height: 120px; object-fit: contain; } +.personal-queue-feedback { margin: 0; padding: 0 10px 8px; color: var(--pw-muted); overflow-wrap: anywhere; } diff --git a/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.tsx b/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.tsx new file mode 100644 index 0000000000..ee44b325d2 --- /dev/null +++ b/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(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
  • { + if (composerKeyAction({...event, isComposing: event.nativeEvent.isComposing, keyCode: event.nativeEvent.keyCode}, false) === "steer" && canSteer) { + event.preventDefault(); event.stopPropagation(); void steer(); + } + }}>
    setOpen(event.currentTarget.open)}> + + {turn.preview || (zh ? "附件消息" : "Attachments")} + {turn.attachment_count ? {zh ? `${turn.attachment_count} 个附件` : `${turn.attachment_count} attachments`} : null} + + {open ?
    + {failed ?

    {zh ? "完整消息读取失败" : "Full message unavailable"}

    + : !message ?

    {zh ? "正在读取完整消息…" : "Reading the full message…"}

    + : <>

    {message.reply_context?.source_text ?? message.text}

    + {message.attachments?.map(attachment => isTextDocument(attachment.mime_type) + ? + : {attachment.name}/)}} +
    : null} +
    + {held || actionError || turn.steering?.status === "failed" ?

    + {actionError || (held ? (zh ? "插入尚未确认，已暂停自动出队。" : "Insertion is unconfirmed; automatic dispatch is paused.") + : (zh ? "上次插入未送达，消息仍在队列中。" : "The previous insertion was not delivered. The message is still queued."))} +

    : null} +
    • ; +} + +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
    +
    {zh ? "待处理" : "Queued"}{error ? "" : ` · ${turns.length}`} + {error ? <>{zh ? "读取失败" : "Read failed"} +

    {error.message}

     : null}
    +
      {turns.map((turn, index) => )}
    +
    ; +} diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts index 5e45241f1a..482d2c89b8 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts +++ b/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; diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx index c82ef06455..6da55ae3ad 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx @@ -1,5 +1,5 @@ import {DocumentAttachment, isTextDocument} from "./document-attachment"; -import {AlertCircle, Bot, CalendarClock, CircleHelp, FileText, ListPlus, MessageCircleQuestion, Paperclip, Plus, RefreshCw, Send, X} from "lucide-react"; +import {AlertCircle, Bot, CalendarClock, CircleHelp, FileText, ListPlus, MessageCircleQuestion, Paperclip, Plus, RefreshCw, Send, Square, X} from "lucide-react"; import {ContextHelp} from "./context-help"; import {readSettingsRoute, writeSettingsRoute} from "./workspace-settings-route"; import {hasConfiguredGoalChecks} from "../../data/goal-acceptance-observation"; @@ -11,6 +11,8 @@ import { readSteeringRequest, retainSteeringRequest, retireSteeringRequest, lega import { conversationInputKey, useConversationInputState } from "./use-conversation-input-state"; import { composerKeyAction } from "./composer-keyboard"; import type { ConversationHistoryStatus } from "../../data/use-conversation-history"; +import { useConversationQueue } from "../../data/use-conversation-queue"; +import { ConversationQueue } from "./conversation-queue"; import { GoalDraftCard } from "./goal-draft-card"; import type { GoalDraft } from "../../../../../../loopx/control_plane/collaboration/goal_draft.js"; import { CollaborationCards } from "./collaboration-card"; @@ -1391,6 +1393,13 @@ export function PersonalWorkspacePage({ && callbacks.onSteerConversationTurn ? runningMessage?.sourceTurnId : undefined; const queueTurnId = conversationTurnRunning && !readOnly && callbacks.onQueueConversationTurn ? runningMessage?.sourceTurnId : undefined; + const conversationQueue = useConversationQueue({ + sessionId: !readOnly && !workspaceConversations?.entryGate ? conversationSessionId : undefined, + scope: `${statusSourceControl?.activeSource.statusUrl ?? "/status.json"}:${conversationKey}`, + active: Boolean(runningMessage || selectedAgentModelReadback?.active_turn_id), + refreshKey: `${managerMessages.length}:${Boolean(managerMessages.at(-1)?.pending)}:${steering}`, + }); + const queuedTurnKeys = new Set(conversationQueue.turns.map(turn => JSON.stringify([conversationSessionId, turn.turn_id]))); const legacyDrafts = conversationSessionId ? legacySteeringRequests(conversationSessionId, composerDraftKey) : []; function restoreLegacyDraft(key: string) { if (composer.trim() || !conversationSessionId) return; @@ -1435,6 +1444,7 @@ export function PersonalWorkspacePage({ ); const conversationOpen = (!agentWorkActive || managedWorkActive) && (selectedGoal ? selectedGoalTab === "chat" : managerChatOpen); const timelineReading = { + queuedTurnKeys, messageFocus: conversationMessageFocus, historyPhase: conversationHistoryState?.phase, onRetryHistory: conversationHistoryState?.retry, @@ -2484,7 +2494,6 @@ export function PersonalWorkspacePage({ const acceptedImages = new Set(request.attachments?.map(image => image.id)); setImageAttachments(current => current.filter(image => !acceptedImages.has(image.id))); } - setActionFeedback(locale === "zh-CN" ? "已排队，将在当前回合结束后处理。" : "Queued for the next turn."); } catch (error) { if (error instanceof ChatApiError && error.payload.delivery_state === "not_delivered") retireSteeringRequest(composerDraftKey, request.id); setActionFeedback(error instanceof Error ? error.message : t("feedback.sendGenericError")); @@ -3200,9 +3209,8 @@ export function PersonalWorkspacePage({ {retryingSteering && retryingSteering.turnId !== steeringTurnId ?

    {locale === "zh-CN" ? "上次投递尚未确认，发送将重试原请求。" : "The previous delivery is unconfirmed. Send retries the original request."}

    : null} - {conversationTurnRunning ?

    {queueTurnId - ? (locale === "zh-CN" ? "本轮进行中 · 新消息将在下一轮处理" : "Turn in progress · new messages queue for the next turn") - : t("composer.turnRunning")}

    : null} + {conversationSessionId ? : null}
    { @@ -3271,6 +3279,12 @@ export function PersonalWorkspacePage({ onEffort={effort => saveModelSettings(() => callbacks.onSelectAgentReasoningEffort?.(selectedAgentId, effort))} onDefaults={selectedGoal && !originalConversation ? () => openSettings({kind: "settings", tab: "conversation"}, true) : undefined} /> : null} + {queueTurnId && callbacks.onInterruptConversationTurn ? : null}
    diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css index 0028f0a23f..b5ed92ce26 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css +++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace.css @@ -2243,6 +2243,8 @@ button.personal-execution-chip:focus-visible { outline: 2px solid #0070f3; outli .personal-composer-toolbar .goal-loopx-message-mode { margin: 0; min-width: 0; } .personal-composer-toolbar .goal-loopx-message-mode select { max-width: 140px; min-height: 36px; border: 0; background: transparent; color: var(--pw-text); } .personal-composer-toolbar > .personal-composer-send { flex: none; margin-left: auto; } +.personal-composer-stop { display: grid; place-items: center; flex: none; width: 36px; height: 36px; padding: 0; border: 1px solid var(--pw-line-strong); border-radius: 10px; background: #fff; color: var(--pw-text); cursor: pointer; } +.personal-composer-stop:focus-visible { outline: 2px solid var(--pw-blue); outline-offset: 2px; } @media (max-width: 760px) { .personal-composer-runtime .personal-chat-model-menu { position: static; max-width: min(140px, 36vw); } .personal-composer-runtime .personal-chat-model-panel { left: 0; right: 0; width: auto; bottom: calc(100% + 8px); } diff --git a/apps/presentation/dashboard/src/views/dashboard-page.tsx b/apps/presentation/dashboard/src/views/dashboard-page.tsx index 5cf8a8878f..3188a1f200 100644 --- a/apps/presentation/dashboard/src/views/dashboard-page.tsx +++ b/apps/presentation/dashboard/src/views/dashboard-page.tsx @@ -531,6 +531,7 @@ type PersonalHomeModel = { workers?: WorkspaceWorker[]; }; type PersonalManagerMessage = { + queued?: boolean; replyContext?: ChatVisibleMessage["reply_context"]; goalDraft?: GoalDraft | null; sourceMessageId?: string; @@ -2408,7 +2409,7 @@ function PersonalGoalHome({ })(), onDelta: (delta: string) => { streamedText += delta; - updateConversationMessage(targetContextId, streamingMessageId, { text: streamedText }); + updateConversationMessage(targetContextId, streamingMessageId, { text: streamedText, queued: false }); }, onActivity: (label: string, step: TurnStep | null) => { setMessagesByContext((messages) => ({ @@ -2432,14 +2433,14 @@ function PersonalGoalHome({ if (phase === "turn.queued") { updateConversationMessage(targetContextId, streamingMessageId, { pending: false, preparing: false, endedAt: undefined, - text: locale === "zh-CN" ? "已排队，等待当前回合结束。" : "Queued for the next turn.", activity: [], + text: "", queued: true, activity: [], }); return; } // Queued is a durable receipt, not observed execution. Only the // original worker's start/stream may take the active controls. if (phase === "turn.started") { - updateConversationMessage(targetContextId, streamingMessageId, { pending: true, startedAt: Date.now(), endedAt: undefined, text: "" }); + updateConversationMessage(targetContextId, streamingMessageId, { queued: false, pending: true, startedAt: Date.now(), endedAt: undefined, text: "" }); const previousActive = activeTurnIds.current.get(targetContextId); // Replaying an acknowledged queued Turn can include a historical // start. Its original Session, not that replay, owns live controls. @@ -2485,11 +2486,12 @@ function PersonalGoalHome({ } const response = streamed.response; updateConversationMessage(targetContextId, streamingMessageId, { + queued: Boolean(streamed.steeredIntoTurnId), goalDraft: response.goal_draft, lines: response.gate ? [response.gate.summary, response.gate.next_action].filter(Boolean).slice(0, 2) : [], pending: false, - text: visibleAgentMessage(response.message || streamedText.trim()) - || `${answerIdentityLabel(targetContextId, selectedRoute.label)} 已完成分析。`, + text: streamed.steeredIntoTurnId ? "" : (visibleAgentMessage(response.message || streamedText.trim()) + || `${answerIdentityLabel(targetContextId, selectedRoute.label)} 已完成分析。`), }); // The completed transcript is the immutable answer owner. Resolve its // stored identity before offering a link; a failed read never hides the @@ -2543,6 +2545,7 @@ function PersonalGoalHome({ || (error instanceof ChatApiError && error.payload.error_code === "turn_interrupted"); if (userInterrupted) { const interruptedMessage = { + queued: false, agentLabel: answerIdentityLabel(targetContextId, selectedRoute.label), lines: [], pending: false, @@ -2617,6 +2620,7 @@ function PersonalGoalHome({ // an unacknowledged dispatch after that boundary remains uncertain. const preparationFailed = preparationControllers.current.get(targetContextId) === preparationController; const failureMessage = { + queued: false, agentLabel: answerIdentityLabel(targetContextId, selectedRoute.label), lines: gateSummary ? [gateSummary] : [], pending: false, @@ -2905,6 +2909,7 @@ function PersonalGoalHome({ attachments: message.attachments, id: String(message.id), pending: message.pending, + queued: message.queued, preparing: message.preparing, startedAt: message.startedAt, updatedAt: message.updatedAt, diff --git a/examples/personal-workspace-browser/composer-session-admission.mjs b/examples/personal-workspace-browser/composer-session-admission.mjs index 649738c2bd..3d8bde9da2 100644 --- a/examples/personal-workspace-browser/composer-session-admission.mjs +++ b/examples/personal-workspace-browser/composer-session-admission.mjs @@ -19,7 +19,7 @@ export const composerSessionAdmissionScenario = { const notes = []; const composerInput = page.getByLabel("向 LoopX 发送消息"); const sendButton = page.getByRole("button", { name: "发送", exact: true }); - const turnRunningHint = page.locator(".personal-composer-status", { hasText: "新消息将在下一轮处理" }); + const turnRunningHint = page.getByRole("button", {name: "中断本轮", exact: true}); const pasteDraftImage = async (name) => { await composerInput.evaluate((target, filename) => { const png = Uint8Array.from(atob("iVBORw0KGgoAAAANSUhEUgAAAAEAAAABCAQAAAC1HAwCAAAAC0lEQVR42mP8/x8AAwMCAO+jN3sAAAAASUVORK5CYII="), (char) => char.charCodeAt(0)); @@ -48,6 +48,7 @@ export const composerSessionAdmissionScenario = { // Start the Turn outside this page and reload, so the page learns of it // only from the Session read. Its stream is held so the Turn stays running. const reloadWithRunningTurn = async (sessionMode, turnId, adapterKind = "codex_app_server") => { + page.__loopxRuntime.queuedTurns.set(sessionId, []); page.__loopxRuntime.turnMessages.set(turnId, "另一入口发起的运行中回合"); page.__loopxRuntime.sessions.set(sessionId, { ...page.__loopxRuntime.sessions.get(sessionId), @@ -68,6 +69,9 @@ export const composerSessionAdmissionScenario = { posts.push(body.message); postBodies.push(body); const queuedTurnId = `${turnId}-queued-${posts.length}`; + if (sessionMode === "attached_host" || body.delivery === "queue") page.__loopxRuntime.queuedTurns.get(sessionId).push({turn_id: queuedTurnId, + created_at: "2026-08-13T01:00:01Z", origin: "web", preview: body.message.slice(0, 240), + character_count: body.message.length, attachment_count: body.attachments?.length ?? 0, steering: null}); page.__loopxRuntime.messages.get(sessionId).push({ message_id: `${queuedTurnId}-user`, turn_id: queuedTurnId, role: "user", text: body.message, created_at: "2026-08-13T01:00:01Z" }); // A queued follow-up waits behind the running Turn; its stream is held too. @@ -266,7 +270,8 @@ export const composerSessionAdmissionScenario = { const queuedMessage = "当前工作完成后再检查下一项。"; await composerInput.fill(queuedMessage); await composerInput.press("Enter"); - await page.locator(".personal-action-feedback", { hasText: "已排队，将在当前回合结束后处理" }).waitFor(); + await page.locator(".personal-conversation-queue li").filter({hasText: queuedMessage}).waitFor(); + if (await page.locator(".personal-action-feedback", {hasText: "已排队"}).count()) throw new Error("Queue acceptance duplicated its content with a banner"); if (queued.posts.length !== 1 || queued.posts[0] !== queuedMessage || queued.postBodies[0].delivery !== "queue" || !queued.postBodies[0].client_turn_id || adjustments.length !== 10 || await composerInput.inputValue() @@ -289,7 +294,7 @@ export const composerSessionAdmissionScenario = { const attached = await reloadWithRunningTurn("attached_host", `turn-attached-${Date.now()}`); const followUp = "宿主回合运行中排队的下一条消息"; await composerInput.fill(followUp); - if (await turnRunningHint.count()) throw new Error("An attached host Session showed the managed-runtime running-Turn hint"); + if (await page.locator(".personal-composer-status", {hasText: "新消息将在下一轮处理"}).count()) throw new Error("The removed duplicate queue hint returned"); if (await sendButton.isDisabled()) throw new Error("An attached host Session's composer closed while the host Turn ran, though the service queues follow-ups"); await sendButton.click(); for (let attempt = 0; attempt < 100 && attached.posts.length === 0; attempt += 1) await page.waitForTimeout(50); diff --git a/examples/personal-workspace-browser/fixture.mjs b/examples/personal-workspace-browser/fixture.mjs index 7a8993e5f8..64bb616ad2 100644 --- a/examples/personal-workspace-browser/fixture.mjs +++ b/examples/personal-workspace-browser/fixture.mjs @@ -423,6 +423,7 @@ export async function installApi(page, { fixturePort = port, goalSubagentConfigu const sessions = runtime.sessions; const messages = runtime.messages; const turnMessages = runtime.turnMessages; + const queuedTurns = runtime.queuedTurns ??= new Map(); const loopxModes = runtime.loopxModes ??= new Map(); for (const proposal of initialActionProposals) { actionProposals.set(proposal.proposal_id, structuredClone(proposal)); @@ -1843,6 +1844,11 @@ export async function installApi(page, { fixturePort = port, goalSubagentConfigu await route.fulfill({ contentType: "application/json", json: configured, status: 200 }); return; } + const queue = url.pathname.match(/^\/api\/chat\/sessions\/([^/]+)\/queue$/); + if (queue && request.method() === "GET") { + await route.fulfill({json: {ok: true, schema_version: "loopx_chat_store_v1", session_id: queue[1], turns: queuedTurns.get(queue[1]) ?? []}}); + return; + } const snapshot = url.pathname.match(/^\/api\/chat\/sessions\/([^/]+)$/); if (snapshot && request.method() === "GET") { const session = sessions.get(snapshot[1]); diff --git a/loopx/capabilities/native_chat/queue_api.py b/loopx/capabilities/native_chat/queue_api.py new file mode 100644 index 0000000000..209905470b --- /dev/null +++ b/loopx/capabilities/native_chat/queue_api.py @@ -0,0 +1,59 @@ +"""HTTP projection and explicit steering of the existing persisted queue.""" + +from typing import Any + +from ...chat_store import CHAT_STORE_SCHEMA_VERSION + + +class ChatQueueRequestMixin: + def _session_queue(self, session_id: str) -> None: + try: + store = self.server.chat_store + session = store.load_session(session_id) + if session is None: + raise KeyError("chat session was not found") + if session.get("steward_context") is not None: + self.server.runtime_controller.project_contexts.session_context(session) + else: + self._session_context(session) + rows: list[dict[str, Any]] = [] + for turn in store.queued_turns(session_id, strict=True, include_reserved=True): + text = str(turn.get("message") or "") + rows.append({ + "turn_id": turn["turn_id"], + "created_at": turn["created_at"], + "origin": turn.get("origin"), + "preview": text[:240], + "character_count": len(text), + "attachment_count": len(turn.get("attachments") or []), + "steering": turn.get("_queue_steering"), + }) + self._send_json({"ok": True, "schema_version": CHAT_STORE_SCHEMA_VERSION, + "session_id": session_id, "turns": rows}) + except KeyError: + self._send_error("chat session was not found", status=404) + except ValueError as exc: + self._send_error(str(exc), status=400) + except (OSError, RuntimeError): + self._send_error("Queued messages are temporarily unavailable", status=503) + + def _steer_queued_turn(self, session_id: str, queued_turn_id: str) -> None: + try: + body = self._read_json() + if set(body) != {"target_turn_id", "client_ingress_id"}: + raise ValueError("target_turn_id and client_ingress_id are required") + for key in body: + if not isinstance(body[key], str) or not body[key] or len(body[key]) > 160: + raise ValueError("invalid queue steering identity") + result = self.server.runtime_controller.steer_queued_turn(session_id=session_id, + queued_turn_id=queued_turn_id, expected_turn_id=body["target_turn_id"], + client_ingress_id=body["client_ingress_id"]) + self._send_json({"ok": True, **result}) + except KeyError: + self._send_error("chat turn was not found", status=404) + except ValueError as exc: + self._send_error(str(exc), status=400) + except RuntimeError as exc: + self._send_error(str(exc), status=409, error_code=str(exc)) + except OSError: + self._send_error("Queue steering is temporarily unavailable; retain the original request", status=503) diff --git a/loopx/chat_ingress.py b/loopx/chat_ingress.py index 5ba04b72b8..b13ee4f076 100644 --- a/loopx/chat_ingress.py +++ b/loopx/chat_ingress.py @@ -72,8 +72,8 @@ class ChatIngressStore: def runtime_activation_lock(self): return chat_runtime_activation_lock(self.root.parent) - def queued_turns(self, session_id: str) -> list[dict[str, Any]]: - from .chat_store import _read_json, CHAT_TURN_SCHEMA_VERSION + def queued_turns(self, session_id: str, *, strict: bool = False, include_reserved: bool = False, capacity_at: str | None = None) -> list[dict[str, Any]]: + from .chat_store import _read_json, CHAT_TURN_SCHEMA_VERSION, TERMINAL_TURN_STATES session = self.load_session(session_id) if session is None: @@ -84,21 +84,82 @@ def queued_turns(self, session_id: str) -> list[dict[str, Any]]: for path in turns_dir.glob("*.json") if turns_dir.is_dir() else []: if path.name.endswith(".events.json"): continue - payload = _read_json(path) + if strict: + try: + payload = json.loads(path.read_text(encoding="utf-8")) + except (OSError, ValueError) as exc: + raise OSError("Queued Turn state is unreadable") from exc + if (not isinstance(payload, dict) + or payload.get("schema_version") != CHAT_TURN_SCHEMA_VERSION + or payload.get("session_id") != session_id + or payload.get("turn_id") != path.stem + or not isinstance(payload.get("status"), str) + or payload.get("status") not in TERMINAL_TURN_STATES | { + "queued", "starting", "running", "completing", "interrupting", + }): + raise OSError("Queued Turn state is invalid") + else: + payload = _read_json(path) if ( payload.get("schema_version") == CHAT_TURN_SCHEMA_VERSION and payload.get("status") == "queued" and str(payload.get("turn_id") or "") != active_turn_id ): rows.append(payload) + from .control_plane.chat_turn_acceptance import queued_chat_turn_ids + selected = set(queued_chat_turn_ids(rows, active_turn_id, include_reserved=include_reserved, capacity_at=capacity_at)) return sorted( - rows, + [row for row in rows if row["turn_id"] in selected], key=lambda item: ( str(item.get("created_at") or ""), str(item.get("turn_id") or ""), ), ) + def queue_steering_lock(self, session_id: str, turn_id: str): + """Use the same Session→Turn fences as automatic queue consumption.""" + from contextlib import ExitStack + stack = ExitStack() + try: + stack.enter_context(self._session_lock(session_id)) + stack.enter_context(exclusive_file_lock(self._session_path(session_id), + agent_id="loopx-chat", operation="steer_session_queue_turn")) + stack.enter_context(exclusive_file_lock(self._turn_path(session_id, turn_id), + agent_id="loopx-chat", operation="steer_queued_chat_turn")) + return stack + except BaseException: + stack.close() + raise + + def steering_receipt(self, session_id: str, ingress_id: str): + """A corrupt receipt cannot be treated as absence and resent.""" + from .chat_store import CHAT_INGRESS_SCHEMA_VERSION + path = self._ingress_path(session_id, ingress_id) + try: + value = json.loads(path.read_text(encoding="utf-8")) + except FileNotFoundError: + return None + if (not isinstance(value, dict) or value.get("schema_version") != CHAT_INGRESS_SCHEMA_VERSION + or value.get("session_id") != session_id or value.get("client_ingress_id") != ingress_id): + raise OSError("Steering receipt is unreadable") + return value + + def write_queue_steering(self, session_id: str, turn: dict[str, Any], plan: dict[str, Any]): + """Persist the typed plan while the caller holds both queue fences.""" + from .chat_store import _atomic_write_json, utc_now + if plan["action"] not in {"dispatch", "complete", "release"}: + return turn + turn = {**turn, "_queue_steering": plan["reservation"]} + if plan["action"] == "complete": + turn.update(status="completed", completed_at=turn.get("completed_at") or utc_now(), response="") + _atomic_write_json(self._turn_path(session_id, turn["turn_id"]), turn, preserve_mode=True) + if plan["action"] == "complete": + payload = {"steered_into_turn_id": plan["reservation"]["target_turn_id"]} + if not any(e.get("kind") == "turn.completed" and e.get("payload") == payload + for e in self.events_after(session_id, turn["turn_id"], None)): + self.append_event(session_id, turn["turn_id"], kind="turn.completed", payload=payload) + return turn + def complete_recovered_terminalization( self, session_id: str, turn_id: str, session: dict[str, Any] | None, turn: dict[str, Any] | None, @@ -158,7 +219,7 @@ def create_queued_turn( source revalidation and Stop remain independent bounds. """ - from .chat_store import _opaque_id, _instant, _atomic_write_json, utc_now, CHAT_TURN_SCHEMA_VERSION + from .chat_store import _opaque_id, _atomic_write_json, utc_now, CHAT_TURN_SCHEMA_VERSION client_id = _opaque_id(client_turn_id, field="client_turn_id") from .chat_attachments import normalize_chat_attachments, validate_chat_turn_envelope @@ -195,11 +256,7 @@ def create_queued_turn( if session is None or session.get("status") == "closed": raise KeyError("chat session was not found") now = datetime.now(timezone.utc) - queued = [ - turn - for turn in self.queued_turns(session_id) - if (_instant(turn.get("expires_at")) or now + timedelta(seconds=1)) > now - ] + queued = self.queued_turns(session_id, include_reserved=True, capacity_at=now.isoformat()) if len(queued) >= SESSION_QUEUE_MAX_PENDING: raise RuntimeError("session_queue_full") turn_id = uuid.uuid4().hex diff --git a/loopx/chat_runtime.py b/loopx/chat_runtime.py index ecd1b6dbb5..7b68ea45b7 100644 --- a/loopx/chat_runtime.py +++ b/loopx/chat_runtime.py @@ -1269,16 +1269,7 @@ def _start_accepted_turn_worker( raise return True - def steer_active_turn( - self, - *, - session_id: str, - client_ingress_id: str, - message: str, - expected_turn_id: str | None = None, - ) -> tuple[dict[str, Any], bool]: - """Steer the exact active Codex Turn with durable ingress deduplication.""" - + def _require_steering_session(self, session_id: str) -> dict[str, Any]: session = self.store.load_session(session_id) if session is None or session.get("status") == "closed": raise KeyError("chat session was not found") @@ -1295,6 +1286,59 @@ def steer_active_turn( registry_path=self.registry_path, session_id=session_id, ) + return session + + def steer_queued_turn( + self, *, session_id: str, queued_turn_id: str, expected_turn_id: str, + client_ingress_id: str, + ) -> dict[str, Any]: + """Transfer original queued input under the automatic consumer's fences. + + Only a proven non-delivery restores FIFO eligibility. A lost native + acknowledgement keeps the original input durably reserved. + """ + from .control_plane.chat_turn_acceptance import plan_queued_chat_steering + self._require_steering_session(session_id) + ingress_id = "queue-steer-" + hashlib.sha256( + json.dumps([session_id, queued_turn_id, client_ingress_id]).encode() + ).hexdigest()[:40] + with self.store.queue_steering_lock(session_id, queued_turn_id): + session = self.store.load_session(session_id) + turn = self.store.load_turn(session_id, queued_turn_id) + if session is None or turn is None: + raise KeyError("chat turn was not found") + active = self.store.load_turn(session_id, str(session.get("active_turn_id") or "")) if session.get("active_turn_id") else None + + def plan(phase): + return plan_queued_chat_steering(session, turn, active, + self.store.steering_receipt(session_id, ingress_id), + ingress_id=ingress_id, operation_id=client_ingress_id, target=expected_turn_id, phase=phase) + + decision = plan("prepare") + if decision["action"] == "dispatch": + turn = self.store.write_queue_steering(session_id, turn, decision) + try: + self.steer_active_turn(session_id=session_id, expected_turn_id=expected_turn_id, + client_ingress_id=ingress_id, message=turn["message"], + _queued_turn_id=queued_turn_id) + except Exception: + # The durable ingress, including absence before any send, + # owns the outcome. Exception wording cannot release work. + pass + decision = plan("settle") + self.store.write_queue_steering(session_id, turn, decision) + if decision["action"] != "complete": + raise RuntimeError(decision["error_code"]) + return {"session_id": session_id, "queued_turn_id": queued_turn_id, + "turn_id": expected_turn_id, "client_ingress_id": client_ingress_id, + "status": "delivered"} + + def steer_active_turn( + self, *, session_id: str, client_ingress_id: str, message: str, + expected_turn_id: str | None = None, _queued_turn_id: str | None = None, + ) -> tuple[dict[str, Any], bool]: + """Steer the exact active Codex Turn with durable ingress deduplication.""" + session = self._require_steering_session(session_id) receipt, created = self.store.create_ingress_receipt( session_id, client_ingress_id=client_ingress_id, @@ -1383,17 +1427,14 @@ def steer_active_turn( error_code="live_steering_rejected", ) raise - self.store.append_message( - session_id, - role="user", - text=message, - turn_id=active_turn_id, - ) + if _queued_turn_id is None: + self.store.append_message(session_id, role="user", text=message, turn_id=active_turn_id) self.store.append_event( session_id, active_turn_id, kind="turn.steered", - payload={"client_ingress_id": client_ingress_id}, + payload={"client_ingress_id": client_ingress_id, + **({"queued_turn_id": _queued_turn_id} if _queued_turn_id is not None else {})}, ) self.store.update_ingress_receipt( session_id, diff --git a/loopx/chat_server.py b/loopx/chat_server.py index 1166fe1a33..b60461503a 100644 --- a/loopx/chat_server.py +++ b/loopx/chat_server.py @@ -42,6 +42,7 @@ add_goal_subagent_routes, ) from .capabilities.native_chat.model_settings_api import ChatModelSettingsRequestMixin +from .capabilities.native_chat.queue_api import ChatQueueRequestMixin from .chat_status_api import ChatStatusRequestMixin from .chat_runtime import ( STEERING_NOT_DELIVERED_CODES, @@ -496,6 +497,7 @@ def server_close(self) -> None: class ChatRequestHandler( background_api.BackgroundServicesRequestMixin, ChatModelSettingsRequestMixin, + ChatQueueRequestMixin, GoalSourceRequestMixin, GoalAcceptanceRequestMixin, ZCodeGoalRequestMixin, @@ -1737,6 +1739,8 @@ def do_GET(self) -> None: if len(action_parts) == 3 and action_parts[:2] == ["api", "actions"]: return self._action_snapshot(action_parts[2]) session_parts = path.strip("/").split("/") + if len(session_parts) == 5 and session_parts[:3] == ["api", "chat", "sessions"] and session_parts[4] == "queue": + return self._session_queue(session_parts[3]) if len(session_parts) == 5 and session_parts[:3] == ["api", "chat", "sessions"] and session_parts[4] == "native-turn": return self._managed_native_turn(session_parts[3]) if len(session_parts) == 5 and session_parts[:3] == ["api", "chat", "sessions"] and session_parts[4] == "loopx": @@ -1761,6 +1765,8 @@ def do_POST(self) -> None: if path == "/api/chat/collaboration": return self._project_collaboration(apply=True) peer_parts = path.strip("/").split("/") + if len(peer_parts) == 7 and peer_parts[:3] == ["api", "chat", "sessions"] and peer_parts[4] == "queue" and peer_parts[6] == "steer": + return self._steer_queued_turn(peer_parts[3], peer_parts[5]) if len(peer_parts) == 5 and peer_parts[:3] == ["api", "chat", "sessions"] and peer_parts[4] == "collaboration": return self._project_collaboration(peer_parts[3], apply=True) if self._dispatch_zcode_goal(path, apply=True): diff --git a/loopx/control_plane/chat_turn_acceptance.py b/loopx/control_plane/chat_turn_acceptance.py index 008336acfc..4f0396fbcb 100644 --- a/loopx/control_plane/chat_turn_acceptance.py +++ b/loopx/control_plane/chat_turn_acceptance.py @@ -170,6 +170,41 @@ def _required_string(payload: dict[str, Any], field: str) -> str: return value +def queued_chat_turn_ids( + turns: list[dict[str, Any]], active_turn_id: str | None, *, + include_reserved: bool = False, capacity_at: str | None = None, +) -> list[str]: + """Project storage observations to the typed queue/Turn owner.""" + return effect_runtime_result("chat.queue.select", { + "active_turn_id": active_turn_id, "include_reserved": include_reserved, + "capacity_at": capacity_at, + "turns": [{"turn_id": t["turn_id"], "status": t["status"], + "expires_at": t.get("expires_at"), "reservation": t.get("_queue_steering")} for t in turns], + })["turn_ids"] + + +def plan_queued_chat_steering( + session: dict[str, Any], turn: dict[str, Any], active: dict[str, Any] | None, + receipt: dict[str, Any] | None, *, ingress_id: str, operation_id: str, target: str, phase: str, +) -> dict[str, Any]: + return effect_runtime_result("chat.queue.steer", { + "phase": phase, + "request": {"client_ingress_id": ingress_id, "operation_id": operation_id, "target_turn_id": target}, + "session": {"status": session.get("status"), "active_turn_id": session.get("active_turn_id")}, + "active_status": (active or {}).get("status"), + "turn": {"turn_id": turn["turn_id"], "status": turn["status"], + "origin": turn.get("origin"), "loopx_execution": turn.get("loopx_execution", False), + "has_collaboration_request": turn.get("collaboration_request") is not None, + "has_external_agent_target": turn.get("external_agent_target") is not None, + "attachment_count": len(turn.get("attachments") or []), + "reservation": turn.get("_queue_steering")}, + "receipt": ({key: receipt.get(key) for key in ( + "status", "mode", "expected_turn_id", "active_turn_id", "error_code", + )} if receipt is not None else None), + "receipt_message_matches": receipt is not None and receipt.get("message") == turn.get("message"), + }) + + def _required_bool(payload: dict[str, Any], field: str) -> bool: value = payload.get(field) if not isinstance(value, bool): diff --git a/loopx/control_plane/effect_runtime_handlers.ts b/loopx/control_plane/effect_runtime_handlers.ts index 2bd25a35cb..5e135a5af9 100644 --- a/loopx/control_plane/effect_runtime_handlers.ts +++ b/loopx/control_plane/effect_runtime_handlers.ts @@ -619,6 +619,8 @@ export function createEffectRuntimeHandlers( ["collaboration.source.configure_evidence_scope", lazyHandler(() => import("./collaboration/source_grants.ts"), ({configureSourceEvidenceScope}) => configureSourceEvidenceScope)], ["collaboration.conversation.reply_context", lazyHandler(() => import("./collaboration/conversation_reply_context.ts"), ({projectConversationReplyContext}) => projectConversationReplyContext)], ["chat.turn.accept", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({planChatTurnAcceptance}) => planChatTurnAcceptance)], + ["chat.queue.select", lazyHandler(() => import("./turn_driver/chat_queue_steering.ts"), ({selectQueuedChatTurns}) => selectQueuedChatTurns)], + ["chat.queue.steer", lazyHandler(() => import("./turn_driver/chat_queue_steering.ts"), ({planQueuedChatSteering}) => planQueuedChatSteering)], ["chat.turn.execution_allowed", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({mayContinueChatTurn}) => mayContinueChatTurn)], ["chat.turn.recovery_allowed", lazyHandler(() => import("./turn_driver/chat_turn_acceptance.ts"), ({mayRecoverChatTurn}) => mayRecoverChatTurn)], ["collaboration.delegation.observe", lazyHandler(() => import("./collaboration/delegation.ts"), ({transitionDelegationObservation}) => transitionDelegationObservation)], diff --git a/loopx/control_plane/turn_driver/chat_queue_steering.ts b/loopx/control_plane/turn_driver/chat_queue_steering.ts new file mode 100644 index 0000000000..69d1ddb338 --- /dev/null +++ b/loopx/control_plane/turn_driver/chat_queue_steering.ts @@ -0,0 +1,99 @@ +import type {JsonObject} from "../effect_program.ts"; +import {requireJsonObject, requireNonEmptyString} from "../runtime_decode.ts"; + +// Queue handoff is part of the existing Turn/ingress owner. A reservation is +// persisted before native I/O; uncertain I/O must never become a queued start. +export const QUEUE_STEERING_SCHEMA = "loopx_chat_queue_steering_v1"; +const NOT_DELIVERED = new Set([ + "attached_session_live_steering_unavailable", "live_steering_turn_mismatch", + "live_steering_requires_active_turn", "live_steering_session_not_attached", + "live_steering_turn_not_started", +]); + +type QueueSteeringPlan = + | {action: "dispatch" | "complete"; reservation: JsonObject} + | {action: "release"; reservation: JsonObject; error_code: string} + | {action: "reject" | "unresolved"; error_code: string}; + +function reservation(value: unknown): JsonObject | null { + if (value === null || value === undefined) return null; + const row = requireJsonObject(value, "queue steering reservation"); + if (row.schema_version !== QUEUE_STEERING_SCHEMA || + !["pending", "delivered", "failed"].includes(String(row.status))) { + throw new Error("Invalid queue steering reservation"); + } + requireNonEmptyString(row.client_ingress_id, "reservation.client_ingress_id"); + requireNonEmptyString(row.target_turn_id, "reservation.target_turn_id"); + requireNonEmptyString(row.operation_id, "reservation.operation_id"); + return row; +} + +export function selectQueuedChatTurns(input: JsonObject): JsonObject { + if (!Array.isArray(input.turns)) throw new Error("Queue turns must be an array"); + const ids = input.turns.flatMap(value => { + const turn = requireJsonObject(value, "queue turn"); + const held = reservation(turn.reservation); + if (input.capacity_at && held?.status !== "pending" && typeof turn.expires_at === "string" && + Date.parse(turn.expires_at) <= Date.parse(String(input.capacity_at))) return []; + return turn.status === "queued" && turn.turn_id !== input.active_turn_id && + (input.include_reserved === true || held?.status !== "pending") + ? [requireNonEmptyString(turn.turn_id, "turn.turn_id")] : []; + }); + return {turn_ids: ids}; +} + +export function planQueuedChatSteering(input: JsonObject): QueueSteeringPlan { + const session = requireJsonObject(input.session, "session"); + const turn = requireJsonObject(input.turn, "queued turn"); + const request = requireJsonObject(input.request, "request"); + const id = requireNonEmptyString(request.client_ingress_id, "client_ingress_id"); + const target = requireNonEmptyString(request.target_turn_id, "target_turn_id"); + const held = reservation(turn.reservation); + const reject = (code: string): QueueSteeringPlan => ({action: "reject", error_code: code}); + if (session.status === "closed") return reject("session_closed"); + if (held?.status === "pending" && + (held.client_ingress_id !== id || held.target_turn_id !== target)) { + return reject("queue_steering_in_progress"); + } + const same = held?.client_ingress_id === id && held.target_turn_id === target; + // Settlement/replay precedes current active-Turn checks: an acknowledged + // native send remains acknowledged after that Turn finishes or changes. + if (same && held.status === "delivered") return {action: "complete", reservation: held}; + const receipt = input.receipt == null ? null : requireJsonObject(input.receipt, "ingress receipt"); + if (receipt) { + if (!same || receipt.expected_turn_id !== target || receipt.mode !== "live_steering" || + input.receipt_message_matches !== true) return reject("queue_steering_receipt_conflict"); + if (receipt.status === "delivered" && receipt.active_turn_id === target) { + return {action: "complete", reservation: {...held, status: "delivered"}}; + } + if (receipt.status === "failed" && NOT_DELIVERED.has(String(receipt.error_code))) { + return {action: "release", error_code: String(receipt.error_code), + reservation: {...held, status: "failed", error_code: receipt.error_code}}; + } + return {action: "unresolved", error_code: "live_steering_delivery_unresolved"}; + } + if (input.phase === "settle") { + // No ingress was reserved: no provider call could have happened. + return {action: "release", error_code: "live_steering_session_not_attached", + reservation: {...held, status: "failed"}}; + } + if (turn.status !== "queued" || session.active_turn_id === turn.turn_id) { + return reject("queue_turn_already_started"); + } + if (!["web", "external"].includes(String(turn.origin)) || turn.has_collaboration_request === true || + turn.has_external_agent_target === true || turn.loopx_execution === true) { + return reject("queue_steering_source_unavailable"); + } + if (session.active_turn_id !== target || !["starting", "running"].includes(String(input.active_status))) { + if (same && held.status === "pending") return {action: "release", + error_code: "live_steering_turn_mismatch", reservation: {...held, status: "failed"}}; + return reject("live_steering_turn_mismatch"); + } + // Native live steering currently accepts text only. Retain attachments and + // the pending Turn instead of silently dropping them during conversion. + if (turn.attachment_count !== 0) return reject("queue_steering_attachments_unavailable"); + return {action: "dispatch", reservation: { + schema_version: QUEUE_STEERING_SCHEMA, client_ingress_id: id, + target_turn_id: target, operation_id: requireNonEmptyString(request.operation_id, "operation_id"), status: "pending", + }}; +} diff --git a/tests/control_plane_ts/chat_queue_steering.test.ts b/tests/control_plane_ts/chat_queue_steering.test.ts new file mode 100644 index 0000000000..41681d24ab --- /dev/null +++ b/tests/control_plane_ts/chat_queue_steering.test.ts @@ -0,0 +1,49 @@ +import assert from "node:assert/strict"; +import test from "node:test"; +import {planQueuedChatSteering, selectQueuedChatTurns, QUEUE_STEERING_SCHEMA} from "../../loopx/control_plane/turn_driver/chat_queue_steering.ts"; + +function facts() { + return {phase: "prepare", session: {status: "busy", active_turn_id: "active"}, active_status: "running", + turn: {turn_id: "queued", status: "queued", origin: "web", attachment_count: 0, reservation: null as null | object}, + request: {client_ingress_id: "native-op", operation_id: "op", target_turn_id: "active"}, + receipt: null as null | object, receipt_message_matches: true}; +} +const held = {schema_version: QUEUE_STEERING_SCHEMA, status: "pending", client_ingress_id: "native-op", operation_id: "op", target_turn_id: "active"}; +const receipt = {status: "delivered", expected_turn_id: "active", active_turn_id: "active", mode: "live_steering"}; + +test("queued steering requires an unclaimed original and exact live target", () => { + const f = facts(); assert.equal(planQueuedChatSteering(f).action, "dispatch"); + for (const status of ["starting", "running", "interrupting", "completed", "failed"]) { + assert.equal(planQueuedChatSteering({...f, turn: {...f.turn, status}}).action, "reject"); + } + assert.equal(planQueuedChatSteering({...f, session: {...f.session, active_turn_id: "queued"}}).action, "reject"); + assert.equal(planQueuedChatSteering({...f, session: {...f.session, status: "closed"}}).action, "reject"); + assert.deepEqual(planQueuedChatSteering({...f, turn: {...f.turn, attachment_count: 1}}), {action: "reject", error_code: "queue_steering_attachments_unavailable"}); + for (const change of [{origin: "peer"}, {has_collaboration_request: true}, {has_external_agent_target: true}, {loopx_execution: true}]) { + assert.deepEqual(planQueuedChatSteering({...f, turn: {...f.turn, ...change}}), {action: "reject", error_code: "queue_steering_source_unavailable"}); + } +}); + +test("uncertain delivery never dispatches again or becomes eligible FIFO work", () => { + const f = facts(); f.turn.reservation = held; + for (const state of ["pending", "failed"]) { + assert.equal(planQueuedChatSteering({...f, receipt: {...receipt, status: state}}).action, "unresolved"); + } + assert.deepEqual(planQueuedChatSteering({...f, request: {...f.request, client_ingress_id: "other"}}), {action: "reject", error_code: "queue_steering_in_progress"}); + const turns = [{turn_id: "queued", status: "queued", reservation: held}]; + assert.deepEqual(selectQueuedChatTurns({turns}).turn_ids, []); + assert.deepEqual(selectQueuedChatTurns({turns, include_reserved: true}).turn_ids, ["queued"]); + const expired = {...turns[0], expires_at: "2025-01-01T00:00:00Z"}; + assert.deepEqual(selectQueuedChatTurns({turns: [expired], include_reserved: true, capacity_at: "2026-01-01T00:00:00Z"}).turn_ids, ["queued"]); + assert.deepEqual(selectQueuedChatTurns({turns: [{...expired, reservation: null}], include_reserved: true, capacity_at: "2026-01-01T00:00:00Z"}).turn_ids, []); +}); + +test("acknowledgement recovers after target completion while proven rejection restores queue", () => { + const f = facts(); f.turn.reservation = held; + assert.equal(planQueuedChatSteering({...f, session: {status: "ready", active_turn_id: null}, receipt}).action, "complete"); + assert.equal(planQueuedChatSteering({...f, receipt: {...receipt, expected_turn_id: "other"}}).action, "reject"); + assert.equal(planQueuedChatSteering({...f, receipt_message_matches: false, receipt}).action, "reject"); + assert.equal(planQueuedChatSteering({...f, receipt: {...receipt, status: "failed", error_code: "live_steering_turn_mismatch"}}).action, "release"); + assert.equal(planQueuedChatSteering({...f, session: {status: "ready", active_turn_id: null}}).action, "release"); + assert.equal(planQueuedChatSteering({...f, phase: "settle"}).action, "release"); +}); diff --git a/tests/presentation/test_chat_turn_queue.py b/tests/presentation/test_chat_turn_queue.py index ff39c03e8c..07326cf32d 100644 --- a/tests/presentation/test_chat_turn_queue.py +++ b/tests/presentation/test_chat_turn_queue.py @@ -7,6 +7,205 @@ from test_chat_ordinary_project import ordinary # noqa: F401 +def held_queue(ordinary, text="Check the second source.", attachments=None): + store, runtime, contexts, request, capture, *_ = ordinary + sid = request("/api/chat/sessions", {"context_kind": "project", + "project_ref": contexts.available()[0]["project_ref"]})[1]["session_id"] + first = request(f"/api/chat/sessions/{sid}/turns", { + "message": "wait for steer", "client_turn_id": "held"})[1] + body = {"message": text, "client_turn_id": "pending", "delivery": "queue"} + if attachments: + body["attachments"] = attachments + code, queued = request(f"/api/chat/sessions/{sid}/turns", body) + assert code == 202 + return sid, first["turn_id"], queued["turn_id"], body + + +def test_queue_read_is_ordered_bounded_read_only_and_full_input_remains_readable(ordinary): # noqa: F811 + store, runtime, contexts, request, capture, *_ = ordinary + text = ("A long original input.\n" * 600).strip() + sid, active, queued, _ = held_queue(ordinary, text) + before = capture.read_bytes() + code, snapshot = request(f"/api/chat/sessions/{sid}/queue") + assert code == 200 and snapshot["session_id"] == sid + assert [(t["turn_id"], t["character_count"], t["preview"]) for t in snapshot["turns"]] == [(queued, len(text), text[:240])] + assert snapshot["turns"][0]["steering"] is None + assert store.load_session(sid)["active_turn_id"] == active + assert capture.read_bytes() == before + assert request(f"/api/chat/sessions/{sid}?turn_id={queued}&message_limit=200")[1]["messages"][0]["text"] == text + # Corruption cannot quietly turn an unknown queue into an empty success. + original = store.load_turn(sid, queued) + for invalid in [{**original, "status": "invented"}, {**original, "status": []}, + {**original, "_queue_steering": {"schema_version": "invalid"}}]: + store._turn_path(sid, queued).write_text(json.dumps(invalid)) + assert request(f"/api/chat/sessions/{sid}/queue")[0] == 503 + store._turn_path(sid, queued).write_text("{") + assert request(f"/api/chat/sessions/{sid}/queue")[0] == 503 + + +def test_queued_steering_uses_original_input_once_and_replays_after_target_finished(ordinary): # noqa: F811 + store, runtime, contexts, request, capture, *_ = ordinary + sid, active, queued, original = held_queue(ordinary) + path = f"/api/chat/sessions/{sid}/queue/{queued}/steer" + body = {"target_turn_id": active, "client_ingress_id": "insert-original"} + assert request(path)[0] == 404 # GET never mutates a queue. + code, result = request(path, body) + assert code == 200 and result["status"] == "delivered" + assert runtime.wait_for_turn(session_id=sid, turn_id=active, timeout_sec=10)["status"] == "completed" + assert request(path, body) == (code, result) + assert request(f"/api/chat/sessions/{sid}/turns", original)[1]["turn_id"] == queued + assert store.load_turn(sid, queued)["status"] == "completed" + assert request(f"/api/chat/sessions/{sid}/queue")[1]["turns"] == [] + rows = [json.loads(line) for line in capture.read_text().splitlines()] + assert len([r for r in rows if r["method"] == "turn/start"]) == 1 + steers = [r for r in rows if r["method"] == "turn/steer"] + assert len(steers) == 1 and steers[0]["params"]["input"][0]["text"] == original["message"] + assert [r["text"] for r in store.messages(sid) if r["role"] == "user"] == ["wait for steer", original["message"]] + assert len([e for e in store.events_after(sid, queued, None) if e["kind"] == "turn.completed"]) == 1 + + +def test_unknown_native_delivery_stays_reserved_across_reload_retry_and_queue_claim(ordinary, monkeypatch): # noqa: F811 + store, runtime, contexts, request, capture, *_ = ordinary + sid, active, queued, _ = held_queue(ordinary) + adapter = runtime.adapters[sid] + calls = [] + + def uncertain(*args): + calls.append(args) + raise TimeoutError("native acknowledgement lost") + + with monkeypatch.context() as patch: + patch.setattr(adapter, "steer_turn", uncertain) + path = f"/api/chat/sessions/{sid}/queue/{queued}/steer" + body = {"target_turn_id": active, "client_ingress_id": "lost"} + assert request(path, body)[0] == 409 + assert request(path, body)[0] == 409 + assert request(path, {**body, "client_ingress_id": "new"})[1]["error_code"] == "queue_steering_in_progress" + assert len(calls) == 1 + monkeypatch.setattr("loopx.chat_ingress.SESSION_QUEUE_MAX_PENDING", 1) + assert request(f"/api/chat/sessions/{sid}/turns", { + "message": "No capacity escape", "client_turn_id": "over-capacity", "delivery": "queue"})[0] == 429 + # A fresh store instance sees the same durable pending input and reservation. + from loopx.chat_store import ChatSessionStore + fresh = ChatSessionStore(store.root.parent) + assert fresh.queued_turns(sid) == [] + held = fresh.queued_turns(sid, include_reserved=True) + assert len(held) == 1 and held[0]["_queue_steering"]["operation_id"] == "lost" + runtime.steer_active_turn(session_id=sid, expected_turn_id=active, + client_ingress_id="release-active", message="Complete this check.") + runtime.wait_for_turn(session_id=sid, turn_id=active, timeout_sec=10) + assert fresh.claim_next_queued_turn(sid) is None + assert request(f"/api/chat/sessions/{sid}/queue")[1]["turns"][0]["steering"]["status"] == "pending" + + +def test_proven_non_delivery_preserves_fifo_and_fresh_attempt_can_steer(ordinary): # noqa: F811 + store, runtime, contexts, request, capture, *_ = ordinary + sid, active, queued, _ = held_queue(ordinary) + path = f"/api/chat/sessions/{sid}/queue/{queued}/steer" + body = {"target_turn_id": active, "client_ingress_id": "not-attached"} + adapter = runtime.adapters.pop(sid) + try: + assert request(path, body)[1]["error_code"] == "live_steering_session_not_attached" + assert [t["turn_id"] for t in store.queued_turns(sid)] == [queued] + finally: + runtime.adapters[sid] = adapter + assert request(path, {**body, "client_ingress_id": "recovered"})[0] == 200 + assert request(path, body)[0] == 409 + assert store.load_turn(sid, queued)["_queue_steering"]["status"] == "delivered" + + +def test_queue_steering_revoked_scope_and_wrong_turn_leave_original_unchanged(ordinary): # noqa: F811 + from loopx.capabilities.native_chat.project_context import ChatProjectContexts + store, runtime, contexts, request, capture, *_ = ordinary + sid, active, queued, _ = held_queue(ordinary) + original = store.load_turn(sid, queued) + path = f"/api/chat/sessions/{sid}/queue/{queued}/steer" + assert request(path, {"target_turn_id": "wrong", "client_ingress_id": "wrong-target"})[0] == 409 + runtime.project_contexts = ChatProjectContexts([]) + before = capture.read_bytes() + assert request(path, {"target_turn_id": active, "client_ingress_id": "revoked"})[0] == 400 + assert request(f"/api/chat/sessions/{sid}/queue")[0] == 400 + assert store.load_turn(sid, queued) == original and capture.read_bytes() == before + + +def test_attachment_queue_is_preserved_when_native_steering_is_text_only(ordinary): # noqa: F811 + import base64 + store, runtime, contexts, request, capture, *_ = ordinary + data = b"# Original attachment\n" + attachment = {"id": "original-note", "name": "note.md", "mime_type": "text/markdown", + "size": len(data), "data_url": "data:text/markdown;base64," + base64.b64encode(data).decode()} + sid, active, queued, _ = held_queue(ordinary, attachments=[attachment]) + original = store.load_turn(sid, queued) + code, error = request(f"/api/chat/sessions/{sid}/queue/{queued}/steer", { + "target_turn_id": active, "client_ingress_id": "text-only"}) + assert code == 409 and error["error_code"] == "queue_steering_attachments_unavailable" + assert store.load_turn(sid, queued) == original + assert store.queued_turns(sid)[0]["attachments"] == [attachment] + + +def test_peer_queue_cannot_bypass_its_original_collaboration_execution(ordinary): # noqa: F811 + store, runtime, contexts, request, capture, *_ = ordinary + sid, active, queued, _ = held_queue(ordinary) + peer, _ = store.create_queued_turn(sid, client_turn_id="peer", message="Original peer work", origin="peer") + before = capture.read_bytes() + code, error = request(f"/api/chat/sessions/{sid}/queue/{peer['turn_id']}/steer", { + "target_turn_id": active, "client_ingress_id": "cannot-adopt-peer"}) + assert code == 409 and error["error_code"] == "queue_steering_source_unavailable" + assert store.load_turn(sid, peer["turn_id"]) == peer and capture.read_bytes() == before + + +def test_queue_steering_reservation_fences_concurrent_automatic_claim(ordinary, monkeypatch): # noqa: F811 + from concurrent.futures import ThreadPoolExecutor + from threading import Event + store, runtime, contexts, request, capture, *_ = ordinary + sid, active, queued, _ = held_queue(ordinary) + entered, proceed = Event(), Event() + native = runtime.steer_active_turn + + def pause(**kwargs): + entered.set() + assert proceed.wait(5) + return native(**kwargs) + + monkeypatch.setattr(runtime, "steer_active_turn", pause) + with ThreadPoolExecutor(max_workers=2) as pool: + steering = pool.submit(request, f"/api/chat/sessions/{sid}/queue/{queued}/steer", + {"target_turn_id": active, "client_ingress_id": "concurrent"}) + assert entered.wait(5) + consumer = pool.submit(store.claim_next_queued_turn, sid) + assert not consumer.done() + proceed.set() + assert steering.result(timeout=10)[0] == 200 + assert consumer.result(timeout=10) is None + runtime.wait_for_turn(session_id=sid, turn_id=active, timeout_sec=10) + assert store.load_turn(sid, queued)["status"] == "completed" + starts = [json.loads(line) for line in capture.read_text().splitlines() if json.loads(line)["method"] == "turn/start"] + assert len(starts) == 1 + + +def test_native_ack_survives_queue_settlement_failure_without_resending(ordinary, monkeypatch): # noqa: F811 + store, runtime, contexts, request, capture, *_ = ordinary + sid, active, queued, _ = held_queue(ordinary) + write = store.write_queue_steering + + def fail_commit(sid, turn, plan): + if plan["action"] == "complete": + raise OSError("settlement write unavailable") + return write(sid, turn, plan) + + path = f"/api/chat/sessions/{sid}/queue/{queued}/steer" + body = {"target_turn_id": active, "client_ingress_id": "recover-commit"} + with monkeypatch.context() as patch: + patch.setattr(store, "write_queue_steering", fail_commit) + assert request(path, body)[0] == 503 + runtime.wait_for_turn(session_id=sid, turn_id=active, timeout_sec=10) + assert store.queued_turns(sid) == [] + assert request(path, body)[0] == 200 + rows = [json.loads(line) for line in capture.read_text().splitlines()] + assert len([r for r in rows if r["method"] == "turn/steer"]) == 1 + assert len([r for r in rows if r["method"] == "turn/start"]) == 1 + + def test_browser_queue_preserves_active_turn_then_delivers_fifo_once(ordinary): # noqa: F811 store, runtime, contexts, request, capture, _fake, _workspace = ordinary sid = request("/api/chat/sessions", {"context_kind": "project", From 7d7d9225007499dd6f0cc10135929a255af0f1d4 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 01:14:05 +0800 Subject: [PATCH 02/11] fix(chat): follow the active turn after automatic queue advancement Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- apps/presentation/dashboard/src/data/chat.ts | 3 ++- .../dashboard/src/data/use-conversation-queue.ts | 16 +++++++++------- .../personal-workspace-model.ts | 1 + .../personal-workspace-page.tsx | 13 +++++++++++++ .../dashboard/src/views/dashboard-page.tsx | 16 ++++++++++++++++ .../composer-session-admission.mjs | 12 ++++++++++++ examples/personal-workspace-browser/fixture.mjs | 4 +++- loopx/capabilities/native_chat/queue_api.py | 2 +- tests/presentation/test_chat_turn_queue.py | 1 + 9 files changed, 58 insertions(+), 10 deletions(-) diff --git a/apps/presentation/dashboard/src/data/chat.ts b/apps/presentation/dashboard/src/data/chat.ts index 9901dbbfc0..3752a60a1b 100644 --- a/apps/presentation/dashboard/src/data/chat.ts +++ b/apps/presentation/dashboard/src/data/chat.ts @@ -967,6 +967,7 @@ export async function fetchChatSession(sessionId: string, signal?: AbortSignal) 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(), @@ -989,7 +990,7 @@ export async function fetchChatQueue(sessionId: string, signal?: AbortSignal) { || 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.turns; + return result.data; } export async function steerQueuedChatTurn(sessionId: string, queuedId: string, targetId: string, ingressId: string) { diff --git a/apps/presentation/dashboard/src/data/use-conversation-queue.ts b/apps/presentation/dashboard/src/data/use-conversation-queue.ts index 398d6799d7..76de2df8e1 100644 --- a/apps/presentation/dashboard/src/data/use-conversation-queue.ts +++ b/apps/presentation/dashboard/src/data/use-conversation-queue.ts @@ -7,7 +7,7 @@ export function useConversationQueue({sessionId, scope, active, refreshKey}: { }) { const key = JSON.stringify([scope, sessionId]); const [attempt, setAttempt] = useState(0); - const [state, setState] = useState<{key: string; turns: ChatQueuedTurn[]; error: Error | null} | null>(null); + const [state, setState] = useState<{key: string; turns: ChatQueuedTurn[]; activeTurnId: string | null; error: Error | null} | null>(null); useEffect(() => { if (!sessionId) return; const controller = new AbortController(); @@ -15,18 +15,20 @@ export function useConversationQueue({sessionId, scope, active, refreshKey}: { let failures = 0, pending = false; const read = async () => { try { - const turns = await fetchChatQueue(sessionId, AbortSignal.any([controller.signal, AbortSignal.timeout(5000)])); + const snapshot = await fetchChatQueue(sessionId, AbortSignal.any([controller.signal, AbortSignal.timeout(5000)])); if (controller.signal.aborted) return; - failures = 0; pending = turns.length > 0; - setState({key, turns, error: null}); + 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: [], error: cause instanceof Error ? cause : new Error(String(cause))}); + setState({key, turns: [], activeTurnId: null, error: cause instanceof Error ? cause : new Error(String(cause))}); } - if (!controller.signal.aborted && (active || pending)) { + // 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); } }; @@ -34,6 +36,6 @@ export function useConversationQueue({sessionId, scope, active, refreshKey}: { return () => { controller.abort(); clearTimeout(timer); }; }, [key, sessionId, active, refreshKey, attempt]); const current = sessionId && state?.key === key ? state : null; - return {turns: current?.turns ?? [], error: current?.error ?? null, + return {turns: current?.turns ?? [], activeTurnId: current?.activeTurnId ?? null, error: current?.error ?? null, retry: () => setAttempt(value => value + 1)}; } diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts index 482d2c89b8..3d6bbf718a 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts +++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-model.ts @@ -575,6 +575,7 @@ export type PersonalWorkspaceCallbacks = { onInterruptConversationTurn?: (contextId: string, turnId: string) => Promise; onSteerConversationTurn?: (contextId: string, turnId: string, message: string, ingressId: string) => Promise; onQueueConversationTurn?: (contextId: string, sessionId: string, message: string, ingressId: string, attachments?: WorkspaceImageAttachment[]) => Promise; + onObserveConversationTurn?: (contextId: string, sessionId: string, turnId: string) => Promise; onOpenGoal?: (goalId: string) => void | Promise; onOpenGoalView?: (tab: WorkspaceGoalTab) => void; onOpenRunSession?: (run: WorkspaceRun) => void | Promise; diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx index 6da55ae3ad..937b421765 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx @@ -1400,6 +1400,18 @@ export function PersonalWorkspacePage({ refreshKey: `${managerMessages.length}:${Boolean(managerMessages.at(-1)?.pending)}:${steering}`, }); const queuedTurnKeys = new Set(conversationQueue.turns.map(turn => JSON.stringify([conversationSessionId, turn.turn_id]))); + const observedQueueTurn = useRef(""); + useEffect(() => { + const turnId = conversationQueue.activeTurnId; + if (!conversationSessionId || !turnId || turnId === runningMessage?.sourceTurnId || !callbacks.onObserveConversationTurn) return; + const key = JSON.stringify([statusSourceControl?.activeSource.statusUrl, conversationKey, conversationSessionId, turnId]); + if (observedQueueTurn.current === key) return; + observedQueueTurn.current = key; + void callbacks.onObserveConversationTurn(conversationKey, conversationSessionId, turnId).catch(() => { + if (observedQueueTurn.current === key) observedQueueTurn.current = ""; + }); + }, [conversationQueue.activeTurnId, conversationQueue.turns, conversationSessionId, runningMessage?.sourceTurnId, + callbacks.onObserveConversationTurn, conversationKey, statusSourceControl?.activeSource.statusUrl]); const legacyDrafts = conversationSessionId ? legacySteeringRequests(conversationSessionId, composerDraftKey) : []; function restoreLegacyDraft(key: string) { if (composer.trim() || !conversationSessionId) return; @@ -3280,6 +3292,7 @@ export function PersonalWorkspacePage({ onDefaults={selectedGoal && !originalConversation ? () => openSettings({kind: "settings", tab: "conversation"}, true) : undefined} /> : null} {queueTurnId && callbacks.onInterruptConversationTurn ? : null} diff --git a/apps/presentation/dashboard/src/views/dashboard-page.tsx b/apps/presentation/dashboard/src/views/dashboard-page.tsx index 7fd0c3a260..77a108a16b 100644 --- a/apps/presentation/dashboard/src/views/dashboard-page.tsx +++ b/apps/presentation/dashboard/src/views/dashboard-page.tsx @@ -2364,6 +2364,8 @@ function PersonalGoalHome({ let submittedTurnId: string | undefined; let submittedSessionId: string | undefined; let streamController: AbortController | undefined; + const previousStreamController = streamControllers.current.get(targetContextId); + let submittedAsQueue = Boolean(route?.queue); let queueAccepted = false; let streamEnded = false; let handedOff = false; @@ -2431,14 +2433,24 @@ function PersonalGoalHome({ sourceTurnId: turnId, sourceSessionId: sessionId, ...(phase === "turn.accepted" ? { activity: [locale === "zh-CN" ? "请求已接收，等待执行器输出" : "Request accepted · waiting for executor output"] } : {}), }); - if (route?.queue) { - if (phase === "turn.queued") { - updateConversationMessage(targetContextId, streamingMessageId, { - pending: false, preparing: false, endedAt: undefined, - text: "", queued: true, activity: [], - }); - return; + if (phase === "turn.queued") { + submittedAsQueue = true; + updateConversationMessage(targetContextId, streamingMessageId, { + pending: false, preparing: false, endedAt: undefined, + text: "", queued: true, activity: [], + }); + // Native hosts can queue ordinary sends; their receipt must not + // replace the observed running Turn's stream controls. + if (!route?.queue && streamControllers.current.get(targetContextId) === streamController) { + if (previousStreamController && !previousStreamController.signal.aborted) { + streamControllers.current.set(targetContextId, previousStreamController); + } else { + streamControllers.current.delete(targetContextId); + } } + return; + } + if (submittedAsQueue) { // Queued is a durable receipt, not observed execution. Only the // original worker's start/stream may take the active controls. if (phase === "turn.started") { diff --git a/examples/personal-workspace-browser/attached-host-follow-up.mjs b/examples/personal-workspace-browser/attached-host-follow-up.mjs index a9c890fb7f..2863422c60 100644 --- a/examples/personal-workspace-browser/attached-host-follow-up.mjs +++ b/examples/personal-workspace-browser/attached-host-follow-up.mjs @@ -1,4 +1,7 @@ +import { resolve } from "node:path"; + import { runCodexAttached } from "./codex-attached.mjs"; +import { outputDir } from "./fixture.mjs"; import { openWorkspacePage } from "./scenario-context.mjs"; // An attached_host Session queues a follow-up while its host runs a Turn @@ -39,8 +42,8 @@ export const attachedHostFollowUpScenario = { if (route.request().method() !== "POST") return route.fallback(); const sessionId = new URL(route.request().url()).pathname.split("/")[4]; const session = runtime.sessions.get(sessionId); - const { message } = route.request().postDataJSON(); - posts.push({ message, sessionId }); + const { message, delivery } = route.request().postDataJSON(); + posts.push({ message, sessionId, delivery }); if (session?.session_mode !== "attached_host") { // The managed_runtime answer the composer exists to avoid. await route.fulfill({ contentType: "application/json", status: 409, @@ -51,6 +54,10 @@ export const attachedHostFollowUpScenario = { // Turn stays the Session's active Turn. const turnId = `turn-queued-${posts.length}`; runtime.messages.get(sessionId)?.push({ message_id: `${turnId}-user`, turn_id: turnId, role: "user", text: message, created_at: "2026-08-13T01:00:02Z" }); + runtime.queuedTurns.set(sessionId, [...(runtime.queuedTurns.get(sessionId) ?? []), { + turn_id: turnId, created_at: "2026-08-13T01:00:02Z", origin: "web", + preview: message, character_count: message.length, attachment_count: 0, steering: null, + }]); await route.fulfill({ contentType: "application/json", status: 202, json: { ok: true, schema_version: "loopx_chat_turn_accepted_v1", session_id: sessionId, turn_id: turnId, created: true, status: "queued", events_url: `/api/chat/sessions/${sessionId}/turns/${turnId}/events` } }); }); @@ -59,7 +66,6 @@ export const attachedHostFollowUpScenario = { const { page } = context; const composer = page.getByLabel("向 LoopX 发送消息"); const sendButton = page.getByRole("button", { name: "发送", exact: true }); - const turnRunningHint = page.locator(".personal-composer-status", { hasText: "本轮回答进行中" }); const waitForStream = async (turnId) => { const streamed = () => heldStreams.some((route) => route.request().url().includes(`/turns/${turnId}/events`)); for (let attempt = 0; attempt < 100 && !streamed(); attempt += 1) await page.waitForTimeout(50); @@ -71,10 +77,10 @@ export const attachedHostFollowUpScenario = { // The recovery reads the Session and resumes its running Turn. if (!await waitForStream(runningTurnId(goal.id))) throw new Error(`${goal.label} did not resume its running Turn`); await page.getByRole("button", { name: "中断本轮", exact: true }).waitFor({ state: "visible", timeout: 5_000 }); + await page.locator(`.personal-composer-stop[data-turn-id="${runningTurnId(goal.id)}"]`).waitFor({ state: "visible", timeout: 5_000 }); }; try { await openGoalChat(managedGoal); - await page.locator(".personal-composer-status", { hasText: "新消息将在下一轮处理" }).waitFor(); await composer.fill("当前工作先检查依赖"); if (await sendButton.isDisabled()) throw new Error("Managed Codex instructions were blocked"); // This scenario isolates attached queues; exact native steering and retry @@ -85,19 +91,58 @@ export const attachedHostFollowUpScenario = { await openGoalChat(attachedGoal); const followUp = "附着宿主回合进行中的后续消息"; await composer.fill(followUp); - if (await turnRunningHint.count()) throw new Error("An attached_host composer showed the one-Turn wait hint"); if (await sendButton.isDisabled()) throw new Error("An attached_host composer blocked Send while its host ran a Turn"); await sendButton.click(); for (let attempt = 0; attempt < 100 && !posts.length; attempt += 1) await page.waitForTimeout(50); if (posts.length !== 1 || posts[0].sessionId !== sessionIdFor(attachedGoal.id) || posts[0].message !== followUp) { throw new Error(`The attached_host follow-up did not reach its Session queue: ${JSON.stringify(posts)}`); } - await page.locator(".personal-channel-timeline .personal-message").filter({ hasText: followUp }).first().waitFor({ timeout: 5_000 }); - if (!await waitForStream("turn-queued-1")) throw new Error("The page did not follow the queued follow-up Turn it was given"); + await page.locator(".personal-conversation-queue li").filter({ hasText: followUp }).waitFor({ timeout: 10_000 }); + await page.locator(`.personal-composer-stop[data-turn-id="${runningTurnId(attachedGoal.id)}"]`).waitFor({ timeout: 10_000 }); + if (page.__loopxRuntime.sessions.get(sessionIdFor(attachedGoal.id)).active_turn_id !== runningTurnId(attachedGoal.id)) { + throw new Error("Queuing the follow-up replaced the attached host's active Turn"); + } if (await composer.inputValue()) throw new Error("The queued follow-up stayed in the composer as if it had been rejected"); if (await page.getByRole("alert").filter({ hasText: /发送失败|already running/u }).count()) { throw new Error("The queued follow-up was reported as a failed send"); } + await page.screenshot({ path: resolve(outputDir, "attached-host-pending-queue.png"), fullPage: false }); + await page.setViewportSize({ width: 390, height: 844 }); + const stop = page.locator(`.personal-composer-stop[data-turn-id="${runningTurnId(attachedGoal.id)}"]`); + const box = await stop.boundingBox(); + if (!box || box.y < 0 || box.y + box.height > 844) throw new Error("The attached host Stop control left the narrow viewport"); + await page.screenshot({ path: resolve(outputDir, "attached-host-pending-queue-390.png"), fullPage: false }); + const interruptsBefore = context.api.interrupts.length; + await stop.click(); + for (let attempt = 0; attempt < 100 && context.api.interrupts.length === interruptsBefore; attempt += 1) await page.waitForTimeout(50); + const interrupted = context.api.interrupts.slice(interruptsBefore); + if (interrupted.length !== 1 || interrupted[0].sessionId !== sessionIdFor(attachedGoal.id) + || interrupted[0].turnId !== runningTurnId(attachedGoal.id)) { + throw new Error(`Stop targeted the queued follow-up instead of the running attached host: ${JSON.stringify(interrupted)}`); + } + await stop.waitFor({ state: "hidden" }); + // The page still sees idle when another page's host starts a Turn. + // Native admission queues the ordinary send; that receipt is not execution. + const racedTurnId = "turn-host-started-elsewhere"; + const attachedSessionId = sessionIdFor(attachedGoal.id); + let staleQueue = true; + await page.route(`**/api/chat/sessions/${attachedSessionId}/queue`, route => staleQueue + ? route.fulfill({ json: { ok: true, schema_version: "loopx_chat_store_v1", session_id: attachedSessionId, + active_turn_id: null, turns: page.__loopxRuntime.queuedTurns.get(attachedSessionId) ?? [] } }) + : route.fallback()); + page.__loopxRuntime.sessions.set(attachedSessionId, { ...page.__loopxRuntime.sessions.get(attachedSessionId), active_turn_id: racedTurnId, status: "busy" }); + const racedFollowUp = "另一页面启动宿主时仍然只排队"; + await composer.fill(racedFollowUp); + await sendButton.click(); + await page.locator(".personal-conversation-queue li").filter({ hasText: racedFollowUp }).waitFor({ timeout: 10_000 }); + if (posts.length !== 2 || posts[1].delivery !== undefined) throw new Error("The stale page did not exercise ordinary-send native queue admission"); + if (await page.getByText("请求已接收，等待执行器输出", { exact: true }).count()) throw new Error("A queued receipt appeared as an executing assistant reply"); + staleQueue = false; + const racedStop = page.locator(`.personal-composer-stop[data-turn-id="${racedTurnId}"]`); + await racedStop.waitFor({ timeout: 15_000 }); + await racedStop.click(); + for (let attempt = 0; attempt < 100 && context.api.interrupts.length !== interruptsBefore + 2; attempt += 1) await page.waitForTimeout(50); + if (context.api.interrupts.at(-1)?.turnId !== racedTurnId) throw new Error("A native-queued admission took the other page's running Turn controls"); } finally { for (const route of heldStreams.splice(0)) await route.abort().catch(() => {}); await context.close(); @@ -105,7 +150,7 @@ export const attachedHostFollowUpScenario = { return { coverageEntries: context.coverageEntries, note: "managed-runtime-running-turn: native instructions stay available without creating another Turn. " - + "attached-host-follow-up: while the host runs a Turn, Send stays open, one POST reaches the Session queue and the page follows the queued Turn.", + + "attached-host-follow-up: Send stays open, one POST reaches the persistent Session queue and current-Turn controls stay on the running host Turn.", }; }, }; diff --git a/examples/personal-workspace-browser/chat-recovery.mjs b/examples/personal-workspace-browser/chat-recovery.mjs index 171e79b838..d0f721bbb0 100644 --- a/examples/personal-workspace-browser/chat-recovery.mjs +++ b/examples/personal-workspace-browser/chat-recovery.mjs @@ -390,13 +390,14 @@ export const chatRecoveryScenario = { await page.getByRole("button", { name: "发送", exact: true }).click(); while (api.turnRequests.length === turnsBeforeRunningCheck) await page.waitForTimeout(50); const unsupportedSessionId = api.turnRequests.at(-1).sessionId; + const recoveredTurnId = api.turnRequests.at(-1).turnId; page.__loopxRuntime.sessions.set(unsupportedSessionId, { ...page.__loopxRuntime.sessions.get(unsupportedSessionId), adapter_kind: "external" }); await page.reload({ waitUntil: "domcontentloaded" }); await page.getByTestId("personal-goal-home").waitFor({ state: "visible" }); await page.locator(".personal-goal-link").first().click(); await page.getByRole("navigation", { name: "Goal 视图" }).getByRole("button", { name: /^(Chat|对话)$/ }).click(); - const turnRunningHint = page.locator(".personal-composer-status", { hasText: "本轮进行中" }); - await turnRunningHint.waitFor({ state: "visible", timeout: 5_000 }); + const currentTurnControl = page.locator(`.personal-composer-stop[data-turn-id="${recoveredTurnId}"]`); + await currentTurnControl.waitFor({ state: "visible", timeout: 5_000 }); await page.getByLabel("向 LoopX 发送消息").fill("回合进行中不应发送"); if (await page.getByRole("button", { name: "发送", exact: true }).isDisabled()) { throw new Error("Recovered work did not offer explicit next-Turn queuing"); @@ -407,12 +408,12 @@ export const chatRecoveryScenario = { await page.locator(".personal-goal-link").nth(1).click(); await page.getByRole("navigation", { name: "Goal 视图" }).getByRole("button", { name: /^(Chat|对话)$/ }).click(); await page.getByLabel("向 LoopX 发送消息").waitFor({ state: "visible" }); - if (await turnRunningHint.count()) throw new Error("Another Goal's composer waited for this Goal's running Turn"); + if (await currentTurnControl.count()) throw new Error("Another Goal's composer controlled this Goal's running Turn"); await page.locator(".personal-goal-link").first().click(); await page.getByRole("navigation", { name: "Goal 视图" }).getByRole("button", { name: /^(Chat|对话)$/ }).click(); - await turnRunningHint.waitFor({ state: "visible", timeout: 5_000 }); + await currentTurnControl.waitFor({ state: "visible", timeout: 5_000 }); await page.getByLabel("向 LoopX 发送消息").fill("回合进行中不应发送"); - await turnRunningHint.waitFor({ state: "hidden", timeout: 10_000 }); + await currentTurnControl.waitFor({ state: "hidden", timeout: 10_000 }); if (await page.getByRole("button", { name: "发送", exact: true }).isDisabled()) { throw new Error("Composer stayed blocked after the running Turn completed"); } @@ -425,6 +426,7 @@ export const chatRecoveryScenario = { // controls and keep the draft; the owner can explicitly queue it next. const busySessionId = api.turnRequests.at(-1).sessionId; const foreignTurnId = `turn-foreign-${Date.now()}`; + const foreignTurnControl = page.locator(`.personal-composer-stop[data-turn-id="${foreignTurnId}"]`); page.__loopxRuntime.turnMessages.set(foreignTurnId, "另一页面发起的中断控制回合"); page.__loopxRuntime.sessions.set(busySessionId, { ...page.__loopxRuntime.sessions.get(busySessionId), active_turn_id: foreignTurnId, status: "busy" }); let rejectedPosts = 0; @@ -465,7 +467,7 @@ export const chatRecoveryScenario = { holdReads = false; for (const route of heldReads.splice(0)) await route.fallback(); await page.unroute("**/api/chat/sessions?*"); - await turnRunningHint.waitFor({ state: "visible", timeout: 5_000 }); + await foreignTurnControl.waitFor({ state: "visible", timeout: 5_000 }); await page.getByRole("button", { name: "中断本轮" }).waitFor({ state: "visible" }); if (await page.getByRole("button", { name: "中断本轮" }).count() !== 1) { throw new Error("The recovery added a second pending reply instead of adopting the handoff reply"); @@ -476,7 +478,7 @@ export const chatRecoveryScenario = { if (await page.locator(".personal-channel-timeline .personal-message").filter({ hasText: draft }).count()) { throw new Error("A message the service did not accept stayed in the conversation"); } - await turnRunningHint.waitFor({ state: "hidden", timeout: 10_000 }); + await foreignTurnControl.waitFor({ state: "hidden", timeout: 10_000 }); if (await sendButton.isDisabled()) throw new Error("Send stayed blocked after the adopted Turn completed"); if (rejectedPosts !== 1) throw new Error(`The composer posted ${rejectedPosts} times into a running Turn`); await page.unroute(`**/api/chat/sessions/${busySessionId}/turns`); @@ -486,6 +488,7 @@ export const chatRecoveryScenario = { // Turn, so the handoff must retain it and read again rather than // treat the failure as the Turn having ended. const retriedTurnId = `turn-foreign-retry-${Date.now()}`; + const retriedTurnControl = page.locator(`.personal-composer-stop[data-turn-id="${retriedTurnId}"]`); page.__loopxRuntime.turnMessages.set(retriedTurnId, "读取失败后仍在运行的中断控制回合"); page.__loopxRuntime.sessions.set(busySessionId, { ...page.__loopxRuntime.sessions.get(busySessionId), active_turn_id: retriedTurnId, status: "busy" }); let retryRejectedPosts = 0; @@ -517,7 +520,7 @@ export const chatRecoveryScenario = { if (!failedReads) throw new Error("The 409 did not make the page re-read the Session"); if (!heldRetryReads.length) throw new Error("A failed Session read was not retried while the reported Turn could still run"); if (await sendButton.isDisabled()) throw new Error("The retried Session read hid explicit next-Turn queuing"); - await turnRunningHint.waitFor({ state: "visible", timeout: 5_000 }); + await retriedTurnControl.waitFor({ state: "visible", timeout: 5_000 }); await page.getByRole("button", { name: "中断本轮" }).waitFor({ state: "visible" }); if (retryRejectedPosts !== 1) throw new Error(`The composer posted ${retryRejectedPosts} times while the Session read was failing`); failReads = false; @@ -527,7 +530,7 @@ export const chatRecoveryScenario = { throw new Error("The retried recovery added a second pending reply instead of adopting the handoff reply"); } if (await composerInput.inputValue() !== draft) throw new Error("The draft rejected by a running Turn was not kept across the failed read"); - await turnRunningHint.waitFor({ state: "hidden", timeout: 10_000 }); + await retriedTurnControl.waitFor({ state: "hidden", timeout: 10_000 }); if (await sendButton.isDisabled()) throw new Error("Send stayed blocked after the retried recovery saw the Turn complete"); if (retryRejectedPosts !== 1) throw new Error(`The composer posted ${retryRejectedPosts} times into a running Turn`); await page.unroute(`**/api/chat/sessions/${busySessionId}/turns`); @@ -541,6 +544,7 @@ export const chatRecoveryScenario = { // The pending reply shows Adjust/Interrupt from the 409 on, so before any // Session read returns, both must act on the exact reported Turn. const controlledTurnId = `turn-foreign-controls-${Date.now()}`; + const controlledTurnControl = page.locator(`.personal-composer-stop[data-turn-id="${controlledTurnId}"]`); page.__loopxRuntime.turnMessages.set(controlledTurnId, "交接期间可调整和中断的中断控制回合"); page.__loopxRuntime.sessions.set(busySessionId, { ...page.__loopxRuntime.sessions.get(busySessionId), active_turn_id: controlledTurnId, status: "busy" }); let controlRejectedPosts = 0; @@ -567,6 +571,7 @@ export const chatRecoveryScenario = { await sendButton.click(); for (let attempt = 0; attempt < 100 && (!controlRejectedPosts || !heldControlReads.length); attempt += 1) await page.waitForTimeout(50); if (!heldControlReads.length) throw new Error("The 409 did not make the page re-read the Session"); + await controlledTurnControl.waitFor({ state: "visible", timeout: 5_000 }); const handoffReply = page.locator(".personal-message").filter({ has: page.getByRole("button", { name: "中断本轮", exact: true }) }); await handoffReply.getByRole("button", { name: "调整本轮", exact: true }).click(); await composerInput.fill("交接期间先核对依赖。"); @@ -583,7 +588,7 @@ export const chatRecoveryScenario = { throw new Error(`Interrupt during the handoff did not reach the reported Turn: ${JSON.stringify(interrupted)}`); } await page.locator(".personal-message").filter({ hasText: "已中断。你可以在当前会话继续发送消息。" }).last().waitFor({ timeout: 5_000 }); - await turnRunningHint.waitFor({ state: "hidden", timeout: 5_000 }); + await controlledTurnControl.waitFor({ state: "hidden", timeout: 5_000 }); if (await composerInput.inputValue() !== "") throw new Error("The accepted shared-composer adjustment did not clear"); await composerInput.fill("中断后的后续消息草稿"); if (await sendButton.isDisabled()) throw new Error("A new follow-up draft stayed blocked after interruption"); From f225c9a732e46d66cb4598d62aa3a0d9e3e073c3 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 02:34:17 +0800 Subject: [PATCH 06/11] test(chat): avoid shadowing the native queue fixture Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/presentation/test_chat_turn_queue.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/tests/presentation/test_chat_turn_queue.py b/tests/presentation/test_chat_turn_queue.py index 63ec747e8c..3512639fef 100644 --- a/tests/presentation/test_chat_turn_queue.py +++ b/tests/presentation/test_chat_turn_queue.py @@ -7,8 +7,8 @@ from test_chat_ordinary_project import ordinary # noqa: F401 -def held_queue(ordinary, text="Check the second source.", attachments=None): - store, runtime, contexts, request, capture, *_ = ordinary +def held_queue(project_fixture, text="Check the second source.", attachments=None): + store, runtime, contexts, request, capture, *_ = project_fixture sid = request("/api/chat/sessions", {"context_kind": "project", "project_ref": contexts.available()[0]["project_ref"]})[1]["session_id"] first = request(f"/api/chat/sessions/{sid}/turns", { From f73f0cf7a017e526155659000720256ea696c075 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 02:47:43 +0800 Subject: [PATCH 07/11] fix(chat): retain drafts when attached steering is unsupported Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- .../features/personal-workspace/personal-workspace-page.tsx | 2 +- .../personal-workspace-browser/attached-host-follow-up.mjs | 3 +++ 2 files changed, 4 insertions(+), 1 deletion(-) diff --git a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx index 8a466c3278..d1460739f5 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx +++ b/apps/presentation/dashboard/src/features/personal-workspace/personal-workspace-page.tsx @@ -2542,7 +2542,7 @@ export function PersonalWorkspacePage({ } finally { setSteering(false); } return; } - if (intent === "steer" && conversationTurnRunning) { + if (intent === "steer" && (conversationTurnRunning || queueTurnId)) { setActionFeedback(locale === "zh-CN" ? "此执行器不支持本轮纠偏；可按 Enter 排队，草稿已保留。" : "This executor cannot steer this turn. Press Enter to queue; draft retained."); return; } diff --git a/examples/personal-workspace-browser/attached-host-follow-up.mjs b/examples/personal-workspace-browser/attached-host-follow-up.mjs index 2863422c60..139087c18b 100644 --- a/examples/personal-workspace-browser/attached-host-follow-up.mjs +++ b/examples/personal-workspace-browser/attached-host-follow-up.mjs @@ -91,6 +91,9 @@ export const attachedHostFollowUpScenario = { await openGoalChat(attachedGoal); const followUp = "附着宿主回合进行中的后续消息"; await composer.fill(followUp); + await composer.press("Meta+Enter"); + await page.getByRole("status").filter({ hasText: "此执行器不支持本轮纠偏" }).waitFor(); + if (posts.length || await composer.inputValue() !== followUp) throw new Error("Unsupported host steering queued or lost the draft"); if (await sendButton.isDisabled()) throw new Error("An attached_host composer blocked Send while its host ran a Turn"); await sendButton.click(); for (let attempt = 0; attempt < 100 && !posts.length; attempt += 1) await page.waitForTimeout(50); From 6520bd9c396e3ff97de901c1ac9dfda05c76ff22 Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 02:47:43 +0800 Subject: [PATCH 08/11] test: follow the current CLI policy capture contract Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/test_loopx_turn_driver.py | 7 ++++--- 1 file changed, 4 insertions(+), 3 deletions(-) diff --git a/tests/test_loopx_turn_driver.py b/tests/test_loopx_turn_driver.py index 73be78aeee..3afb07ea43 100644 --- a/tests/test_loopx_turn_driver.py +++ b/tests/test_loopx_turn_driver.py @@ -14,6 +14,7 @@ import loopx.cli_commands.turn_run_once as turn_run_once_command from loopx.cli_commands.turn_rendering import render_loopx_turn_execution_markdown +from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload from tests.control_plane.canonical_authority_fixture import ( initialize_canonical_authority, isolate_sqlite_runtime, @@ -1570,7 +1571,7 @@ def test_quota_cli_projects_outer_controller_without_codex_app_action( ] ) - payload = json.loads(output.getvalue()) + payload = read_cli_policy_payload(output.getvalue(), returncode=exit_code) hint = payload["scheduler_hint"] assert exit_code == 0, payload assert hint["execution_context"]["codex_app_applicability"] == "not_applicable" @@ -1604,7 +1605,7 @@ def test_quota_cli_without_scheduler_context_fails_closed(tmp_path: Path) -> Non ] ) - payload = json.loads(output.getvalue()) + payload = read_cli_policy_payload(output.getvalue(), returncode=exit_code) hint = payload["scheduler_hint"] assert exit_code == 0, payload assert hint["action"] == "repair_scheduler_execution_context" @@ -1687,7 +1688,7 @@ def test_quota_cli_short_context_flags_preserve_all_typed_fields( ] ) - payload = json.loads(output.getvalue()) + payload = read_cli_policy_payload(output.getvalue(), returncode=exit_code) context = payload["scheduler_hint"]["execution_context"] assert exit_code == 0, payload assert context["host_surface"] == "generic_cli" From a86f2ea1558a587147b41c059b7cc3ca22c989cc Mon Sep 17 00:00:00 2001 From: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> Date: Thu, 5 Nov 2026 03:01:47 +0800 Subject: [PATCH 09/11] test: read Ark quota continuity from its current capture Signed-off-by: LoopX Agent <337587101+loopx-agent@users.noreply.github.com> --- tests/test_ark_managed_agent_host.py | 4 +++- 1 file changed, 3 insertions(+), 1 deletion(-) diff --git a/tests/test_ark_managed_agent_host.py b/tests/test_ark_managed_agent_host.py index 605b5851cc..f747bf2dad 100644 --- a/tests/test_ark_managed_agent_host.py +++ b/tests/test_ark_managed_agent_host.py @@ -37,6 +37,7 @@ from tests.control_plane.canonical_authority_fixture import ( initialize_canonical_authority, ) +from tests.control_plane.codex_checkpoint_fixture import read_cli_policy_payload REPO_ROOT = Path(__file__).resolve().parents[1] @@ -458,7 +459,7 @@ def _fresh_host_quota_read( runtime: Path, registry: Path, ) -> dict: - return run_json_cli( + payload = run_json_cli( "quota", "should-run", "--goal-id", @@ -470,6 +471,7 @@ def _fresh_host_quota_read( registry_path=registry, runtime_root=runtime, ) + return read_cli_policy_payload(json.dumps(payload), returncode=payload["_returncode"]) def test_fresh_host_reconstructs_frontier_from_durable_loopx_state( From aa33efe949f808c03a12b544bc225ff9e2fdc370 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 5 Nov 2026 11:27:13 +0800 Subject: [PATCH 10/11] fix(chat): use body typography for expanded queued messages Signed-off-by: huangruiteng --- .../personal-workspace/conversation-queue.css | 2 +- .../attached-host-follow-up.mjs | 23 ++++++++++++++++++- 2 files changed, 23 insertions(+), 2 deletions(-) diff --git a/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css b/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css index d097d6bcda..5ec1bbeb40 100644 --- a/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css +++ b/apps/presentation/dashboard/src/features/personal-workspace/conversation-queue.css @@ -12,7 +12,7 @@ .personal-conversation-queue button:disabled { opacity: .45; cursor: default; } .personal-conversation-queue button:focus-visible, .personal-queue-row summary:focus-visible { outline: 2px solid var(--pw-blue); outline-offset: 2px; } .personal-queue-steer { flex-shrink: 0; } -.personal-queue-message { margin: 8px 0 0; } +.personal-queue-message { margin: 8px 0 0; font-size: var(--text-body); line-height: var(--leading-body); } .personal-queue-message p { white-space: pre-wrap; overflow-wrap: anywhere; } .personal-queue-message img { display: block; max-width: 100%; max-height: 120px; object-fit: contain; } .personal-queue-feedback { margin: 0; padding: 0 10px 8px; color: var(--pw-muted); overflow-wrap: anywhere; } diff --git a/examples/personal-workspace-browser/attached-host-follow-up.mjs b/examples/personal-workspace-browser/attached-host-follow-up.mjs index 139087c18b..ad47b123e7 100644 --- a/examples/personal-workspace-browser/attached-host-follow-up.mjs +++ b/examples/personal-workspace-browser/attached-host-follow-up.mjs @@ -100,7 +100,27 @@ export const attachedHostFollowUpScenario = { if (posts.length !== 1 || posts[0].sessionId !== sessionIdFor(attachedGoal.id) || posts[0].message !== followUp) { throw new Error(`The attached_host follow-up did not reach its Session queue: ${JSON.stringify(posts)}`); } - await page.locator(".personal-conversation-queue li").filter({ hasText: followUp }).waitFor({ timeout: 10_000 }); + const queuedFollowUp = page.locator(".personal-conversation-queue li").filter({ hasText: followUp }); + await queuedFollowUp.waitFor({ timeout: 10_000 }); + await queuedFollowUp.locator("summary").click(); + const expandedMessage = queuedFollowUp.locator(".personal-queue-message > p"); + await expandedMessage.waitFor(); + const assertExpandedBodyTypography = async (viewport) => { + const typography = await expandedMessage.evaluate(element => { + const style = getComputedStyle(element); + const root = getComputedStyle(document.documentElement); + return { + fontSize: parseFloat(style.fontSize), + lineHeight: parseFloat(style.lineHeight), + bodySize: parseFloat(root.getPropertyValue("--text-body")), + bodyLineHeight: parseFloat(root.getPropertyValue("--leading-body")), + }; + }); + if (typography.fontSize !== typography.bodySize || typography.lineHeight !== typography.bodyLineHeight) { + throw new Error(`Expanded queued message lost body typography at ${viewport}: ${JSON.stringify(typography)}`); + } + }; + await assertExpandedBodyTypography("1440x900"); await page.locator(`.personal-composer-stop[data-turn-id="${runningTurnId(attachedGoal.id)}"]`).waitFor({ timeout: 10_000 }); if (page.__loopxRuntime.sessions.get(sessionIdFor(attachedGoal.id)).active_turn_id !== runningTurnId(attachedGoal.id)) { throw new Error("Queuing the follow-up replaced the attached host's active Turn"); @@ -111,6 +131,7 @@ export const attachedHostFollowUpScenario = { } await page.screenshot({ path: resolve(outputDir, "attached-host-pending-queue.png"), fullPage: false }); await page.setViewportSize({ width: 390, height: 844 }); + await assertExpandedBodyTypography("390x844"); const stop = page.locator(`.personal-composer-stop[data-turn-id="${runningTurnId(attachedGoal.id)}"]`); const box = await stop.boundingBox(); if (!box || box.y < 0 || box.y + box.height > 844) throw new Error("The attached host Stop control left the narrow viewport"); From d22d1e636009529a30215c86af017a3d8a80a4d5 Mon Sep 17 00:00:00 2001 From: huangruiteng Date: Thu, 5 Nov 2026 11:34:45 +0800 Subject: [PATCH 11/11] test(chat): stabilize queued delivery qualification Signed-off-by: huangruiteng --- .../attached-host-follow-up.mjs | 9 +++++++-- tests/presentation/test_chat_turn_queue.py | 12 ++++++++++++ 2 files changed, 19 insertions(+), 2 deletions(-) diff --git a/examples/personal-workspace-browser/attached-host-follow-up.mjs b/examples/personal-workspace-browser/attached-host-follow-up.mjs index ad47b123e7..99c5ac80f4 100644 --- a/examples/personal-workspace-browser/attached-host-follow-up.mjs +++ b/examples/personal-workspace-browser/attached-host-follow-up.mjs @@ -109,11 +109,16 @@ export const attachedHostFollowUpScenario = { const typography = await expandedMessage.evaluate(element => { const style = getComputedStyle(element); const root = getComputedStyle(document.documentElement); + const rootFontSize = parseFloat(root.fontSize); + const tokenPixels = value => { + const token = value.trim(), size = parseFloat(token); + return token.endsWith("rem") ? size * rootFontSize : size; + }; return { fontSize: parseFloat(style.fontSize), lineHeight: parseFloat(style.lineHeight), - bodySize: parseFloat(root.getPropertyValue("--text-body")), - bodyLineHeight: parseFloat(root.getPropertyValue("--leading-body")), + bodySize: tokenPixels(root.getPropertyValue("--text-body")), + bodyLineHeight: tokenPixels(root.getPropertyValue("--leading-body")), }; }); if (typography.fontSize !== typography.bodySize || typography.lineHeight !== typography.bodyLineHeight) { diff --git a/tests/presentation/test_chat_turn_queue.py b/tests/presentation/test_chat_turn_queue.py index 3512639fef..76d324d158 100644 --- a/tests/presentation/test_chat_turn_queue.py +++ b/tests/presentation/test_chat_turn_queue.py @@ -1,18 +1,30 @@ """Ordinary browser queue intent through real HTTP, store and native stdio.""" import json +import time import pytest from test_chat_ordinary_project import ordinary # noqa: F401 +def wait_for_active_turn(store, session_id, turn_id): + deadline = time.monotonic() + 5 + while time.monotonic() < deadline: + turn = store.load_turn(session_id, turn_id) + if turn and turn.get("status") == "running" and turn.get("upstream_turn_id"): + return turn + time.sleep(0.02) + raise AssertionError(f"Turn did not reach its native active state: {turn}") + + def held_queue(project_fixture, text="Check the second source.", attachments=None): store, runtime, contexts, request, capture, *_ = project_fixture sid = request("/api/chat/sessions", {"context_kind": "project", "project_ref": contexts.available()[0]["project_ref"]})[1]["session_id"] first = request(f"/api/chat/sessions/{sid}/turns", { "message": "wait for steer", "client_turn_id": "held"})[1] + wait_for_active_turn(store, sid, first["turn_id"]) body = {"message": text, "client_turn_id": "pending", "delivery": "queue"} if attachments: body["attachments"] = attachments