diff --git a/dashboard/src/api/modules/octopThreads.ts b/dashboard/src/api/modules/octopThreads.ts index 98eacf31..76124b66 100644 --- a/dashboard/src/api/modules/octopThreads.ts +++ b/dashboard/src/api/modules/octopThreads.ts @@ -1,4 +1,5 @@ import type { HitlPendingPayload } from "../types/hitl"; +import type { UserQuestionPendingPayload } from "../types/userQuestions"; import { request } from "../request"; export interface OctopThread { @@ -37,6 +38,8 @@ export interface OctopThreadHistory { turn_active?: boolean; /** Pending tool approval for this thread (survives page reload). */ hitl_pending?: HitlPendingPayload | null; + /** Pending structured question for this thread (durable across server restart). */ + question_pending?: UserQuestionPendingPayload | null; } export interface OctopThreadPatch { diff --git a/dashboard/src/api/types/index.ts b/dashboard/src/api/types/index.ts index a6936cbf..5736a4bf 100644 --- a/dashboard/src/api/types/index.ts +++ b/dashboard/src/api/types/index.ts @@ -1,4 +1,5 @@ export * from "./hitl"; +export * from "./userQuestions"; export * from "./agent"; export * from "./channel"; export * from "./chat"; diff --git a/dashboard/src/api/types/userQuestions.ts b/dashboard/src/api/types/userQuestions.ts new file mode 100644 index 00000000..da3683e6 --- /dev/null +++ b/dashboard/src/api/types/userQuestions.ts @@ -0,0 +1,23 @@ +export interface UserQuestionOption { + label: string; + description?: string; +} + +export interface UserQuestion { + id: string; + question: string; + header?: string; + options: UserQuestionOption[]; + multi_select?: boolean; +} + +export interface UserQuestionAnswer { + id: string; + selected: string[]; + custom?: string; +} + +export interface UserQuestionPendingPayload { + pending_id: string; + questions: UserQuestion[]; +} diff --git a/dashboard/src/locales/en.json b/dashboard/src/locales/en.json index be84f34d..4c4951d7 100644 --- a/dashboard/src/locales/en.json +++ b/dashboard/src/locales/en.json @@ -1043,6 +1043,14 @@ "rejectedLabel": "Rejected", "rejected": "Rejected by user" }, + "questions": { + "recommended": "Recommended", + "customPlaceholder": "Type another answer…", + "skip": "Skip", + "next": "Next", + "submit": "Submit", + "skipped": "Skipped" + }, "modifiedFiles": "Modified files ({{count}})", "openBrowser": "View browser", "openBrowserHint": "Agent is browsing the web", @@ -1069,6 +1077,7 @@ "bash": "Shell (bash)", "current_time": "Current time", "write_todos": "Write plan", + "ask_user_question": "Ask user", "task": "Sub-agent task", "ls": "List directory", "glob": "Find files", diff --git a/dashboard/src/locales/zh.json b/dashboard/src/locales/zh.json index 97ea00e9..97bf7532 100644 --- a/dashboard/src/locales/zh.json +++ b/dashboard/src/locales/zh.json @@ -1042,6 +1042,14 @@ "rejectedLabel": "已拒绝", "rejected": "用户已拒绝" }, + "questions": { + "recommended": "推荐", + "customPlaceholder": "输入其他答案…", + "skip": "跳过", + "next": "继续", + "submit": "提交", + "skipped": "已跳过" + }, "modifiedFiles": "已修改文件（{{count}}）", "openBrowser": "查看浏览器", "openBrowserHint": "Agent 正在网页中操作", @@ -1068,6 +1076,7 @@ "bash": "Shell 命令 (bash)", "current_time": "当前时间", "write_todos": "编写计划", + "ask_user_question": "询问用户", "task": "子智能体任务", "ls": "列出目录", "glob": "查找文件", diff --git a/dashboard/src/pages/Chat/components/AskUserQuestionCard.test.tsx b/dashboard/src/pages/Chat/components/AskUserQuestionCard.test.tsx new file mode 100644 index 00000000..6ceac986 --- /dev/null +++ b/dashboard/src/pages/Chat/components/AskUserQuestionCard.test.tsx @@ -0,0 +1,62 @@ +import { fireEvent, render, screen } from "@testing-library/react"; +import { describe, expect, it, vi } from "vitest"; + +import AskUserQuestionCard from "./AskUserQuestionCard"; + +describe("AskUserQuestionCard", () => { + it("keeps the final single choice explicit until Submit", () => { + const onSubmit = vi.fn(); + render( + , + ); + + fireEvent.click(screen.getByRole("button", { name: /SQLite/ })); + expect(onSubmit).not.toHaveBeenCalled(); + fireEvent.click( + screen.getByRole("button", { name: "chat.questions.submit" }), + ); + expect(onSubmit).toHaveBeenCalledWith("pending-1", [ + { id: "database", selected: ["SQLite"] }, + ]); + }); + + it("advances after a non-final single choice", () => { + render( + , + ); + + fireEvent.click(screen.getByRole("button", { name: /A/ })); + expect(screen.getByText("Second question?")).toBeInTheDocument(); + expect(screen.getByText("2 / 2")).toBeInTheDocument(); + }); +}); diff --git a/dashboard/src/pages/Chat/components/AskUserQuestionCard.tsx b/dashboard/src/pages/Chat/components/AskUserQuestionCard.tsx new file mode 100644 index 00000000..3fa90453 --- /dev/null +++ b/dashboard/src/pages/Chat/components/AskUserQuestionCard.tsx @@ -0,0 +1,234 @@ +import { useMemo, useState } from "react"; +import { Button, Input } from "antd"; +import { Check, ChevronLeft, ChevronRight } from "lucide-react"; +import { useTranslation } from "react-i18next"; + +import type { UserQuestionAnswer } from "../../../api/types/userQuestions"; +import type { UserQuestionRequestData } from "../hooks/sseHelpers"; +import styles from "../index.module.less"; + +interface DraftAnswer { + selected: string[]; + custom: string; + skipped: boolean; +} + +interface AskUserQuestionCardProps { + data: UserQuestionRequestData; + onSubmit?: (pendingId: string, answers: UserQuestionAnswer[]) => void; +} + +function displayOption(label: string): { + label: string; + recommended: boolean; +} { + const suffix = + /\s*(?:\((?:recommended|推荐)\)|（(?:recommended|推荐)）)\s*$/i; + return { + label: label.replace(suffix, ""), + recommended: suffix.test(label), + }; +} + +export default function AskUserQuestionCard({ + data, + onSubmit, +}: AskUserQuestionCardProps) { + const { t } = useTranslation(); + const [index, setIndex] = useState(0); + const [drafts, setDrafts] = useState(() => + data.questions.map(() => ({ selected: [], custom: "", skipped: false })), + ); + const question = data.questions[index]; + const draft = drafts[index]; + const pending = data.status === "pending"; + + const updateDraft = (next: DraftAnswer) => { + setDrafts((current) => + current.map((item, itemIndex) => (itemIndex === index ? next : item)), + ); + }; + + const choose = (label: string) => { + if (!question || !draft || !pending) return; + if (question.multi_select) { + const selected = draft.selected.includes(label) + ? draft.selected.filter((item) => item !== label) + : [...draft.selected, label]; + updateDraft({ ...draft, selected, skipped: false }); + return; + } + updateDraft({ selected: [label], custom: "", skipped: false }); + if (index < data.questions.length - 1) setIndex(index + 1); + }; + + const answered = (value: DraftAnswer) => + value.skipped || value.selected.length > 0 || Boolean(value.custom.trim()); + const allAnswered = drafts.every(answered); + + const submit = (values = drafts) => { + if (!onSubmit || !allAnswered) return; + onSubmit( + data.pendingId, + data.questions.map((item, itemIndex) => { + const value = values[itemIndex]; + const custom = value.custom.trim(); + return { + id: item.id, + selected: value.skipped + ? [] + : custom && !item.multi_select + ? [] + : value.selected, + ...(custom ? { custom } : {}), + }; + }), + ); + }; + + const continueFlow = () => { + if (!draft || !answered(draft)) return; + if (index < data.questions.length - 1) setIndex(index + 1); + else submit(); + }; + + const skip = () => { + const next = drafts.map((item, itemIndex) => + itemIndex === index ? { selected: [], custom: "", skipped: true } : item, + ); + setDrafts(next); + if (index < data.questions.length - 1) setIndex(index + 1); + else if (next.every(answered) && onSubmit) { + onSubmit( + data.pendingId, + data.questions.map((item, itemIndex) => ({ + id: item.id, + selected: next[itemIndex].selected, + ...(next[itemIndex].custom.trim() + ? { custom: next[itemIndex].custom.trim() } + : {}), + })), + ); + } + }; + + const resolvedSummary = useMemo(() => { + if (pending || !data.answers?.length) return ""; + return data.answers + .map((answer) => answer.custom || answer.selected.join(", ")) + .filter(Boolean) + .join(" · "); + }, [data.answers, pending]); + + if (!question || !draft) return null; + + return ( +
+
+
+ {question.header ? ( +
{question.header}
+ ) : null} +
{question.question}
+
+
+ {index + 1} / {data.questions.length} +
+
+ + {pending ? ( +
+ {question.options.map((option, optionIndex) => { + const selected = draft.selected.includes(option.label); + const display = displayOption(option.label); + return ( + + ); + })} + + updateDraft({ + selected: question.multi_select ? draft.selected : [], + custom: event.target.value, + skipped: false, + }) + } + onPressEnter={(event) => { + if (event.shiftKey || event.nativeEvent.isComposing) return; + event.preventDefault(); + continueFlow(); + }} + /> +
+ ) : ( +
+ {resolvedSummary || t("chat.questions.skipped")} +
+ )} + + {pending ? ( +
+
+
+
+ + +
+
+ ) : null} +
+ ); +} diff --git a/dashboard/src/pages/Chat/components/AssistantTurnView.tsx b/dashboard/src/pages/Chat/components/AssistantTurnView.tsx index 17a3117a..eb58ddc3 100644 --- a/dashboard/src/pages/Chat/components/AssistantTurnView.tsx +++ b/dashboard/src/pages/Chat/components/AssistantTurnView.tsx @@ -21,6 +21,7 @@ import MessageBubble from "./MessageBubble"; import { ToolMediaStrip } from "./ToolMediaStrip"; import { collectTurnToolMedia } from "../../../utils/collectTurnToolMedia"; import styles from "../index.module.less"; +import type { UserQuestionAnswer } from "../../../api/types/userQuestions"; interface AssistantTurnViewProps { messages: ChatMessage[]; @@ -34,6 +35,7 @@ interface AssistantTurnViewProps { onHitlDecision?: ( decisions: Array<{ type: string; message?: string }>, ) => void; + onQuestionAnswer?: (pendingId: string, answers: UserQuestionAnswer[]) => void; onOpenBrowser?: () => void; onEditFile?: () => void; onRunShellCommand?: (code: string) => void; @@ -58,6 +60,7 @@ export default function AssistantTurnView({ onEditUserMessage, onAcpPermissionSelect, onHitlDecision, + onQuestionAnswer, onOpenBrowser, onEditFile, onRunShellCommand, @@ -73,7 +76,10 @@ export default function AssistantTurnView({ () => layoutAssistantTurnHitl(messages), [messages], ); - const hasPendingHitl = messages.some((m) => m.hitlData?.status === "pending"); + const hasPendingHitl = messages.some( + (m) => + m.hitlData?.status === "pending" || m.questionData?.status === "pending", + ); const segmentProcess = useMemo( () => @@ -164,6 +170,7 @@ export default function AssistantTurnView({ diff --git a/dashboard/src/pages/Chat/components/MessageBubble.tsx b/dashboard/src/pages/Chat/components/MessageBubble.tsx index 2451e605..bbd61311 100644 --- a/dashboard/src/pages/Chat/components/MessageBubble.tsx +++ b/dashboard/src/pages/Chat/components/MessageBubble.tsx @@ -46,6 +46,8 @@ import { isChatStreamError, } from "../../../utils/chatStreamError"; import { MessageFileCard } from "./MessageFileCard"; +import AskUserQuestionCard from "./AskUserQuestionCard"; +import type { UserQuestionAnswer } from "../../../api/types/userQuestions"; import styles from "../index.module.less"; interface MessageBubbleProps { @@ -60,6 +62,7 @@ interface MessageBubbleProps { onHitlDecision?: ( decisions: Array<{ type: string; message?: string }>, ) => void; + onQuestionAnswer?: (pendingId: string, answers: UserQuestionAnswer[]) => void; /** When true, the outer bubble uses reduced spacing (part of a group). */ compact?: boolean; @@ -501,6 +504,7 @@ function MessageBubble({ forkDisabled, forkDisabledHint, onHitlDecision, + onQuestionAnswer, compact, groupPosition = "only", onRunShellCommand, @@ -625,6 +629,21 @@ function MessageBubble({ ); } + if (message.questionData) { + return ( +
+ +
+ ); + } + const isUser = message.role === "user"; const isStreaming = message.status === "streaming"; const hasToolData = !!message.toolData; diff --git a/dashboard/src/pages/Chat/components/MessageList.tsx b/dashboard/src/pages/Chat/components/MessageList.tsx index 8146dade..5aa46b01 100644 --- a/dashboard/src/pages/Chat/components/MessageList.tsx +++ b/dashboard/src/pages/Chat/components/MessageList.tsx @@ -13,6 +13,7 @@ import { Spin, Button } from "antd"; import { Virtuoso, type Components, type VirtuosoHandle } from "react-virtuoso"; import { useTranslation } from "react-i18next"; import type { ChatMessage } from "../hooks/useChat"; +import type { UserQuestionAnswer } from "../../../api/types/userQuestions"; import type { ComposerTagLookups } from "./UserMessageComposerTags"; import MessageBubble from "./MessageBubble"; import AssistantTurnView from "./AssistantTurnView"; @@ -103,6 +104,7 @@ interface MessageListProps { onHitlDecision?: ( decisions: Array<{ type: string; message?: string }>, ) => void; + onQuestionAnswer?: (pendingId: string, answers: UserQuestionAnswer[]) => void; onOpenBrowser?: () => void; onEditFile?: () => void; onRunShellCommand?: (code: string) => void; @@ -127,6 +129,7 @@ interface GroupRenderContext { onHitlDecision?: ( decisions: Array<{ type: string; message?: string }>, ) => void; + onQuestionAnswer?: (pendingId: string, answers: UserQuestionAnswer[]) => void; onOpenBrowser?: () => void; onEditFile?: () => void; onRunShellCommand?: (code: string) => void; @@ -169,6 +172,7 @@ function renderMessageGroup( onEditUserMessage={ctx.onEditUserMessage} onAcpPermissionSelect={ctx.onAcpPermissionSelect} onHitlDecision={ctx.onHitlDecision} + onQuestionAnswer={ctx.onQuestionAnswer} onOpenBrowser={openBrowserHandler} onEditFile={ctx.onEditFile} onRunShellCommand={ctx.onRunShellCommand} @@ -217,6 +221,7 @@ function renderMessageGroup( onEditUserMessage={ctx.onEditUserMessage} onAcpPermissionSelect={ctx.onAcpPermissionSelect} onHitlDecision={ctx.onHitlDecision} + onQuestionAnswer={ctx.onQuestionAnswer} onOpenBrowser={openBrowserHandler} onEditFile={ctx.onEditFile} onRunShellCommand={ctx.onRunShellCommand} @@ -250,6 +255,7 @@ export default function MessageList(props: MessageListProps) { forkDisabledHint, onAcpPermissionSelect, onHitlDecision, + onQuestionAnswer, onOpenBrowser, onEditFile, onRunShellCommand, @@ -678,6 +684,7 @@ export default function MessageList(props: MessageListProps) { forkDisabledHint, onAcpPermissionSelect, onHitlDecision, + onQuestionAnswer, onOpenBrowser, onEditFile, onRunShellCommand, @@ -700,6 +707,7 @@ export default function MessageList(props: MessageListProps) { forkDisabledHint, onAcpPermissionSelect, onHitlDecision, + onQuestionAnswer, onOpenBrowser, onEditFile, onRunShellCommand, diff --git a/dashboard/src/pages/Chat/hooks/chatStore.ts b/dashboard/src/pages/Chat/hooks/chatStore.ts index 522e5600..f9d53808 100644 --- a/dashboard/src/pages/Chat/hooks/chatStore.ts +++ b/dashboard/src/pages/Chat/hooks/chatStore.ts @@ -26,6 +26,10 @@ import { isChatStreamError } from "../../../utils/chatStreamError"; import { buildUserMessageContent } from "../utils/chatAttachments"; import { sealPriorStreamingAssistants as sealPriorStreamingAssistantsMessages } from "./sealPriorStreamingAssistants"; import { turnStatusAction } from "./turnStatusGate"; +import type { + UserQuestion, + UserQuestionAnswer, +} from "../../../api/types/userQuestions"; import { MAX_STREAM_RESUME_ATTEMPTS, STREAM_STALE_WITHOUT_SOCKET_MS, @@ -1372,12 +1376,77 @@ function resolveHitlPending( ); } +function parseUserQuestions(raw: Record): UserQuestion[] { + if (!Array.isArray(raw.questions)) return []; + return raw.questions + .filter((item): item is Record => { + return Boolean(item && typeof item === "object"); + }) + .map((item) => ({ + id: typeof item.id === "string" ? item.id : "question", + question: typeof item.question === "string" ? item.question : "", + header: typeof item.header === "string" ? item.header : undefined, + options: Array.isArray(item.options) + ? item.options + .filter((option): option is Record => { + return Boolean(option && typeof option === "object"); + }) + .map((option) => ({ + label: typeof option.label === "string" ? option.label : "", + description: + typeof option.description === "string" + ? option.description + : undefined, + })) + .filter((option) => option.label) + : [], + multi_select: item.multi_select === true, + })) + .filter((question) => question.question); +} + +function resolveQuestionPending( + state: SessionStreamState, + pendingId: string, + status: "pending" | "answered" | "cancelled", + answers?: UserQuestionAnswer[], +): void { + state.messages = state.messages.map((message) => + message.questionData?.pendingId === pendingId + ? { + ...message, + questionData: { ...message.questionData, status, answers }, + } + : message, + ); +} + function handleHitlRequired( state: SessionStreamState, request: Record, ): void { finalizeStreamingMessages(state); clearStreamingFlags(state); + if (request.kind === "ask_user_question") { + const pendingId = + typeof request.pending_id === "string" ? request.pending_id : ""; + state.messages = [ + ...state.messages, + { + id: pendingId ? `question-${pendingId}` : generateId(), + role: "assistant", + content: "", + questionData: { + pendingId, + questions: parseUserQuestions(request), + status: "pending", + }, + status: "done", + timestamp: Date.now(), + }, + ]; + return; + } state.messages = [ ...state.messages, { @@ -2073,3 +2142,68 @@ export async function resumeHitl( finish(); } } + +export async function answerUserQuestion( + sessionId: string, + agentId: string, + threadId: string, + pendingId: string, + answers: UserQuestionAnswer[], +): Promise { + const state = getOrCreate(sessionId); + state.abortController?.abort(); + resolveQuestionPending(state, pendingId, "answered", answers); + beginStream(state, sessionId); + notify(state); + emitStreamEvent({ kind: "streamStart", sessionId }); + + const controller = new AbortController(); + state.abortController = controller; + const headers: HeadersInit = { + "Content-Type": "application/json", + Accept: "text/event-stream", + }; + const token = getAuthToken(); + if (token) { + (headers as Record).Authorization = `Bearer ${token}`; + } + const finish = () => { + clearStreamingFlags(state); + clearStreamActivity(sessionId); + pendingResumeBySession.delete(sessionId); + state.abortController = null; + state.streamMsg = ""; + state.streamId = ""; + state.streamBlockType = ""; + sealInFlightAssistantMessages(state); + notify(state); + emitStreamEvent({ kind: "streamEnd", sessionId }); + }; + try { + const res = await fetch( + getApiUrl( + `/agents/${agentId}/chat/questions/${encodeURIComponent( + pendingId, + )}/answer`, + ), + { + method: "POST", + headers, + body: JSON.stringify({ thread_id: threadId, answers }), + signal: controller.signal, + }, + ); + if (!res.ok) { + resolveQuestionPending(state, pendingId, "pending"); + appendErrorBubble(state, `Question resume failed (${res.status})`); + finish(); + return; + } + await consumeSseResponse(state, sessionId, res, controller, finish); + } catch (err: unknown) { + if ((err as Error).name === "AbortError") return; + resolveQuestionPending(state, pendingId, "pending"); + appendErrorBubble(state, (err as Error).message || "Question resume error"); + finish(); + } +} diff --git a/dashboard/src/pages/Chat/hooks/sseHelpers.ts b/dashboard/src/pages/Chat/hooks/sseHelpers.ts index 00351d2f..33474296 100644 --- a/dashboard/src/pages/Chat/hooks/sseHelpers.ts +++ b/dashboard/src/pages/Chat/hooks/sseHelpers.ts @@ -11,6 +11,10 @@ import type { TokenUsage, } from "../../../api/types"; import type { ContentBlockItem } from "../../../utils/messageParser"; +import type { + UserQuestion, + UserQuestionAnswer, +} from "../../../api/types/userQuestions"; export interface ToolCallData { name?: string; @@ -34,6 +38,13 @@ export interface HitlRequestData { status?: "pending" | "approved" | "rejected"; } +export interface UserQuestionRequestData { + pendingId: string; + questions: UserQuestion[]; + answers?: UserQuestionAnswer[]; + status?: "pending" | "answered" | "cancelled"; +} + export interface ChatAttachment { url: string; filename?: string; @@ -64,6 +75,7 @@ export interface ChatMessage { composerContext?: UserComposerContext; toolData?: ToolCallData; hitlData?: HitlRequestData; + questionData?: UserQuestionRequestData; usage?: TokenUsage; metadata?: MessageMetadata; errorInfo?: ProcessErrorInfo; diff --git a/dashboard/src/pages/Chat/hooks/useChat.ts b/dashboard/src/pages/Chat/hooks/useChat.ts index 50d60181..62c4e789 100644 --- a/dashboard/src/pages/Chat/hooks/useChat.ts +++ b/dashboard/src/pages/Chat/hooks/useChat.ts @@ -28,6 +28,7 @@ import { resolveMessageTimestampMs } from "../../../utils/formatMessageTime"; import { isImageAttachment } from "../utils/chatAttachments"; import { agentAttachmentAccessUrl } from "../../../utils/toolMediaBlocks"; import { injectPendingHitlMessage } from "../../../utils/injectPendingHitlMessage"; +import { injectPendingUserQuestion } from "../../../utils/injectPendingUserQuestion"; import type { ChatAttachment, ChatMessage, @@ -550,17 +551,20 @@ async function loadThreadHistory( limit, offset, }); - const messages = injectPendingHitlMessage( - convertHistoryMessages( - history.messages.filter( - (message) => - message.role === "user" || - message.role === "assistant" || - message.role === "tool", + const messages = injectPendingUserQuestion( + injectPendingHitlMessage( + convertHistoryMessages( + history.messages.filter( + (message) => + message.role === "user" || + message.role === "assistant" || + message.role === "tool", + ), + agentId, ), - agentId, + history.hitl_pending, ), - history.hitl_pending, + history.question_pending, ); return { messages, @@ -880,6 +884,28 @@ export function useChat( [agentId, stableSessionId], ); + const answerUserQuestion = useCallback( + ( + pendingId: string, + answers: import("../../../api/types/userQuestions").UserQuestionAnswer[], + storeKey?: string, + ) => { + if (!agentId) return; + const key = storeKey || stableSessionId; + const threadId = + storeKey || (stableSessionId !== "__empty__" ? stableSessionId : ""); + if (!threadId || threadId === "__empty__") return; + void chatStore.answerUserQuestion( + key, + agentId, + threadId, + pendingId, + answers, + ); + }, + [agentId, stableSessionId], + ); + return { messages, isStreaming, @@ -899,5 +925,6 @@ export function useChat( refreshHistory, clearMessages, resumeHitl, + answerUserQuestion, }; } diff --git a/dashboard/src/pages/Chat/index.module.less b/dashboard/src/pages/Chat/index.module.less index b45de9cf..cdf7a305 100644 --- a/dashboard/src/pages/Chat/index.module.less +++ b/dashboard/src/pages/Chat/index.module.less @@ -29,6 +29,144 @@ background: var(--fn-bg-elevated, #fff); } +// Structured question card. It intentionally lives with the assistant turn in +// this MR; the Todo panel's broader visual refresh is kept separate. +.askUserCard { + width: min(620px, 100%); + overflow: hidden; + border: 1px solid var(--border-color, #e5e7eb); + border-radius: 14px; + background: var(--card-bg, #fff); + box-shadow: 0 8px 30px rgb(15 23 42 / 8%); +} + +.askUserHeader { + display: flex; + align-items: flex-start; + justify-content: space-between; + gap: 16px; + padding: 16px 18px 13px; + border-bottom: 1px solid var(--border-color, #e5e7eb); +} + +.askUserEyebrow { + margin-bottom: 4px; + color: var(--text-tertiary, #64748b); + font-size: 11px; + font-weight: 600; + letter-spacing: 0.08em; + text-transform: uppercase; +} + +.askUserTitle { + color: var(--text-primary, #111827); + font-size: 15px; + font-weight: 600; + line-height: 1.5; +} + +.askUserProgress { + flex: none; + padding-top: 2px; + color: var(--text-tertiary, #64748b); + font-size: 12px; +} + +.askUserBody { + display: flex; + flex-direction: column; + gap: 8px; + max-height: min(52vh, 440px); + overflow-y: auto; + padding: 14px 18px; +} + +.askUserOption { + display: flex; + align-items: flex-start; + gap: 10px; + width: 100%; + padding: 10px 12px; + border: 1px solid var(--border-color, #e5e7eb); + border-radius: 10px; + background: transparent; + color: inherit; + text-align: left; + cursor: pointer; + transition: + border-color 0.16s ease, + background 0.16s ease; + + &:hover, + &Selected { + border-color: #4f7cff; + background: rgb(79 124 255 / 7%); + } +} + +.askUserOptionMarker { + display: grid; + flex: none; + place-items: center; + width: 20px; + height: 20px; + border-radius: 6px; + background: var(--fill-secondary, #f1f5f9); + color: var(--text-secondary, #475569); + font-size: 11px; +} + +.askUserOptionCopy { + display: flex; + min-width: 0; + flex-direction: column; + gap: 3px; + font-size: 13px; + line-height: 20px; + + small { + color: var(--text-tertiary, #64748b); + font-size: 12px; + } +} + +.askUserRecommended { + display: inline-flex; + margin-left: 8px; + padding: 0 6px; + border-radius: 999px; + background: rgb(79 124 255 / 12%); + color: #4f7cff; + font-size: 10px; + font-weight: 600; +} + +.askUserCustom { + margin-top: 2px; +} + +.askUserFooter { + display: flex; + align-items: center; + justify-content: space-between; + gap: 12px; + padding: 11px 14px; + border-top: 1px solid var(--border-color, #e5e7eb); +} + +.askUserPager, +.askUserActions { + display: flex; + align-items: center; + gap: 8px; +} + +.askUserResolved { + padding: 14px 18px; + color: var(--text-secondary, #475569); + font-size: 13px; +} + /* ---------- Main chat area ---------- */ .chatMain { flex: 1; diff --git a/dashboard/src/pages/Chat/index.tsx b/dashboard/src/pages/Chat/index.tsx index 36b622f1..9072e961 100644 --- a/dashboard/src/pages/Chat/index.tsx +++ b/dashboard/src/pages/Chat/index.tsx @@ -235,6 +235,7 @@ function ChatPageInner() { refreshHistory, clearMessages, resumeHitl, + answerUserQuestion, } = useChat(activeThreadId, resolvedAgentId); const refreshBrowserRef = useRef<() => void>(() => {}); @@ -544,6 +545,16 @@ function ChatPageInner() { [resumeHitl, activeThreadId], ); + const handleQuestionAnswer = useCallback( + ( + pendingId: string, + answers: import("../../api/types/userQuestions").UserQuestionAnswer[], + ) => { + answerUserQuestion(pendingId, answers, activeThreadId ?? undefined); + }, + [answerUserQuestion, activeThreadId], + ); + useEffect(() => { let cancelled = false; browserApi @@ -593,7 +604,12 @@ function ChatPageInner() { const [forking, setForking] = useState(false); const hasPendingHitl = useMemo( - () => messages.some((message) => message.hitlData?.status === "pending"), + () => + messages.some( + (message) => + message.hitlData?.status === "pending" || + message.questionData?.status === "pending", + ), [messages], ); const forkDisabled = forking || isStreaming || hasPendingHitl; @@ -820,6 +836,7 @@ function ChatPageInner() { forkDisabledHint={forkDisabledHint} onAcpPermissionSelect={handleAcpPermissionSelect} onHitlDecision={handleHitlDecision} + onQuestionAnswer={handleQuestionAnswer} onOpenBrowser={ hasBrowserTool && !isMobile ? openBrowserTab : undefined } @@ -962,7 +979,7 @@ function ChatPageInner() { onCancel={cancelStream} onNewChat={handleNewChat} isStreaming={isStreaming} - disabled={!agentChatReady || noAgents} + disabled={!agentChatReady || noAgents || hasPendingHitl} initialText={prefillInputRef.current} onComposerCleared={() => { prefillInputRef.current = ""; diff --git a/dashboard/src/pages/Chat/utils/layoutAssistantTurnHitl.ts b/dashboard/src/pages/Chat/utils/layoutAssistantTurnHitl.ts index e7965ee5..cde69b72 100644 --- a/dashboard/src/pages/Chat/utils/layoutAssistantTurnHitl.ts +++ b/dashboard/src/pages/Chat/utils/layoutAssistantTurnHitl.ts @@ -22,7 +22,7 @@ export function layoutAssistantTurnHitl( const segments: AssistantTurnHitlSegment[] = []; let buf: ChatMessage[] = []; for (const m of messages) { - if (m.hitlData) { + if (m.hitlData || m.questionData) { segments.push({ processMessages: buf, hitlMessage: m }); buf = []; } else { diff --git a/dashboard/src/utils/injectPendingUserQuestion.ts b/dashboard/src/utils/injectPendingUserQuestion.ts new file mode 100644 index 00000000..a40c39bd --- /dev/null +++ b/dashboard/src/utils/injectPendingUserQuestion.ts @@ -0,0 +1,27 @@ +import type { UserQuestionPendingPayload } from "../api/types/userQuestions"; +import type { ChatMessage } from "../pages/Chat/hooks/useChat"; + +export function injectPendingUserQuestion( + messages: ChatMessage[], + pending: UserQuestionPendingPayload | null | undefined, +): ChatMessage[] { + if (!pending?.pending_id || !pending.questions?.length) return messages; + if (messages.some((message) => message.questionData?.status === "pending")) { + return messages; + } + return [ + ...messages, + { + id: `question-${pending.pending_id}`, + role: "assistant", + content: "", + questionData: { + pendingId: pending.pending_id, + questions: pending.questions, + status: "pending", + }, + status: "done", + timestamp: Date.now(), + }, + ]; +} diff --git a/src/octop/api/routers/chat/history.py b/src/octop/api/routers/chat/history.py index 6967bf7a..cefacbf7 100644 --- a/src/octop/api/routers/chat/history.py +++ b/src/octop/api/routers/chat/history.py @@ -173,6 +173,11 @@ async def get_thread_history( agent_id=agent_id, user_id=effective_uid, ) + question_pending = server.app_runtime.gateway.question_coordinator.pending_payload( + thread_id=thread_id, + agent_id=agent_id, + user_id=effective_uid, + ) return { "thread_id": thread_id, "messages": messages, @@ -181,6 +186,7 @@ async def get_thread_history( "offset": page_offset, "turn_active": server.app_runtime.gateway.ws_hub.is_turn_active(thread_id), "hitl_pending": hitl_pending, + "question_pending": question_pending, } diff --git a/src/octop/api/routers/chat/models.py b/src/octop/api/routers/chat/models.py index 13661020..05efe424 100644 --- a/src/octop/api/routers/chat/models.py +++ b/src/octop/api/routers/chat/models.py @@ -147,3 +147,11 @@ class HitlResumeBody(BaseModel): ..., description='Human decisions, e.g. [{"type": "approve"}] or [{"type": "reject", "message": "..."}].', ) + + +class UserQuestionAnswerBody(BaseModel): + thread_id: str = Field(..., description="Conversation thread awaiting an answer.") + answers: list[dict[str, Any]] = Field( + ..., + description="One structured answer for every question in the pending request.", + ) diff --git a/src/octop/api/routers/chat/routes.py b/src/octop/api/routers/chat/routes.py index 5db1fcd7..8bb3908d 100644 --- a/src/octop/api/routers/chat/routes.py +++ b/src/octop/api/routers/chat/routes.py @@ -11,8 +11,9 @@ from octop.api.common.agent import assert_agent_access from octop.api.deps import current_user, get_server -from octop.api.routers.chat.models import HitlResumeBody, PolishBody +from octop.api.routers.chat.models import HitlResumeBody, PolishBody, UserQuestionAnswerBody from octop.api.routers.chat.sse import format_sse +from octop.i18n import tr from octop.i18n.domains.stream import format_stream_error from octop.infra.agents.experts.catalog import ( default_welcome_payload, @@ -23,6 +24,11 @@ from octop.infra.errors import ErrorCode, OctopError from octop.infra.gateway.hitl.coordinator import HitlChannelCoordinator, HitlStreamContext from octop.infra.gateway.hitl.store import HitlPendingRecord +from octop.infra.gateway.questions.coordinator import ( + UserQuestionCoordinator, + is_user_question_request, + validate_answers, +) from octop.infra.utils.llm_text import ainvoke_text from octop.infra.utils.locale import resolve_request_locale @@ -93,6 +99,7 @@ async def iter_dashboard_hitl_resume_sse( channel_type: str, locale: str, is_disconnected: Callable[[], Awaitable[bool]], + question_coordinator: UserQuestionCoordinator | None = None, ) -> AsyncIterator[str]: """Stream dashboard HITL resume chunks and persist any nested ``hitl_required``. @@ -114,7 +121,24 @@ async def iter_dashboard_hitl_resume_sse( if isinstance(chunk, dict) and chunk.get("type") == "hitl_required": request_payload = chunk.get("request") if isinstance(request_payload, dict): - hitl_coordinator.register_from_request(request_payload, ctx=hitl_ctx) + if ( + is_user_question_request(request_payload) + and question_coordinator is not None + ): + record = question_coordinator.register_from_request( + request_payload, + thread_id=thread_id, + agent_id=agent_id, + user_id=user_id, + session_key=session_key, + channel_type=channel_type, + ) + chunk = { + **chunk, + "request": {**request_payload, "pending_id": record.pending_id}, + } + else: + hitl_coordinator.register_from_request(request_payload, ctx=hitl_ctx) yield format_sse("chunk", chunk) if pending is not None: hitl_coordinator.store.mark_resolved( @@ -183,12 +207,106 @@ async def gen() -> AsyncIterator[str]: channel_type=channel_type, locale=resolve_request_locale(request), is_disconnected=request.is_disconnected, + question_coordinator=server.app_runtime.gateway.question_coordinator, ): yield frame return StreamingResponse(gen(), media_type="text/event-stream") +@router.post( + "/agents/{agent_id}/chat/questions/{pending_id}/answer", + summary="Answer a pending agent question (SSE)", +) +async def answer_user_question( + agent_id: str, + pending_id: str, + body: UserQuestionAnswerBody, + request: Request, + user: Any = Depends(current_user), + server: Any = Depends(get_server), +) -> StreamingResponse: + """Resume a durable ``ask_user_question`` interrupt and stream the result.""" + assert_agent_access(server, agent_id, user) + coordinator = server.app_runtime.gateway.question_coordinator + row = coordinator.repo.get(pending_id) + if ( + row is None + or row.status != "pending" + or row.agent_id != agent_id + or row.user_id != user.id + or row.thread_id != body.thread_id + ): + raise OctopError(ErrorCode.FORBIDDEN, "pending question is unavailable") + try: + answers = validate_answers(row, body.answers) + except ValueError as exc: + raise OctopError(ErrorCode.SLASH_BAD_ARGS, str(exc)) from exc + + async def gen() -> AsyncIterator[str]: + if not coordinator.repo.claim(pending_id, agent_id=agent_id, user_id=user.id): + yield format_sse( + "chunk", + { + "type": "error", + "message": tr("slash.questions.none_pending", resolve_request_locale(request)), + }, + ) + return + completed = False + try: + hitl = server.app_runtime.gateway.processor.hitl_coordinator + hitl_ctx = HitlStreamContext( + thread_id=row.thread_id, + agent_id=row.agent_id, + user_id=row.user_id, + session_key=row.session_key, + channel_type=row.channel_type, + ) + async for chunk in server.app_runtime.agent_registry.resume_hitl( + agent_id, + row.thread_id, + [{"type": "answer", "answers": answers}], + ): + if await request.is_disconnected(): + return + if isinstance(chunk, dict) and chunk.get("type") == "hitl_required": + nested = chunk.get("request") + if isinstance(nested, dict): + if is_user_question_request(nested): + nested_row = coordinator.register_from_request( + nested, + thread_id=row.thread_id, + agent_id=row.agent_id, + user_id=row.user_id, + session_key=row.session_key, + channel_type=row.channel_type, + ) + chunk = { + **chunk, + "request": {**nested, "pending_id": nested_row.pending_id}, + } + else: + hitl.register_from_request(nested, ctx=hitl_ctx) + yield format_sse("chunk", chunk) + coordinator.repo.mark_answered(pending_id, answers) + completed = True + yield format_sse("chunk", {"type": "done"}) + except Exception as exc: + yield format_sse( + "chunk", + { + "type": "error", + "message": format_stream_error(exc, resolve_request_locale(request)), + }, + ) + finally: + if not completed: + coordinator.repo.release(pending_id) + + return StreamingResponse(gen(), media_type="text/event-stream") + + @router.post("/agents/{agent_id}/chat/polish", summary="Polish prompt") async def polish_prompt( agent_id: str, diff --git a/src/octop/i18n/domains/tools.py b/src/octop/i18n/domains/tools.py index 38f62610..65595a1c 100644 --- a/src/octop/i18n/domains/tools.py +++ b/src/octop/i18n/domains/tools.py @@ -15,6 +15,7 @@ "unknown", "current_time", "write_todos", + "ask_user_question", # Memory "memory_search", "memory_get", diff --git a/src/octop/i18n/en.json b/src/octop/i18n/en.json index 55c73726..7e14bb06 100644 --- a/src/octop/i18n/en.json +++ b/src/octop/i18n/en.json @@ -71,6 +71,17 @@ "pending_line": "`{pending_id}` — {count} tool(s)", "dashboard_hint": "Tool approval in chat uses the approval card in the dashboard, or `/approve` / `/reject` in IM channels." }, + "questions": { + "card_title": "❓ The agent needs your input", + "card_footer_single": "Reply with `/answer `. For options, use the number or label.", + "card_footer_multiple": "Reply with `/answer 1=answer; 2=answer`.", + "card_pending_id": "Question ID: `{pending_id}`", + "none_pending": "There is no pending question in this conversation.", + "invalid_answer": "Could not use that answer: {error}", + "answer_ack": "Answer received. Continuing…", + "resume_failed": "Failed to continue after the answer: {error}", + "dashboard_hint": "Use the interactive question card in the dashboard. `/answer` is available in IM channels." + }, "title": { "usage": "Usage: /title ", "done": "Title set: {title}" @@ -240,6 +251,10 @@ "label": "Pending approvals", "description": "List tool approvals waiting for your decision" }, + "answer": { + "label": "Answer question", + "description": "Answer a structured question waiting in this conversation" + }, "title": { "label": "Set title", "description": "Rename the current thread" @@ -430,6 +445,7 @@ "execute": "Execute", "current_time": "Current time", "write_todos": "Write plan", + "ask_user_question": "Ask user", "task": "Sub-agent task", "ls": "List directory", "glob": "Find files", diff --git a/src/octop/i18n/zh.json b/src/octop/i18n/zh.json index 135df9d4..3aced20b 100644 --- a/src/octop/i18n/zh.json +++ b/src/octop/i18n/zh.json @@ -71,6 +71,17 @@ "pending_line": "`{pending_id}` — {count} 个工具", "dashboard_hint": "控制台聊天请使用审批卡片；IM 通道请使用 `/approve` 或 `/reject`。" }, + "questions": { + "card_title": "❓ Agent 需要你的回答", + "card_footer_single": "回复 `/answer <答案>`；选择题可填写序号或选项文字。", + "card_footer_multiple": "回复 `/answer 1=答案; 2=答案`。", + "card_pending_id": "问题 ID：`{pending_id}`", + "none_pending": "当前会话没有待回答问题。", + "invalid_answer": "无法使用这个回答：{error}", + "answer_ack": "已收到回答，继续执行…", + "resume_failed": "回答后恢复执行失败：{error}", + "dashboard_hint": "控制台请使用交互式问题卡片；IM 通道可使用 `/answer`。" + }, "title": { "usage": "用法：/title <标题>", "done": "标题已设置：{title}" @@ -240,6 +251,10 @@ "label": "待审批", "description": "列出等待您决定的工具审批" }, + "answer": { + "label": "回答问题", + "description": "回答当前会话中等待处理的结构化问题" + }, "title": { "label": "设置标题", "description": "为当前对话设置标题" @@ -430,6 +445,7 @@ "execute": "执行指令", "current_time": "当前时间", "write_todos": "编写计划", + "ask_user_question": "询问用户", "task": "子智能体任务", "ls": "列出目录", "glob": "查找文件", diff --git a/src/octop/infra/agents/ask_user_question.py b/src/octop/infra/agents/ask_user_question.py new file mode 100644 index 00000000..65de8101 --- /dev/null +++ b/src/octop/infra/agents/ask_user_question.py @@ -0,0 +1,119 @@ +"""Structured human-question tool backed by a durable LangGraph interrupt.""" + +from __future__ import annotations + +import json +from typing import Annotated, Any + +from langchain_core.tools import StructuredTool +from langgraph.types import interrupt +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator + +ASK_USER_QUESTION_TOOL = "ask_user_question" + + +class AskUserOption(BaseModel): + model_config = ConfigDict(extra="forbid") + + label: str = Field(min_length=1, max_length=80) + description: str | None = Field(default=None, max_length=240) + + @field_validator("label") + @classmethod + def _strip_label(cls, value: str) -> str: + return value.strip() + + +class AskUserPrompt(BaseModel): + model_config = ConfigDict(extra="forbid") + + id: str = Field(min_length=1, max_length=64, pattern=r"^[A-Za-z0-9_-]+$") + question: str = Field(min_length=1, max_length=500) + header: str | None = Field(default=None, max_length=24) + options: list[AskUserOption] = Field(default_factory=list, max_length=6) + multi_select: bool = False + + @field_validator("question") + @classmethod + def _strip_question(cls, value: str) -> str: + return value.strip() + + @model_validator(mode="after") + def _unique_options(self) -> AskUserPrompt: + labels = [option.label for option in self.options] + if len(labels) != len(set(labels)): + raise ValueError("option labels must be unique within a question") + return self + + +def _answer_from_resume(response: Any, questions: list[AskUserPrompt]) -> dict[str, Any]: + decisions = response.get("decisions") if isinstance(response, dict) else None + decision = decisions[0] if isinstance(decisions, list) and decisions else None + answers = decision.get("answers") if isinstance(decision, dict) else None + if not isinstance(answers, list): + raise ValueError("ask_user_question did not receive a user answer") + + expected = {question.id for question in questions} + normalized: list[dict[str, Any]] = [] + seen: set[str] = set() + for raw in answers: + if not isinstance(raw, dict): + raise ValueError("ask_user_question received an invalid answer") + answer_id = raw.get("id") + if not isinstance(answer_id, str) or answer_id not in expected or answer_id in seen: + raise ValueError("ask_user_question received an unknown or duplicate answer id") + selected_raw = raw.get("selected") + selected = ( + [str(item) for item in selected_raw if isinstance(item, str)] + if isinstance(selected_raw, list) + else [] + ) + custom = raw.get("custom") + item: dict[str, Any] = {"id": answer_id, "selected": selected} + if isinstance(custom, str) and custom.strip(): + item["custom"] = custom.strip() + normalized.append(item) + seen.add(answer_id) + if seen != expected: + raise ValueError("ask_user_question requires one answer for every question") + return {"answers": normalized} + + +def build_ask_user_question_tool() -> StructuredTool: + """Return the model-facing question tool used by every Octop agent.""" + + async def ask_user_question( + questions: Annotated[ + list[AskUserPrompt], + Field( + min_length=1, + max_length=3, + description=( + "One to three concise questions. Put the recommended option first and " + "suffix its label with '(Recommended)'." + ), + ), + ], + ) -> str: + ids = [question.id for question in questions] + if len(ids) != len(set(ids)): + raise ValueError("question ids must be unique") + payload = { + "kind": ASK_USER_QUESTION_TOOL, + "questions": [question.model_dump(exclude_none=True) for question in questions], + } + response = interrupt(payload) + answer = _answer_from_resume(response, questions) + return json.dumps(answer, ensure_ascii=False, separators=(",", ":")) + + return StructuredTool.from_function( + coroutine=ask_user_question, + name=ASK_USER_QUESTION_TOOL, + description=( + "Ask the user one to three structured clarification questions and wait for the " + "answer before continuing. Use this when a user choice materially changes the work." + ), + ) + + +__all__ = ["ASK_USER_QUESTION_TOOL", "build_ask_user_question_tool"] diff --git a/src/octop/infra/agents/manager.py b/src/octop/infra/agents/manager.py index 71da880c..0d44245e 100644 --- a/src/octop/infra/agents/manager.py +++ b/src/octop/infra/agents/manager.py @@ -1955,6 +1955,9 @@ def _build_harness_config(self, row: AgentRow) -> HarnessAgentConfig: ] merged_tools: list[Any] = [] + from octop.infra.agents.ask_user_question import build_ask_user_question_tool + + merged_tools.append(build_ask_user_question_tool()) if cron_tools: merged_tools.extend(cron_tools) merged_tools.extend(knowledge_tools) diff --git a/src/octop/infra/db/migrations/007_pending_user_questions.pg.sql b/src/octop/infra/db/migrations/007_pending_user_questions.pg.sql new file mode 100644 index 00000000..9c9bdde7 --- /dev/null +++ b/src/octop/infra/db/migrations/007_pending_user_questions.pg.sql @@ -0,0 +1,22 @@ +CREATE TABLE pending_user_questions ( + pending_id TEXT PRIMARY KEY, + thread_id TEXT NOT NULL, + agent_id TEXT NOT NULL, + user_id BIGINT NOT NULL, + session_key TEXT NOT NULL, + channel_type TEXT NOT NULL, + questions_json TEXT NOT NULL, + answer_json TEXT, + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'resuming', 'answered', 'cancelled')), + created_at BIGINT NOT NULL, + answered_at BIGINT +); + +CREATE INDEX idx_pending_user_questions_thread + ON pending_user_questions(thread_id, agent_id, user_id, status, created_at DESC); + +CREATE INDEX idx_pending_user_questions_session + ON pending_user_questions(session_key, agent_id, status, created_at DESC); + +UPDATE _schema_version SET version = 7; diff --git a/src/octop/infra/db/migrations/007_pending_user_questions.sql b/src/octop/infra/db/migrations/007_pending_user_questions.sql new file mode 100644 index 00000000..72b0f0dd --- /dev/null +++ b/src/octop/infra/db/migrations/007_pending_user_questions.sql @@ -0,0 +1,22 @@ +CREATE TABLE pending_user_questions ( + pending_id TEXT PRIMARY KEY, + thread_id TEXT NOT NULL, + agent_id TEXT NOT NULL, + user_id INTEGER NOT NULL, + session_key TEXT NOT NULL, + channel_type TEXT NOT NULL, + questions_json TEXT NOT NULL, + answer_json TEXT, + status TEXT NOT NULL DEFAULT 'pending' + CHECK (status IN ('pending', 'resuming', 'answered', 'cancelled')), + created_at INTEGER NOT NULL, + answered_at INTEGER +); + +CREATE INDEX idx_pending_user_questions_thread + ON pending_user_questions(thread_id, agent_id, user_id, status, created_at DESC); + +CREATE INDEX idx_pending_user_questions_session + ON pending_user_questions(session_key, agent_id, status, created_at DESC); + +UPDATE _schema_version SET version = 7; diff --git a/src/octop/infra/db/repos/user_questions.py b/src/octop/infra/db/repos/user_questions.py new file mode 100644 index 00000000..cc7ffdfb --- /dev/null +++ b/src/octop/infra/db/repos/user_questions.py @@ -0,0 +1,170 @@ +"""Durable pending ``ask_user_question`` records.""" + +from __future__ import annotations + +import json +import secrets +from dataclasses import dataclass +from typing import Any, Literal + +from octop.infra.db.pool import DatabasePool +from octop.infra.db.repos._base import DbRow, now_ts + +QuestionStatus = Literal["pending", "resuming", "answered", "cancelled"] + + +def _json_list(value: Any) -> list[dict[str, Any]]: + parsed = json.loads(value) if isinstance(value, str) else value + if not isinstance(parsed, list): + return [] + return [dict(item) for item in parsed if isinstance(item, dict)] + + +@dataclass(frozen=True) +class PendingUserQuestionRow: + pending_id: str + thread_id: str + agent_id: str + user_id: int + session_key: str + channel_type: str + questions: list[dict[str, Any]] + answer: list[dict[str, Any]] | None + status: QuestionStatus + created_at: int + answered_at: int | None + + @classmethod + def from_row(cls, row: DbRow) -> PendingUserQuestionRow: + raw_answer = row["answer_json"] + return cls( + pending_id=str(row["pending_id"]), + thread_id=str(row["thread_id"]), + agent_id=str(row["agent_id"]), + user_id=int(row["user_id"]), + session_key=str(row["session_key"]), + channel_type=str(row["channel_type"]), + questions=_json_list(row["questions_json"]), + answer=_json_list(raw_answer) if raw_answer is not None else None, + status=str(row["status"]), # type: ignore[arg-type] + created_at=int(row["created_at"]), + answered_at=int(row["answered_at"]) if row["answered_at"] is not None else None, + ) + + +class PendingUserQuestionRepo: + def __init__(self, db: DatabasePool) -> None: + self._db = db + + def recover_interrupted_resumes(self) -> None: + with self._db.transaction() as conn: + conn.execute( + "UPDATE pending_user_questions SET status = 'pending' WHERE status = 'resuming'" + ) + + def register( + self, + *, + thread_id: str, + agent_id: str, + user_id: int, + session_key: str, + channel_type: str, + questions: list[dict[str, Any]], + ) -> PendingUserQuestionRow: + encoded = json.dumps(questions, ensure_ascii=False, separators=(",", ":")) + with self._db.transaction() as conn: + existing = conn.execute( + "SELECT * FROM pending_user_questions " + "WHERE thread_id = ? AND agent_id = ? AND user_id = ? AND status = 'pending' " + "ORDER BY created_at DESC LIMIT 1", + (thread_id, agent_id, user_id), + ).fetchone() + if existing is not None and str(existing["questions_json"]) == encoded: + return PendingUserQuestionRow.from_row(existing) + conn.execute( + "UPDATE pending_user_questions SET status = 'cancelled' " + "WHERE thread_id = ? AND agent_id = ? AND status = 'pending'", + (thread_id, agent_id), + ) + pending_id = secrets.token_hex(6) + created_at = now_ts() + conn.execute( + "INSERT INTO pending_user_questions(" + "pending_id, thread_id, agent_id, user_id, session_key, channel_type, " + "questions_json, status, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, 'pending', ?)", + ( + pending_id, + thread_id, + agent_id, + user_id, + session_key, + channel_type, + encoded, + created_at, + ), + ) + row = self.get(pending_id) + assert row is not None + return row + + def get(self, pending_id: str) -> PendingUserQuestionRow | None: + with self._db.connect() as conn: + row = conn.execute( + "SELECT * FROM pending_user_questions WHERE pending_id = ?", (pending_id,) + ).fetchone() + return PendingUserQuestionRow.from_row(row) if row is not None else None + + def pending_for_thread( + self, thread_id: str, *, agent_id: str, user_id: int + ) -> PendingUserQuestionRow | None: + with self._db.connect() as conn: + row = conn.execute( + "SELECT * FROM pending_user_questions " + "WHERE thread_id = ? AND agent_id = ? AND user_id = ? AND status = 'pending' " + "ORDER BY created_at DESC LIMIT 1", + (thread_id, agent_id, user_id), + ).fetchone() + return PendingUserQuestionRow.from_row(row) if row is not None else None + + def pending_for_session( + self, session_key: str, *, agent_id: str + ) -> PendingUserQuestionRow | None: + with self._db.connect() as conn: + row = conn.execute( + "SELECT * FROM pending_user_questions " + "WHERE session_key = ? AND agent_id = ? AND status = 'pending' " + "ORDER BY created_at DESC LIMIT 1", + (session_key, agent_id), + ).fetchone() + return PendingUserQuestionRow.from_row(row) if row is not None else None + + def claim(self, pending_id: str, *, agent_id: str, user_id: int) -> bool: + with self._db.transaction() as conn: + cursor = conn.execute( + "UPDATE pending_user_questions SET status = 'resuming' " + "WHERE pending_id = ? AND agent_id = ? AND user_id = ? AND status = 'pending'", + (pending_id, agent_id, user_id), + ) + return bool(cursor.rowcount) + + def release(self, pending_id: str) -> None: + with self._db.transaction() as conn: + conn.execute( + "UPDATE pending_user_questions SET status = 'pending' " + "WHERE pending_id = ? AND status = 'resuming'", + (pending_id,), + ) + + def mark_answered(self, pending_id: str, answers: list[dict[str, Any]]) -> None: + encoded = json.dumps(answers, ensure_ascii=False, separators=(",", ":")) + with self._db.transaction() as conn: + conn.execute( + "UPDATE pending_user_questions " + "SET status = 'answered', answer_json = ?, answered_at = ? " + "WHERE pending_id = ? AND status = 'resuming'", + (encoded, now_ts(), pending_id), + ) + + +__all__ = ["PendingUserQuestionRepo", "PendingUserQuestionRow"] diff --git a/src/octop/infra/db/services.py b/src/octop/infra/db/services.py index 4814d93f..6b4182a1 100644 --- a/src/octop/infra/db/services.py +++ b/src/octop/infra/db/services.py @@ -24,6 +24,7 @@ from octop.infra.db.repos.sso import SsoRepo from octop.infra.db.repos.threads import ThreadRepo from octop.infra.db.repos.usage import UsageRepo +from octop.infra.db.repos.user_questions import PendingUserQuestionRepo from octop.infra.db.repos.users import UserRepo from octop.infra.db.repos.voice_providers import VoiceProviderRepo from octop.infra.utils.paths import PathLayout @@ -53,6 +54,7 @@ class RepoBundle: care_push_repo: CarePushRepo proactive_care_config_repo: ProactiveCareConfigRepo sso_repo: SsoRepo + pending_user_question_repo: PendingUserQuestionRepo @classmethod def from_pool(cls, db: DatabasePool) -> RepoBundle: @@ -78,6 +80,7 @@ def from_pool(cls, db: DatabasePool) -> RepoBundle: care_push_repo=CarePushRepo(db), proactive_care_config_repo=ProactiveCareConfigRepo(db), sso_repo=SsoRepo(db), + pending_user_question_repo=PendingUserQuestionRepo(db), ) @@ -171,6 +174,10 @@ def proactive_care_config_repo(self) -> ProactiveCareConfigRepo: def sso_repo(self) -> SsoRepo: return self.repos.sso_repo + @property + def pending_user_question_repo(self) -> PendingUserQuestionRepo: + return self.repos.pending_user_question_repo + def build_shared_services( *, db: DatabasePool, paths: PathLayout, config: OctopConfig diff --git a/src/octop/infra/gateway/gateway.py b/src/octop/infra/gateway/gateway.py index 13c0f6f7..65208d68 100644 --- a/src/octop/infra/gateway/gateway.py +++ b/src/octop/infra/gateway/gateway.py @@ -26,6 +26,7 @@ normalize_channel_response_mode, processor_for_response_mode, ) +from octop.infra.gateway.questions import UserQuestionCoordinator from octop.infra.gateway.slash.dispatcher import SlashDispatcher, build_default_dispatcher from octop.infra.gateway.threads import ThreadRegistry from octop.infra.gateway.ws import WS_CHANNEL_ID, WebSocketChannel, WebSocketHub @@ -112,6 +113,7 @@ def __init__( self._ws_channel: WebSocketChannel | None = None self._cli_channel: CliChannel | None = None self._runtime_status: dict[str, ChannelRuntimeStatus] = {} + self._questions = UserQuestionCoordinator(repos.pending_user_question_repo) def replace_repos(self, repos: RepoBundle) -> None: """Point channel/thread persistence at a rebound control-plane pool.""" @@ -120,6 +122,9 @@ def replace_repos(self, repos: RepoBundle) -> None: session_repo=repos.session_repo, thread_repo=repos.thread_repo, ) + self._questions = UserQuestionCoordinator(repos.pending_user_question_repo) + if self._processor is not None: + self._processor.replace_question_coordinator(self._questions) @property def ws_hub(self) -> WebSocketHub: @@ -151,6 +156,10 @@ def processor(self) -> GlobalProcessor: raise RuntimeError("gateway not booted") return self._processor + @property + def question_coordinator(self) -> UserQuestionCoordinator: + return self._questions + @property def slash_meta(self) -> SlashRuntimeMeta | None: return self._slash_meta @@ -196,6 +205,7 @@ async def boot(self) -> None: dispatcher=self._dispatcher, usage_repo=self._repos.usage_repo, gateway=self, + questions=self._questions, ) self._channel_manager = ChannelManager(channels={}) diff --git a/src/octop/infra/gateway/hitl/coordinator.py b/src/octop/infra/gateway/hitl/coordinator.py index b5db90be..cb52903d 100644 --- a/src/octop/infra/gateway/hitl/coordinator.py +++ b/src/octop/infra/gateway/hitl/coordinator.py @@ -105,6 +105,7 @@ async def iter_slash_resolution( locale: str, usage_tracker: UsageTracker | None = None, outcome: HitlSlashOutcome | None = None, + question_coordinator: Any | None = None, ) -> AsyncIterator[MessageEvent]: lang = normalize_locale(locale) if cmd.name == "pending": @@ -160,6 +161,7 @@ async def iter_slash_resolution( projection_state=projection_state, hitl_coordinator=self, hitl_ctx=hitl_ctx, + question_coordinator=question_coordinator, ): if record is not None and not ack_sent: ack = ( diff --git a/src/octop/infra/gateway/process/processor.py b/src/octop/infra/gateway/process/processor.py index 96318aa7..5d387e50 100644 --- a/src/octop/infra/gateway/process/processor.py +++ b/src/octop/infra/gateway/process/processor.py @@ -17,6 +17,7 @@ TextContent, ) +from octop.i18n import tr from octop.i18n.domains.stream import format_stream_error from octop.infra.agents.providers.reasoning import reasoning_request_parameters from octop.infra.gateway.hitl.coordinator import ( @@ -49,6 +50,11 @@ project_stream, ) from octop.infra.gateway.process.usage_record import UsageTracker, record_turn_usage +from octop.infra.gateway.questions.coordinator import ( + QuestionResumeOutcome, + UserQuestionCoordinator, + is_user_question_request, +) from octop.infra.gateway.slash.ctx import SlashCtx, build_slash_ctx from octop.infra.gateway.slash.runner import try_handle_slash from octop.infra.knowledge.default_open import merge_knowledge_base_ids @@ -104,6 +110,7 @@ def __init__( usage_repo: Any | None = None, gateway: Any | None = None, hitl: HitlChannelCoordinator | None = None, + questions: UserQuestionCoordinator | None = None, ) -> None: self._agent_manager = agent_manager self._thread_registry = thread_registry @@ -124,11 +131,21 @@ def __init__( self._usage_repo = usage_repo self._gateway = gateway self._hitl = hitl or HitlChannelCoordinator() + self._questions = questions @property def hitl_coordinator(self) -> HitlChannelCoordinator: return self._hitl + @property + def question_coordinator(self) -> UserQuestionCoordinator: + if self._questions is None: + raise RuntimeError("user-question coordinator is unavailable") + return self._questions + + def replace_question_coordinator(self, coordinator: UserQuestionCoordinator) -> None: + self._questions = coordinator + # -- TeamProcessor (harness inbox async peer collaboration) ---------------- def compose_followup( @@ -359,6 +376,7 @@ async def __call__(self, msg: InboundMessage) -> AsyncIterator[MessageEvent]: locale=locale, usage_tracker=usage_tracker, outcome=slash_outcome, + question_coordinator=self._questions, ): yield ev if slash_outcome.completed_turn: @@ -384,6 +402,56 @@ async def __call__(self, msg: InboundMessage) -> AsyncIterator[MessageEvent]: yield MessageEvent.completed() return + if cmd is not None and cmd.name == "answer": + if self._questions is None: + yield MessageEvent.text(tr("slash.questions.none_pending", locale)) + yield MessageEvent.completed() + return + bound_thread_id = self._thread_registry.get_bound_thread_id(session_key) + record = ( + self._questions.repo.pending_for_thread( + bound_thread_id, + agent_id=agent_id, + user_id=user_id, + ) + if bound_thread_id is not None + else None + ) + answer_text = cmd.args.strip() + if record is not None and answer_text.startswith(record.pending_id): + answer_text = answer_text[len(record.pending_id) :].strip() + if record is None or record.user_id != user_id or record.thread_id != bound_thread_id: + yield MessageEvent.text(tr("slash.questions.none_pending", locale)) + yield MessageEvent.completed() + return + usage_tracker = UsageTracker() + outcome = QuestionResumeOutcome() + try: + async for event in self._questions.iter_channel_answer( + record=record, + answer_text=answer_text, + agent_manager=self._agent_manager, + hitl_coordinator=self._hitl, + locale=locale, + usage_tracker=usage_tracker, + outcome=outcome, + ): + yield event + except Exception as exc: + yield MessageEvent.error_event( + tr("slash.questions.resume_failed", locale, error=str(exc)) + ) + if outcome.completed_turn: + self._touch_thread_after_turn(record.thread_id, msg.text) + self._record_turn_usage( + agent_id=agent_id, + user_id=user_id, + thread_id=record.thread_id, + usage=usage_tracker.usage, + ) + yield MessageEvent.completed() + return + if cmd is not None: sink = _MessageEventSink() handled = await self._dispatcher.handle( @@ -403,6 +471,50 @@ async def __call__(self, msg: InboundMessage) -> AsyncIterator[MessageEvent]: yield MessageEvent.completed() return + if self._questions is not None and msg.text and not msg.content[1:]: + bound_thread_id = self._thread_registry.get_bound_thread_id(session_key) + record = ( + self._questions.auto_answer_for_thread( + thread_id=bound_thread_id, + agent_id=agent_id, + user_id=user_id, + ) + if bound_thread_id is not None + else None + ) + if ( + record is not None + and record.user_id == user_id + and record.thread_id == bound_thread_id + ): + usage_tracker = UsageTracker() + outcome = QuestionResumeOutcome() + try: + async for event in self._questions.iter_channel_answer( + record=record, + answer_text=msg.text, + agent_manager=self._agent_manager, + hitl_coordinator=self._hitl, + locale=locale, + usage_tracker=usage_tracker, + outcome=outcome, + ): + yield event + except Exception as exc: + yield MessageEvent.error_event( + tr("slash.questions.resume_failed", locale, error=str(exc)) + ) + if outcome.completed_turn: + self._touch_thread_after_turn(record.thread_id, msg.text) + self._record_turn_usage( + agent_id=agent_id, + user_id=user_id, + thread_id=record.thread_id, + usage=usage_tracker.usage, + ) + yield MessageEvent.completed() + return + thread_id = await self._thread_registry.get_or_create_by_key( session_key=session_key, agent_id=agent_id, @@ -492,6 +604,7 @@ async def __call__(self, msg: InboundMessage) -> AsyncIterator[MessageEvent]: session_key=session_key, channel_type=channel_type, ), + question_coordinator=self._questions, ): yield ev stream_ok = True @@ -600,16 +713,33 @@ async def iter_turn_chunks(self, msg: InboundMessage) -> AsyncIterator[dict[str, if isinstance(request_payload, dict): from octop.infra.gateway.hitl.coordinator import HitlStreamContext - self._hitl.register_from_request( - request_payload, - ctx=HitlStreamContext( + if ( + is_user_question_request(request_payload) + and self._questions is not None + ): + record = self._questions.register_from_request( + request_payload, thread_id=thread_id, agent_id=agent_id, user_id=user_id, session_key=session_key, channel_type=channel_type, - ), - ) + ) + chunk = { + **chunk, + "request": {**request_payload, "pending_id": record.pending_id}, + } + else: + self._hitl.register_from_request( + request_payload, + ctx=HitlStreamContext( + thread_id=thread_id, + agent_id=agent_id, + user_id=user_id, + session_key=session_key, + channel_type=channel_type, + ), + ) if chunk.get("type") == "tool_result": if harness_workspace is not None: chunk = await enrich_tool_result_with_backend( diff --git a/src/octop/infra/gateway/process/stream_project.py b/src/octop/infra/gateway/process/stream_project.py index 131c8299..308df9d4 100644 --- a/src/octop/infra/gateway/process/stream_project.py +++ b/src/octop/infra/gateway/process/stream_project.py @@ -19,6 +19,7 @@ if TYPE_CHECKING: from octop.infra.agents.manager import AgentManager from octop.infra.gateway.hitl.coordinator import HitlChannelCoordinator, HitlStreamContext + from octop.infra.gateway.questions.coordinator import UserQuestionCoordinator @dataclass @@ -101,6 +102,7 @@ async def _project_chunks( projection_state: StreamProjectionState | None, hitl_coordinator: HitlChannelCoordinator | None, hitl_ctx: HitlStreamContext | None, + question_coordinator: UserQuestionCoordinator | None, ) -> AsyncIterator[MessageEvent]: loc = normalize_locale(str(locale)) tool_state = _ToolProjectionState() @@ -190,16 +192,36 @@ def _tool_end(raw: str) -> MessageEvent: request = chunk.get("request") if not isinstance(request, dict): request = {} - if hitl_coordinator is not None and hitl_ctx is not None: - record = hitl_coordinator.register_from_request(request, ctx=hitl_ctx) + from octop.infra.gateway.questions.coordinator import is_user_question_request + + if ( + is_user_question_request(request) + and question_coordinator is not None + and hitl_ctx is not None + ): + question_record = question_coordinator.register_from_request( + request, + thread_id=hitl_ctx.thread_id, + agent_id=hitl_ctx.agent_id, + user_id=hitl_ctx.user_id, + session_key=hitl_ctx.session_key, + channel_type=hitl_ctx.channel_type, + ) + if projection_state is not None: + projection_state.hitl_paused = True + yield MessageEvent.text( + question_coordinator.channel_card(question_record, str(loc)) + ) + elif hitl_coordinator is not None and hitl_ctx is not None: + hitl_record = hitl_coordinator.register_from_request(request, ctx=hitl_ctx) card = format_hitl_card( - record.action_requests, - pending_id=record.pending_id, + hitl_record.action_requests, + pending_id=hitl_record.pending_id, locale=loc, ) if projection_state is not None: projection_state.hitl_paused = True - projection_state.hitl_pending_id = record.pending_id + projection_state.hitl_pending_id = hitl_record.pending_id yield MessageEvent.text(card) return @@ -220,6 +242,7 @@ async def project_stream( projection_state: StreamProjectionState | None = None, hitl_coordinator: HitlChannelCoordinator | None = None, hitl_ctx: HitlStreamContext | None = None, + question_coordinator: UserQuestionCoordinator | None = None, ) -> AsyncIterator[MessageEvent]: del media_backend # IM tool media uses agent.backend directly async for ev in _project_chunks( @@ -231,6 +254,7 @@ async def project_stream( projection_state=projection_state, hitl_coordinator=hitl_coordinator, hitl_ctx=hitl_ctx, + question_coordinator=question_coordinator, ): yield ev @@ -246,6 +270,7 @@ async def project_resume_stream( projection_state: StreamProjectionState | None = None, hitl_coordinator: HitlChannelCoordinator | None = None, hitl_ctx: HitlStreamContext | None = None, + question_coordinator: UserQuestionCoordinator | None = None, ) -> AsyncIterator[MessageEvent]: async for ev in _project_chunks( agent_manager.resume_hitl(agent_id, thread_id, decisions), @@ -256,5 +281,6 @@ async def project_resume_stream( projection_state=projection_state, hitl_coordinator=hitl_coordinator, hitl_ctx=hitl_ctx, + question_coordinator=question_coordinator, ): yield ev diff --git a/src/octop/infra/gateway/questions/__init__.py b/src/octop/infra/gateway/questions/__init__.py new file mode 100644 index 00000000..5fb62829 --- /dev/null +++ b/src/octop/infra/gateway/questions/__init__.py @@ -0,0 +1,5 @@ +"""Human-question pause/resume support.""" + +from octop.infra.gateway.questions.coordinator import UserQuestionCoordinator + +__all__ = ["UserQuestionCoordinator"] diff --git a/src/octop/infra/gateway/questions/coordinator.py b/src/octop/infra/gateway/questions/coordinator.py new file mode 100644 index 00000000..a3c7d6a7 --- /dev/null +++ b/src/octop/infra/gateway/questions/coordinator.py @@ -0,0 +1,275 @@ +"""Persist and resume structured questions raised by ``ask_user_question``.""" + +from __future__ import annotations + +from collections.abc import AsyncIterator +from dataclasses import dataclass +from typing import Any + +from harness_gateway.models import MessageEvent + +from octop.i18n import tr +from octop.infra.db.repos.user_questions import ( + PendingUserQuestionRepo, + PendingUserQuestionRow, +) +from octop.infra.gateway.process.usage_record import UsageTracker +from octop.infra.utils.locale import normalize_locale + + +def is_user_question_request(request: Any) -> bool: + return isinstance(request, dict) and request.get("kind") == "ask_user_question" + + +def _questions_from_request(request: dict[str, Any]) -> list[dict[str, Any]]: + raw = request.get("questions") + if not isinstance(raw, list) or not raw: + raise ValueError("ask_user_question requires at least one question") + questions: list[dict[str, Any]] = [] + seen: set[str] = set() + for item in raw: + if not isinstance(item, dict): + raise ValueError("ask_user_question contains an invalid question") + question_id = item.get("id") + text = item.get("question") + if not isinstance(question_id, str) or not question_id or question_id in seen: + raise ValueError("ask_user_question contains an invalid or duplicate id") + if not isinstance(text, str) or not text.strip(): + raise ValueError("ask_user_question contains an empty question") + options_raw = item.get("options") + options = [dict(option) for option in options_raw or [] if isinstance(option, dict)] + questions.append( + { + "id": question_id, + "question": text.strip(), + **({"header": item["header"]} if isinstance(item.get("header"), str) else {}), + "options": options, + "multi_select": item.get("multi_select") is True, + } + ) + seen.add(question_id) + return questions + + +def _format_channel_card(record: PendingUserQuestionRow, locale: str) -> str: + lines = [tr("slash.questions.card_title", locale)] + for index, question in enumerate(record.questions, start=1): + lines.append(f"{index}. **{question['question']}**") + options = question.get("options") + if isinstance(options, list): + for option_index, option in enumerate(options, start=1): + if isinstance(option, dict) and isinstance(option.get("label"), str): + label = option["label"] + description = option.get("description") + suffix = f" — {description}" if isinstance(description, str) else "" + lines.append(f" {option_index}) {label}{suffix}") + if len(record.questions) == 1: + lines.append(tr("slash.questions.card_footer_single", locale)) + else: + lines.append(tr("slash.questions.card_footer_multiple", locale)) + lines.append(tr("slash.questions.card_pending_id", locale, pending_id=record.pending_id)) + return "\n".join(lines) + + +def _single_answer(question: dict[str, Any], text: str) -> dict[str, Any]: + value = text.strip() + if not value: + raise ValueError("answer is empty") + options = question.get("options") + labels = [ + str(option["label"]) + for option in options or [] + if isinstance(option, dict) and isinstance(option.get("label"), str) + ] + selected: list[str] = [] + custom: str | None = None + if labels: + pieces = [part.strip() for part in value.split(",") if part.strip()] + for piece in pieces: + if piece.isdigit() and 1 <= int(piece) <= len(labels): + label = labels[int(piece) - 1] + else: + label = next((item for item in labels if item.casefold() == piece.casefold()), "") + if label and label not in selected: + selected.append(label) + elif not label: + custom = value + selected = [] if question.get("multi_select") is not True else selected + break + if question.get("multi_select") is not True and len(selected) > 1: + raise ValueError("single-select question accepts only one option") + else: + custom = value + answer: dict[str, Any] = {"id": question["id"], "selected": selected} + if custom: + answer["custom"] = custom + return answer + + +def parse_channel_answers(record: PendingUserQuestionRow, text: str) -> list[dict[str, Any]]: + """Parse `/answer` text; batches use `1=value; 2=value`.""" + if len(record.questions) == 1: + return [_single_answer(record.questions[0], text)] + parts = [part.strip() for part in text.split(";") if part.strip()] + values: dict[str, str] = {} + for part in parts: + key, separator, value = part.partition("=") + if not separator: + raise ValueError("multiple questions require `1=answer; 2=answer`") + values[key.strip()] = value.strip() + answers: list[dict[str, Any]] = [] + for index, question in enumerate(record.questions, start=1): + answer_value = values.get(str(index)) or values.get(str(question["id"])) + if answer_value is None: + raise ValueError(f"missing answer for question {index}") + answers.append(_single_answer(question, answer_value)) + return answers + + +def validate_answers( + record: PendingUserQuestionRow, raw_answers: list[dict[str, Any]] +) -> list[dict[str, Any]]: + """Validate a structured Dashboard response against the persisted request.""" + by_id = {str(item.get("id")): item for item in raw_answers if isinstance(item, dict)} + if len(by_id) != len(record.questions): + raise ValueError("one answer is required for every question") + normalized: list[dict[str, Any]] = [] + for question in record.questions: + question_id = str(question["id"]) + raw = by_id.get(question_id) + if raw is None: + raise ValueError(f"missing answer for {question_id}") + selected_raw = raw.get("selected") + selected = ( + [str(item) for item in selected_raw if isinstance(item, str)] + if isinstance(selected_raw, list) + else [] + ) + labels = { + str(option["label"]) + for option in question.get("options") or [] + if isinstance(option, dict) and isinstance(option.get("label"), str) + } + if any(label not in labels for label in selected): + raise ValueError(f"unknown option for {question_id}") + if question.get("multi_select") is not True and len(selected) > 1: + raise ValueError(f"{question_id} accepts only one option") + custom_raw = raw.get("custom") + custom = custom_raw.strip() if isinstance(custom_raw, str) else "" + item: dict[str, Any] = {"id": question_id, "selected": selected} + if custom: + item["custom"] = custom + normalized.append(item) + return normalized + + +@dataclass +class QuestionResumeOutcome: + completed_turn: bool = False + + +class UserQuestionCoordinator: + def __init__(self, repo: PendingUserQuestionRepo) -> None: + self.repo = repo + self.repo.recover_interrupted_resumes() + + def register_from_request( + self, + request: dict[str, Any], + *, + thread_id: str, + agent_id: str, + user_id: int, + session_key: str, + channel_type: str, + ) -> PendingUserQuestionRow: + return self.repo.register( + thread_id=thread_id, + agent_id=agent_id, + user_id=user_id, + session_key=session_key, + channel_type=channel_type, + questions=_questions_from_request(request), + ) + + def pending_payload( + self, *, thread_id: str, agent_id: str, user_id: int + ) -> dict[str, Any] | None: + row = self.repo.pending_for_thread(thread_id, agent_id=agent_id, user_id=user_id) + if row is None: + return None + return {"pending_id": row.pending_id, "questions": row.questions} + + def channel_card(self, record: PendingUserQuestionRow, locale: str) -> str: + return _format_channel_card(record, normalize_locale(locale)) + + def auto_answer_for_thread( + self, *, thread_id: str, agent_id: str, user_id: int + ) -> PendingUserQuestionRow | None: + record = self.repo.pending_for_thread(thread_id, agent_id=agent_id, user_id=user_id) + if record is None or len(record.questions) != 1: + return None + return record if not record.questions[0].get("options") else None + + async def iter_channel_answer( + self, + *, + record: PendingUserQuestionRow, + answer_text: str, + agent_manager: Any, + hitl_coordinator: Any, + locale: str, + usage_tracker: UsageTracker, + outcome: QuestionResumeOutcome, + ) -> AsyncIterator[MessageEvent]: + lang = normalize_locale(locale) + try: + answers = parse_channel_answers(record, answer_text) + except ValueError as exc: + yield MessageEvent.text(tr("slash.questions.invalid_answer", lang, error=str(exc))) + return + if not self.repo.claim(record.pending_id, agent_id=record.agent_id, user_id=record.user_id): + yield MessageEvent.text(tr("slash.questions.none_pending", lang)) + return + yield MessageEvent.text(tr("slash.questions.answer_ack", lang)) + try: + from octop.infra.gateway.hitl.coordinator import HitlStreamContext + from octop.infra.gateway.process.stream_project import ( + StreamProjectionState, + project_resume_stream, + ) + + projection = StreamProjectionState() + async for event in project_resume_stream( + agent_manager, + record.agent_id, + record.thread_id, + [{"type": "answer", "answers": answers}], + usage_tracker=usage_tracker, + locale=lang, + projection_state=projection, + hitl_coordinator=hitl_coordinator, + hitl_ctx=HitlStreamContext( + thread_id=record.thread_id, + agent_id=record.agent_id, + user_id=record.user_id, + session_key=record.session_key, + channel_type=record.channel_type, + ), + question_coordinator=self, + ): + yield event + self.repo.mark_answered(record.pending_id, answers) + outcome.completed_turn = True + except Exception: + self.repo.release(record.pending_id) + raise + + +__all__ = [ + "QuestionResumeOutcome", + "UserQuestionCoordinator", + "is_user_question_request", + "parse_channel_answers", + "validate_answers", +] diff --git a/src/octop/infra/gateway/slash/catalog.py b/src/octop/infra/gateway/slash/catalog.py index 67a1f2ec..33eda1b8 100644 --- a/src/octop/infra/gateway/slash/catalog.py +++ b/src/octop/infra/gateway/slash/catalog.py @@ -174,6 +174,14 @@ def label_for(self, locale: str) -> str: category="session", origins=frozenset({"im", "cli"}), ), + SlashCommandSpec( + name="answer", + usage="/answer [pending_id] ", + icon="MessageCircleQuestion", + tone="blue", + category="session", + origins=frozenset({"im", "cli"}), + ), SlashCommandSpec( name="title", usage="/title ", diff --git a/src/octop/infra/gateway/slash/handlers/hitl.py b/src/octop/infra/gateway/slash/handlers/hitl.py index 6f2a4cf5..d49a13f1 100644 --- a/src/octop/infra/gateway/slash/handlers/hitl.py +++ b/src/octop/infra/gateway/slash/handlers/hitl.py @@ -25,4 +25,5 @@ async def _dashboard_hint( "approve": _dashboard_hint, "reject": _dashboard_hint, "pending": _dashboard_hint, + "answer": _dashboard_hint, } diff --git a/tests/unit/agents/test_agent_manager.py b/tests/unit/agents/test_agent_manager.py index 07c60952..7d95e4f6 100644 --- a/tests/unit/agents/test_agent_manager.py +++ b/tests/unit/agents/test_agent_manager.py @@ -207,6 +207,7 @@ def test_build_harness_config_includes_cronjob_tools_when_cron_manager_set( "cronjob_delete", "cronjob_run_now", "search_knowledge", + "ask_user_question", } @@ -217,7 +218,7 @@ def test_build_harness_config_includes_search_knowledge_without_cron( cfg = manager._build_harness_config(_row(agent_id="AGT001")) assert cfg.tools is not None - assert {t.name for t in cfg.tools} == {"search_knowledge"} + assert {t.name for t in cfg.tools} == {"search_knowledge", "ask_user_question"} assert any(isinstance(item, KnowledgeSearchHintMiddleware) for item in (cfg.middleware or [])) diff --git a/tests/unit/agents/test_ask_user_question.py b/tests/unit/agents/test_ask_user_question.py new file mode 100644 index 00000000..e92ba7bf --- /dev/null +++ b/tests/unit/agents/test_ask_user_question.py @@ -0,0 +1,62 @@ +from __future__ import annotations + +import json +from typing import Any + +import pytest + +from octop.infra.agents.ask_user_question import build_ask_user_question_tool + + +@pytest.mark.asyncio +async def test_ask_user_question_interrupts_and_returns_compact_answers( + monkeypatch: pytest.MonkeyPatch, +) -> None: + captured: dict[str, Any] = {} + + def fake_interrupt(request: dict[str, Any]) -> dict[str, Any]: + captured.update(request) + return { + "decisions": [ + { + "type": "answer", + "answers": [{"id": "database", "selected": ["SQLite"]}], + } + ] + } + + monkeypatch.setattr("octop.infra.agents.ask_user_question.interrupt", fake_interrupt) + tool = build_ask_user_question_tool() + result = await tool.ainvoke( + { + "questions": [ + { + "id": "database", + "header": "Storage", + "question": "Which database?", + "options": [ + {"label": "SQLite", "description": "Single-node"}, + {"label": "PostgreSQL"}, + ], + } + ] + } + ) + + assert captured["kind"] == "ask_user_question" + assert captured["questions"][0]["id"] == "database" + assert json.loads(result) == {"answers": [{"id": "database", "selected": ["SQLite"]}]} + + +@pytest.mark.asyncio +async def test_ask_user_question_rejects_duplicate_question_ids() -> None: + tool = build_ask_user_question_tool() + with pytest.raises(ValueError, match="question ids must be unique"): + await tool.ainvoke( + { + "questions": [ + {"id": "same", "question": "First?"}, + {"id": "same", "question": "Second?"}, + ] + } + ) diff --git a/tests/unit/db/test_clip_thread_title.py b/tests/unit/db/test_clip_thread_title.py index 657bb1a7..8c4d0775 100644 --- a/tests/unit/db/test_clip_thread_title.py +++ b/tests/unit/db/test_clip_thread_title.py @@ -82,7 +82,7 @@ def test_migration_003_repairs_stored_hard_cuts(tmp_path: Path) -> None: with pool.connect() as conn: v = conn.execute("SELECT version FROM _schema_version").fetchone()[0] title = conn.execute("SELECT title FROM threads WHERE thread_id = ?", ("t1",)).fetchone()[0] - assert v == 6 + assert v == 7 assert title == "x" * 39 + "…" # Idempotent repair assert repair_all_legacy_thread_titles(pool) == 0 diff --git a/tests/unit/db/test_db_pool.py b/tests/unit/db/test_db_pool.py index 76353bce..d12c7b18 100644 --- a/tests/unit/db/test_db_pool.py +++ b/tests/unit/db/test_db_pool.py @@ -65,6 +65,7 @@ def test_run_migrations_creates_tables(db: SqlitePool): "knowledge_documents", "sso_providers", "sso_login_states", + "pending_user_questions", } assert expected.issubset(names) @@ -76,7 +77,7 @@ def test_run_migrations_idempotent(db: SqlitePool): cols = {r["name"] for r in conn.execute("PRAGMA table_info(users)").fetchall()} cron_cols = {r["name"] for r in conn.execute("PRAGMA table_info(cron_jobs)").fetchall()} thread_cols = {r["name"] for r in conn.execute("PRAGMA table_info(threads)").fetchall()} - assert v == 6 + assert v == 7 assert "login_failed_count" in cols assert "login_locked_until" in cols assert "preferences_json" in cols @@ -126,7 +127,7 @@ def test_migration_002_idempotent_when_column_already_present(tmp_path: Path) -> with pool.connect() as conn: v = conn.execute("SELECT version FROM _schema_version").fetchone()[0] cron_cols = {r["name"] for r in conn.execute("PRAGMA table_info(cron_jobs)").fetchall()} - assert v == 6 + assert v == 7 assert "mcp_servers" in cron_cols assert "skill_packages" in { r["name"] @@ -263,7 +264,7 @@ def test_stuck_version_6_without_permissions_column_is_repaired(tmp_path: Path) with pool.connect() as conn: cols = {r["name"] for r in conn.execute("PRAGMA table_info(users)").fetchall()} version = conn.execute("SELECT version FROM _schema_version").fetchone()[0] - assert version == 6 + assert version == 7 assert "permissions" in cols @@ -300,7 +301,7 @@ def test_pre_squash_schema_version_clamped_and_knowledge_tables_filled( for row in conn.execute("SELECT name FROM sqlite_master WHERE type='table'").fetchall() } user_cols = {r["name"] for r in conn.execute("PRAGMA table_info(users)").fetchall()} - assert version == 6 + assert version == 7 assert "permissions" in user_cols assert { "published_experts", diff --git a/tests/unit/db/test_published_experts_repo.py b/tests/unit/db/test_published_experts_repo.py index fc5a6927..2de1f8b4 100644 --- a/tests/unit/db/test_published_experts_repo.py +++ b/tests/unit/db/test_published_experts_repo.py @@ -26,7 +26,7 @@ def test_published_experts_table_exists(db: SqlitePool) -> None: } v = conn.execute("SELECT version FROM _schema_version").fetchone()[0] assert "published_experts" in names - assert v == 6 + assert v == 7 def test_published_expert_repo_create_get_list_delete(db: SqlitePool) -> None: diff --git a/tests/unit/db/test_repo_knowledge.py b/tests/unit/db/test_repo_knowledge.py index 4b34a14b..cd22f107 100644 --- a/tests/unit/db/test_repo_knowledge.py +++ b/tests/unit/db/test_repo_knowledge.py @@ -51,7 +51,7 @@ def test_knowledge_tables_migrated(db: SqlitePool) -> None: "knowledge_base_members", "knowledge_documents", }.issubset(names) - assert v == 6 + assert v == 7 def test_path_layout_knowledge_dir(tmp_path: Path) -> None: diff --git a/tests/unit/db/test_skill_package_icons.py b/tests/unit/db/test_skill_package_icons.py index e08a3b85..aabdd30f 100644 --- a/tests/unit/db/test_skill_package_icons.py +++ b/tests/unit/db/test_skill_package_icons.py @@ -91,7 +91,7 @@ def test_migration_002_idempotent_when_icon_columns_already_present(tmp_path: Pa "SELECT name FROM sqlite_master WHERE type='index' AND tbl_name='skill_packages'" ) } - assert v == 6 + assert v == 7 assert "icon_name" in cols assert "icon_url" in cols assert "idx_skill_packages_name" in indexes @@ -109,6 +109,6 @@ def test_repair_legacy_schema_adds_icon_columns_at_version_2(tmp_path: Path) -> with pool.connect() as conn: v = conn.execute("SELECT version FROM _schema_version").fetchone()[0] cols = {r["name"] for r in conn.execute("PRAGMA table_info(skill_packages)").fetchall()} - assert v == 6 + assert v == 7 assert "icon_name" in cols assert "icon_url" in cols diff --git a/tests/unit/db/test_skill_packages_repo.py b/tests/unit/db/test_skill_packages_repo.py index e20e6139..cd47fcde 100644 --- a/tests/unit/db/test_skill_packages_repo.py +++ b/tests/unit/db/test_skill_packages_repo.py @@ -25,7 +25,7 @@ def test_skill_packages_table_exists(db: SqlitePool) -> None: } v = conn.execute("SELECT version FROM _schema_version").fetchone()[0] assert "skill_packages" in names - assert v == 6 + assert v == 7 def test_skill_package_repo_create_get(db: SqlitePool) -> None: diff --git a/tests/unit/gateway/test_dashboard_ws.py b/tests/unit/gateway/test_dashboard_ws.py index 602f43eb..f3b012e7 100644 --- a/tests/unit/gateway/test_dashboard_ws.py +++ b/tests/unit/gateway/test_dashboard_ws.py @@ -4,6 +4,7 @@ import base64 from collections.abc import AsyncIterator +from pathlib import Path from typing import Any import pytest @@ -264,6 +265,62 @@ async def _stream(*_args: object, **_kwargs: object): assert pending.action_requests[0]["name"] == "execute" +@pytest.mark.asyncio +async def test_global_processor_persists_and_enriches_user_question(tmp_path: Path) -> None: + from unittest.mock import AsyncMock, MagicMock + + from octop.infra.db.migrate import run_migrations + from octop.infra.db.pool import SqlitePool + from octop.infra.db.repos.user_questions import PendingUserQuestionRepo + from octop.infra.gateway.process.processor import GlobalProcessor + from octop.infra.gateway.questions import UserQuestionCoordinator + from octop.infra.gateway.slash.dispatcher import SlashDispatcher + + async def _stream(*_args: object, **_kwargs: object): + yield { + "type": "hitl_required", + "request": { + "kind": "ask_user_question", + "questions": [{"id": "name", "question": "What name?", "options": []}], + }, + } + + pool = SqlitePool(tmp_path / "octop.db") + run_migrations(pool) + questions = UserQuestionCoordinator(PendingUserQuestionRepo(pool)) + agent_manager = MagicMock() + agent_manager.stream = _stream + agent_manager.merge_turn_mcp_servers = MagicMock(return_value=None) + agent_manager.prepare_chat_mcp = AsyncMock(return_value=[]) + thread_registry = MagicMock() + thread_registry.get_or_create_by_key = AsyncMock(return_value="thread-question") + processor = GlobalProcessor( + agent_manager=agent_manager, + thread_registry=thread_registry, + audit_repo=MagicMock(), + agent_repo=MagicMock(), + user_repo=MagicMock(), + connector_repo=MagicMock(), + dispatcher=SlashDispatcher(), + questions=questions, + ) + msg = InboundMessage( + channel_id=WS_CHANNEL_ID, + channel_type="dashboard", + tenant_id="agent-1", + channel_subject=ChannelSubject(subject_id="1"), + content=[TextContent(text="start")], + metadata={"session_key": "sk", "thread_id": "thread-question"}, + ) + + chunks = [chunk async for chunk in processor.iter_turn_chunks(msg)] + request = next(chunk["request"] for chunk in chunks if chunk.get("type") == "hitl_required") + assert request["pending_id"] + pending = questions.pending_payload(thread_id="thread-question", agent_id="agent-1", user_id=1) + assert pending is not None + assert pending["pending_id"] == request["pending_id"] + + @pytest.mark.asyncio async def test_global_processor_iter_turn_chunks_slash() -> None: from unittest.mock import AsyncMock, MagicMock diff --git a/tests/unit/gateway/test_user_questions.py b/tests/unit/gateway/test_user_questions.py new file mode 100644 index 00000000..6fd3007b --- /dev/null +++ b/tests/unit/gateway/test_user_questions.py @@ -0,0 +1,134 @@ +from __future__ import annotations + +from pathlib import Path +from unittest.mock import MagicMock + +import pytest +from harness_gateway.models import MessageEventType, TextContent + +from octop.infra.db.migrate import run_migrations +from octop.infra.db.pool import SqlitePool +from octop.infra.db.repos.user_questions import PendingUserQuestionRepo +from octop.infra.gateway.hitl.coordinator import HitlChannelCoordinator, HitlStreamContext +from octop.infra.gateway.process.stream_project import StreamProjectionState, project_stream +from octop.infra.gateway.questions.coordinator import ( + UserQuestionCoordinator, + parse_channel_answers, + validate_answers, +) + + +def _repo(tmp_path: Path) -> PendingUserQuestionRepo: + pool = SqlitePool(tmp_path / "octop.db") + run_migrations(pool) + return PendingUserQuestionRepo(pool) + + +def _record(tmp_path: Path): + repo = _repo(tmp_path) + coordinator = UserQuestionCoordinator(repo) + record = coordinator.register_from_request( + { + "kind": "ask_user_question", + "questions": [ + { + "id": "database", + "question": "Which database?", + "options": [{"label": "SQLite"}, {"label": "PostgreSQL"}], + } + ], + }, + thread_id="thread-1", + agent_id="agent-1", + user_id=7, + session_key="session-1", + channel_type="wechat", + ) + return repo, coordinator, record + + +def test_pending_question_survives_coordinator_recreation(tmp_path: Path) -> None: + repo, _coordinator, record = _record(tmp_path) + assert repo.claim(record.pending_id, agent_id="agent-1", user_id=7) + + recreated = UserQuestionCoordinator(repo) + recovered = recreated.pending_payload(thread_id="thread-1", agent_id="agent-1", user_id=7) + + assert recovered is not None + assert recovered["pending_id"] == record.pending_id + assert recovered["questions"][0]["id"] == "database" + + +def test_pending_question_claim_is_exactly_once(tmp_path: Path) -> None: + repo, _coordinator, record = _record(tmp_path) + assert repo.claim(record.pending_id, agent_id="agent-1", user_id=7) + assert not repo.claim(record.pending_id, agent_id="agent-1", user_id=7) + + +def test_channel_answer_accepts_option_number(tmp_path: Path) -> None: + _repo_value, _coordinator, record = _record(tmp_path) + assert parse_channel_answers(record, "2") == [{"id": "database", "selected": ["PostgreSQL"]}] + + +def test_dashboard_answer_rejects_unknown_option(tmp_path: Path) -> None: + _repo_value, _coordinator, record = _record(tmp_path) + try: + validate_answers(record, [{"id": "database", "selected": ["MongoDB"]}]) + except ValueError as exc: + assert "unknown option" in str(exc) + else: # pragma: no cover - assertion guard + raise AssertionError("unknown option was accepted") + + +@pytest.mark.asyncio +async def test_channel_projection_persists_and_renders_question(tmp_path: Path) -> None: + async def _stream(*_args: object, **_kwargs: object): + yield { + "type": "hitl_required", + "request": { + "kind": "ask_user_question", + "questions": [ + { + "id": "database", + "question": "Which database?", + "options": [{"label": "SQLite"}, {"label": "PostgreSQL"}], + } + ], + }, + } + + repo = _repo(tmp_path) + questions = UserQuestionCoordinator(repo) + agent_manager = MagicMock() + agent_manager.stream = _stream + projection = StreamProjectionState() + + events = [ + event + async for event in project_stream( + agent_manager, + "agent-1", + {"thread_id": "thread-1", "messages": []}, + projection_state=projection, + hitl_coordinator=HitlChannelCoordinator(), + hitl_ctx=HitlStreamContext( + thread_id="thread-1", + agent_id="agent-1", + user_id=7, + session_key="session-1", + channel_type="wechat", + ), + question_coordinator=questions, + ) + ] + + assert projection.hitl_paused is True + assert len(events) == 1 + assert events[0].type == MessageEventType.MESSAGE + content = events[0].content[0] + assert isinstance(content, TextContent) + assert "Which database?" in content.text + assert "/answer" in content.text + assert ( + questions.pending_payload(thread_id="thread-1", agent_id="agent-1", user_id=7) is not None + ) diff --git a/tests/unit/i18n/test_tools.py b/tests/unit/i18n/test_tools.py index bd57bd7d..58ab092b 100644 --- a/tests/unit/i18n/test_tools.py +++ b/tests/unit/i18n/test_tools.py @@ -32,6 +32,7 @@ def test_hitl_tool_catalog_excludes_must_use_tools(): assert "unknown" not in names assert "current_time" not in names assert "write_todos" not in names + assert "ask_user_question" not in names assert "memory_search" not in names assert "search_knowledge" not in names assert "cronjob_create" not in names