diff --git a/dashboard/src/api/modules/octopThreads.ts b/dashboard/src/api/modules/octopThreads.ts index 8d031552..e474f62d 100644 --- a/dashboard/src/api/modules/octopThreads.ts +++ b/dashboard/src/api/modules/octopThreads.ts @@ -1,6 +1,36 @@ import type { HitlPendingPayload } from "../types/hitl"; import { request } from "../request"; +export type ThreadTaskItemStatus = + | "pending" + | "in_progress" + | "completed" + | "cancelled"; + +export interface ThreadTaskItem { + id: string; + content: string; + status: ThreadTaskItemStatus; +} + +export interface ThreadTaskState { + thread_id: string; + available: boolean; + status: "idle" | "active" | "completed"; + items: ThreadTaskItem[]; + completed: number; + total: number; +} + +export interface ThreadTaskSummary extends ThreadTaskState { + agent_id: string; + title: string | null; + last_active: number; + created_at: number; + turn_active: boolean; + turn_started_at: number | null; +} + export interface OctopThread { thread_id: string; title: string | null; @@ -39,6 +69,8 @@ export interface OctopThreadHistory { /** Pending tool approval for this thread (survives page reload). */ hitl_pending?: HitlPendingPayload | null; artifacts?: string[]; + /** Authoritative DeepAgents todo projection from the thread checkpoint. */ + task_state?: ThreadTaskState; } export interface OctopThreadPatch { @@ -97,6 +129,32 @@ export const octopThreadsApi = { ); }, + taskState: (agentId: string, threadId: string) => + request( + `/agents/${encodeURIComponent(agentId)}/threads/${encodeURIComponent( + threadId, + )}/task-state`, + ), + + tasks: ( + agentId: string, + status: "active" | "completed" | "all" = "all", + limit = 50, + ) => + request( + `/agents/${encodeURIComponent( + agentId, + )}/thread-tasks?status=${status}&limit=${limit}`, + ), + + cancelTurn: (agentId: string, threadId: string) => + request<{ thread_id: string; cancelled: boolean }>( + `/agents/${encodeURIComponent(agentId)}/threads/${encodeURIComponent( + threadId, + )}/cancel`, + { method: "POST" }, + ), + contextUsage: ( agentId: string, threadId: string, diff --git a/dashboard/src/components/BrowserAiPanel.tsx b/dashboard/src/components/BrowserAiPanel.tsx index 9ab87afc..beb4b942 100644 --- a/dashboard/src/components/BrowserAiPanel.tsx +++ b/dashboard/src/components/BrowserAiPanel.tsx @@ -131,6 +131,7 @@ export default function BrowserAiPanel({ bootError, messages, isStreaming, + taskPlan, send, cancelStream, } = useAgentThreadChat(agentId); @@ -636,6 +637,7 @@ export default function BrowserAiPanel({ ) : ( -
-
- {t("chatUsage.todoListTitle", "Task progress")} + {showHeader ? ( +
+
+ {t("chatUsage.todoListTitle", "Task progress")} +
+
+ {t("chatUsage.todoListSummary", { + completed, + total: items.length, + defaultValue: "{{completed}}/{{total}} done", + })} +
-
- {t("chatUsage.todoListSummary", { - completed, - total: items.length, - defaultValue: "{{completed}}/{{total}} done", - })} -
-
+ ) : null}
    {items.map((item) => (
  • @@ -102,14 +109,62 @@ export function TodoProgressPanel({ isStreaming?: boolean; followingProcessSummary?: boolean; }) { + const { t } = useTranslation(); + const [collapsed, setCollapsed] = useState(false); + const completed = countCompletedTodos(items); + const current = + items.find((item) => item.status === "in_progress") || + items.find((item) => item.status === "pending"); + const percent = items.length + ? Math.round((completed / items.length) * 100) + : 0; return ( - +
    + +
    + +
    + {!collapsed ? ( + + ) : null} +
    ); } diff --git a/dashboard/src/hooks/useAgentThreadChat.ts b/dashboard/src/hooks/useAgentThreadChat.ts index 3dd00bd1..56981622 100644 --- a/dashboard/src/hooks/useAgentThreadChat.ts +++ b/dashboard/src/hooks/useAgentThreadChat.ts @@ -15,8 +15,14 @@ export function useAgentThreadChat(agentId: string | null) { const [booting, setBooting] = useState(false); const [bootError, setBootError] = useState(null); - const { messages, isStreaming, sendMessage, cancelStream, loadHistory } = - useChat(threadId, agentId); + const { + messages, + isStreaming, + taskPlan, + sendMessage, + cancelStream, + loadHistory, + } = useChat(threadId, agentId); useEffect(() => { if (!agentId) { @@ -83,6 +89,7 @@ export function useAgentThreadChat(agentId: string | null) { bootError, messages, isStreaming, + taskPlan, send, cancelStream, }; diff --git a/dashboard/src/locales/en.json b/dashboard/src/locales/en.json index 6d6ff52a..231568e0 100644 --- a/dashboard/src/locales/en.json +++ b/dashboard/src/locales/en.json @@ -4391,6 +4391,49 @@ } } }, + "taskCenter": { + "tabs": { + "active": "Active", + "scheduled": "Scheduled", + "history": "History" + }, + "noAgent": "Select an expert to view its tasks", + "noActive": "No active thread plans", + "noHistory": "No completed thread plans", + "count": "{{count}} task threads", + "untitled": "Untitled thread", + "activeStatus": "In progress", + "completedStatus": "Completed", + "runningStatus": "Running", + "waitingStatus": "Waiting", + "openThread": "Open conversation", + "continueThread": "Continue", + "viewDetails": "Details", + "stopRun": "Stop", + "activeHeading": "Agent runs", + "historyHeading": "Completed runs", + "liveRefresh": "Live refresh", + "runningNow": "Running now", + "awaitingInput": "Awaiting input", + "stepsCompleted": "Steps completed", + "currentRun": "Current run", + "completedRun": "Completed run", + "awaitingRun": "Awaiting your input", + "nextPlan": "Next plan", + "latestCompleted": "Latest completed", + "otherPlans": "Other plans", + "earlierRuns": "Earlier runs", + "currentStep": "Current step", + "stepProgress": "{{completed}} of {{total}} steps", + "moreSteps": "+{{count}} more steps", + "progress": "Progress", + "elapsed": "Elapsed", + "executionPlan": "Execution plan", + "threadId": "Thread ID", + "cancelConfirmTitle": "Stop this run?", + "cancelConfirmDesc": "The agent's current response and tool execution will be cancelled. The plan remains available to continue later.", + "cancelled": "Run stopped" + }, "pageShell": { "experts": { "title": "Experts", @@ -4398,7 +4441,7 @@ }, "tasks": { "title": "Tasks", - "subtitle": "Scheduled tasks that run on a cron trigger" + "subtitle": "Track active agent plans, scheduled work, and completed threads" }, "connectors": { "title": "Connectors", diff --git a/dashboard/src/locales/zh.json b/dashboard/src/locales/zh.json index 108a4f11..15ad78d2 100644 --- a/dashboard/src/locales/zh.json +++ b/dashboard/src/locales/zh.json @@ -4536,14 +4536,57 @@ } } }, + "taskCenter": { + "tabs": { + "active": "进行中", + "scheduled": "已计划", + "history": "历史" + }, + "noAgent": "请选择一个专家查看任务", + "noActive": "暂无进行中的线程计划", + "noHistory": "暂无已完成的线程计划", + "count": "共 {{count}} 个任务线程", + "untitled": "未命名线程", + "activeStatus": "进行中", + "completedStatus": "已完成", + "runningStatus": "正在运行", + "waitingStatus": "等待继续", + "openThread": "打开对话", + "continueThread": "继续任务", + "viewDetails": "查看详情", + "stopRun": "停止运行", + "activeHeading": "Agent 运行任务", + "historyHeading": "已完成的运行", + "liveRefresh": "实时刷新", + "runningNow": "当前运行", + "awaitingInput": "等待输入", + "stepsCompleted": "已完成步骤", + "currentRun": "当前运行", + "completedRun": "已完成运行", + "awaitingRun": "等待你继续", + "nextPlan": "下一项计划", + "latestCompleted": "最近完成", + "otherPlans": "其他计划", + "earlierRuns": "更早的运行", + "currentStep": "当前步骤", + "stepProgress": "已完成 {{completed}} / {{total}} 步", + "moreSteps": "还有 {{count}} 个步骤", + "progress": "完成进度", + "elapsed": "已运行", + "executionPlan": "执行计划", + "threadId": "线程 ID", + "cancelConfirmTitle": "停止这次运行？", + "cancelConfirmDesc": "Agent 当前的回复和工具执行会被取消，任务计划仍会保留，之后可以继续。", + "cancelled": "运行已停止" + }, "pageShell": { "experts": { "title": "专家", "subtitle": "创建与管理你的 AI 专家，也可从模板库挑选场景一键新建。" }, "tasks": { - "title": "定时任务", - "subtitle": "配置 Agent 定期执行的计划任务" + "title": "任务中心", + "subtitle": "统一查看 Agent 进行中的计划、定时任务和已完成记录" }, "connectors": { "title": "连接器", diff --git a/dashboard/src/pages/Chat/components/AssistantTurnView.tsx b/dashboard/src/pages/Chat/components/AssistantTurnView.tsx index 1e877808..30412723 100644 --- a/dashboard/src/pages/Chat/components/AssistantTurnView.tsx +++ b/dashboard/src/pages/Chat/components/AssistantTurnView.tsx @@ -2,6 +2,7 @@ import { useMemo } from "react"; import { ChevronRight, FilePen, Globe } from "lucide-react"; import { useTranslation } from "react-i18next"; import type { ChatMessage } from "../hooks/useChat"; +import type { ThreadTaskState } from "../../../api/modules/octopThreads"; import { splitAssistantTurn, toAnswerOnlyMessage, @@ -47,6 +48,8 @@ interface AssistantTurnViewProps { shellCommandDisabled?: boolean; shellCommandDisabledTitle?: string; compactProcess?: boolean; + /** Server-side checkpoint projection; preferred over tool-message parsing. */ + taskPlan?: ThreadTaskState | null; } function hasProcessContent( @@ -74,6 +77,7 @@ export default function AssistantTurnView({ shellCommandDisabled, shellCommandDisabledTitle, compactProcess = false, + taskPlan, }: AssistantTurnViewProps) { const { t } = useTranslation(); const { activeAgentId } = useAgent(); @@ -128,15 +132,18 @@ export default function AssistantTurnView({ toolMedia.videos.length > 0 || toolMedia.files.length > 0; - const todoItems = useMemo( + const parsedTodoItems = useMemo( () => collectWriteTodosFromMessages(messages), [messages], ); + const todoItems = taskPlan?.available ? taskPlan.items : parsedTodoItems; const todoStreaming = turnStreaming && - messages.some( - (m) => m.status === "streaming" && isWriteTodosToolName(m.toolData?.name), - ); + (taskPlan?.items.some((item) => item.status === "in_progress") || + messages.some( + (m) => + m.status === "streaming" && isWriteTodosToolName(m.toolData?.name), + )); const firstProcessSegmentIdx = compactProcess ? -1 diff --git a/dashboard/src/pages/Chat/components/MessageList.tsx b/dashboard/src/pages/Chat/components/MessageList.tsx index 750dc493..694bf22a 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 { ThreadTaskState } from "../../../api/modules/octopThreads"; import type { ComposerTagLookups } from "./UserMessageComposerTags"; import MessageBubble from "./MessageBubble"; import AssistantTurnView from "./AssistantTurnView"; @@ -109,6 +110,7 @@ interface MessageListProps { shellCommandDisabled?: boolean; shellCommandDisabledTitle?: string; compactProcess?: boolean; + taskPlan?: ThreadTaskState | null; } interface GroupRenderContext { @@ -133,6 +135,7 @@ interface GroupRenderContext { shellCommandDisabled?: boolean; shellCommandDisabledTitle?: string; compactProcess?: boolean; + taskPlan?: ThreadTaskState | null; registerBubbleRef: (messageId: string, el: HTMLDivElement | null) => void; } @@ -178,6 +181,11 @@ function renderMessageGroup( shellCommandDisabled={ctx.shellCommandDisabled} shellCommandDisabledTitle={ctx.shellCommandDisabledTitle} compactProcess={ctx.compactProcess} + taskPlan={ + groupIndex === ctx.lastAssistantGroupIndex + ? ctx.taskPlan + : undefined + } />
); @@ -226,6 +234,9 @@ function renderMessageGroup( shellCommandDisabled={ctx.shellCommandDisabled} shellCommandDisabledTitle={ctx.shellCommandDisabledTitle} compactProcess={ctx.compactProcess} + taskPlan={ + groupIndex === ctx.lastAssistantGroupIndex ? ctx.taskPlan : undefined + } /> ); @@ -259,6 +270,7 @@ export default function MessageList(props: MessageListProps) { shellCommandDisabled, shellCommandDisabledTitle, compactProcess, + taskPlan, } = props; const { t } = useTranslation(); @@ -687,6 +699,7 @@ export default function MessageList(props: MessageListProps) { shellCommandDisabled, shellCommandDisabledTitle, compactProcess, + taskPlan, registerBubbleRef, }), [ @@ -709,6 +722,7 @@ export default function MessageList(props: MessageListProps) { shellCommandDisabled, shellCommandDisabledTitle, compactProcess, + taskPlan, registerBubbleRef, ], ); diff --git a/dashboard/src/pages/Chat/hooks/chatStore.ts b/dashboard/src/pages/Chat/hooks/chatStore.ts index 49aa4538..74447a4c 100644 --- a/dashboard/src/pages/Chat/hooks/chatStore.ts +++ b/dashboard/src/pages/Chat/hooks/chatStore.ts @@ -204,6 +204,7 @@ const EMPTY_SNAPSHOT: SessionSnapshot = Object.freeze({ thinkingStartedAt: null, runUsage: null, contextUsage: null, + taskPlan: null, historyHasMore: false, historyLoadingMore: false, historyNextOffset: 0, @@ -330,6 +331,7 @@ function buildSnapshot(state: SessionStreamState): SessionSnapshot { thinkingStartedAt: state.thinkingStartedAt, runUsage: state.runUsage, contextUsage: state.contextUsage, + taskPlan: state.taskPlan, historyHasMore: state.historyHasMore, historyLoadingMore: state.historyLoadingMore, historyNextOffset: state.historyNextOffset, @@ -346,6 +348,7 @@ function getOrCreate(sessionId: string): SessionStreamState { thinkingStartedAt: null, runUsage: null, contextUsage: null, + taskPlan: null, abortController: null, streamMsg: "", streamId: "", @@ -514,7 +517,13 @@ export function setMessages(sessionId: string, messages: ChatMessage[]) { export function setHistoryPage( sessionId: string, messages: ChatMessage[], - opts: { hasMore: boolean; nextOffset: number }, + opts: { + hasMore: boolean; + nextOffset: number; + taskPlan?: + | import("../../../api/modules/octopThreads").ThreadTaskState + | null; + }, ) { const state = getOrCreate(sessionId); state.messages = messages; @@ -524,6 +533,7 @@ export function setHistoryPage( state.historyNextOffset = opts.nextOffset; state.historyLoadingMore = false; state.historyHydrated = true; + if (opts.taskPlan !== undefined) state.taskPlan = opts.taskPlan; notify(state); } @@ -642,6 +652,7 @@ export function clearMessages(sessionId: string) { state.historyNextOffset = 0; state.historyLoadingMore = false; state.historyHydrated = false; + state.taskPlan = null; notify(state); } @@ -936,6 +947,9 @@ function handleHarnessChunk( case "tool_result": closeToolCall(state, chunk.messages, sessionId); break; + case "task_plan_updated": + state.taskPlan = chunk.task_state; + break; case "done": finalizeStreamingMessages(state); break; diff --git a/dashboard/src/pages/Chat/hooks/sseHelpers.ts b/dashboard/src/pages/Chat/hooks/sseHelpers.ts index 00351d2f..bdf57c15 100644 --- a/dashboard/src/pages/Chat/hooks/sseHelpers.ts +++ b/dashboard/src/pages/Chat/hooks/sseHelpers.ts @@ -11,6 +11,7 @@ import type { TokenUsage, } from "../../../api/types"; import type { ContentBlockItem } from "../../../utils/messageParser"; +import type { ThreadTaskState } from "../../../api/modules/octopThreads"; export interface ToolCallData { name?: string; @@ -80,6 +81,8 @@ export interface SessionStreamState { runUsage: TokenUsage | null; /** Latest prompt/context token count from SSE state snapshots. */ contextUsage: TokenUsage | null; + /** Authoritative task plan projected from the server-side checkpoint. */ + taskPlan: ThreadTaskState | null; abortController: AbortController | null; /** Running buffer for in-flight ``token`` chunks. */ streamMsg: string; @@ -109,6 +112,7 @@ export interface SessionSnapshot { thinkingStartedAt: number | null; runUsage: TokenUsage | null; contextUsage: TokenUsage | null; + taskPlan: ThreadTaskState | null; historyHasMore: boolean; historyLoadingMore: boolean; historyNextOffset: number; diff --git a/dashboard/src/pages/Chat/hooks/useChat.ts b/dashboard/src/pages/Chat/hooks/useChat.ts index e8fc9287..c3e46dde 100644 --- a/dashboard/src/pages/Chat/hooks/useChat.ts +++ b/dashboard/src/pages/Chat/hooks/useChat.ts @@ -33,6 +33,7 @@ import type { ChatMessage, UserComposerContext, } from "./sseHelpers"; +import type { ThreadTaskState } from "../../../api/modules/octopThreads"; export type { ToolCallData, @@ -656,6 +657,7 @@ async function loadThreadHistory( nextOffset: number; turnActive: boolean; artifacts: string[]; + taskPlan: ThreadTaskState | null; }> { try { const { octopThreadsApi, CHAT_HISTORY_PAGE_SIZE } = await import( @@ -695,6 +697,7 @@ async function loadThreadHistory( nextOffset: offset + limit, turnActive: Boolean(history.turn_active), artifacts, + taskPlan: history.task_state ?? null, }; } catch (err) { console.error("loadThreadHistory failed", err); @@ -704,6 +707,7 @@ async function loadThreadHistory( nextOffset: 0, turnActive: false, artifacts: [], + taskPlan: null, }; } } @@ -748,6 +752,7 @@ export function useChat( thinkingStartedAt, runUsage, contextUsage, + taskPlan, historyHasMore, historyLoadingMore, historyHydrated, @@ -857,11 +862,13 @@ export function useChat( hasMore, nextOffset, turnActive, + taskPlan, } = await loadThreadHistory(agentId, targetThreadId, { offset: 0 }); if (loadGenRef.current !== gen) return; chatStore.setHistoryPage(key, converted, { hasMore, nextOffset, + taskPlan, }); if (shouldProbeActiveTurn({ isStreaming: false, turnActive })) { attachAfterHistory(key, targetThreadId); @@ -940,6 +947,7 @@ export function useChat( messages: latest, hasMore, nextOffset, + taskPlan, } = await loadThreadHistory(agentId, key, { offset: 0 }); // Stale after a concurrent loadHistory / newer refresh — drop apply only. if (loadGenRef.current !== gen) return; @@ -954,6 +962,7 @@ export function useChat( hasMore: olderPrefix.length > 0 ? snap.historyHasMore : hasMore, nextOffset: olderPrefix.length > 0 ? snap.historyNextOffset : nextOffset, + taskPlan, }); } finally { refreshInFlightRef.current = false; @@ -1020,6 +1029,7 @@ export function useChat( thinkingStartedAt, runUsage, contextUsage, + taskPlan, historyLoading, historyHasMore, historyLoadingMore, diff --git a/dashboard/src/pages/Chat/index.tsx b/dashboard/src/pages/Chat/index.tsx index 761e8c11..93e05b45 100644 --- a/dashboard/src/pages/Chat/index.tsx +++ b/dashboard/src/pages/Chat/index.tsx @@ -263,6 +263,7 @@ function ChatPageInner() { historyRefreshing, historyHydrated, contextUsage, + taskPlan, sendMessage, editAndResend, cancelStream, @@ -922,6 +923,7 @@ function ChatPageInner() { ) : ( + embedded ? ( + <>{content} + ) : ( - - - + {content} ); + + // Until the user picks an agent there is nothing to fetch and no scope + // to write to. Mirror the behaviour of the other octop agent-scoped pages. + if (!activeAgentId) { + return wrap( + + + , + ); } // Keep list shell stable across expert switches, but do NOT show the @@ -256,12 +264,8 @@ function CronJobsPage() { const showBodySpinner = loading || (contentBusy && !showList && !showEmpty); const showToolbar = showList; - return ( - + return wrap( + <> {showToolbar ? (
@@ -390,7 +394,7 @@ function CronJobsPage() { onCancel={handleExecuteNowCancel} onConfirm={handleExecuteNowConfirm} /> - + , ); } diff --git a/dashboard/src/pages/Control/Terminal/components/AiPanel.tsx b/dashboard/src/pages/Control/Terminal/components/AiPanel.tsx index 5717c761..2f708235 100644 --- a/dashboard/src/pages/Control/Terminal/components/AiPanel.tsx +++ b/dashboard/src/pages/Control/Terminal/components/AiPanel.tsx @@ -170,6 +170,7 @@ export default function AiPanel({ bootError, messages, isStreaming, + taskPlan, send, cancelStream, } = useAgentThreadChat(opsAgent?.agent_id ?? null); @@ -422,6 +423,7 @@ export default function AiPanel({ ) : ( ; + if (item.status === "cancelled") return ; + if (item.status === "in_progress" || running) return ; + return ; +} + +function currentItem(task: ThreadTaskSummary): ThreadTaskItem | undefined { + return ( + task.items.find((item) => item.status === "in_progress") || + task.items.find((item) => item.status === "pending") + ); +} + +function elapsedText(startedAt: number | null, now: number): string { + if (!startedAt) return ""; + const seconds = Math.max(0, Math.floor(now / 1000) - startedAt); + const hours = Math.floor(seconds / 3600); + const minutes = Math.floor((seconds % 3600) / 60); + const rest = seconds % 60; + if (hours > 0) return `${hours}h ${minutes}m`; + if (minutes > 0) return `${minutes}m ${rest}s`; + return `${rest}s`; +} + +function LiveElapsed({ startedAt }: { startedAt: number | null }) { + const [now, setNow] = useState(Date.now()); + useEffect(() => { + if (!startedAt) return; + const timer = window.setInterval(() => setNow(Date.now()), 1000); + return () => window.clearInterval(timer); + }, [startedAt]); + return <>{elapsedText(startedAt, now)}; +} + +export function TaskTimeline({ + task, + limit, +}: { + task: ThreadTaskSummary; + limit?: number; +}) { + const { t } = useTranslation(); + const visibleItems = limit ? task.items.slice(0, limit) : task.items; + const active = currentItem(task); + return ( +
    + {visibleItems.map((item) => { + const isCurrent = item.id === active?.id; + return ( +
  1. + + {itemIcon(item, isCurrent && task.turn_active)} + + {item.content} + + {isCurrent && task.turn_active + ? t("taskCenter.runningStatus") + : t(`chatUsage.todoStatus.${item.status}`, item.status)} + +
    2. + ); + })} + {limit && task.items.length > limit ? ( +
  2. + {t("taskCenter.moreSteps", { count: task.items.length - limit })} +
    3. + ) : null} +
+ ); +} + +export function TaskRunCard({ + task, + featured = false, + onDetails, + onOpenThread, + onCancel, +}: { + task: ThreadTaskSummary; + featured?: boolean; + onDetails: () => void; + onOpenThread: () => void; + onCancel: () => void; +}) { + const { t } = useTranslation(); + const active = currentItem(task); + const percent = task.total + ? Math.round((task.completed / task.total) * 100) + : 0; + const isRunning = task.turn_active; + const title = task.title || t("taskCenter.untitled"); + + return ( +
+
{ + if (event.key === "Enter" || event.key === " ") { + event.preventDefault(); + onDetails(); + } + }} + > +
+
+ +
+
+ {isRunning + ? t("taskCenter.currentRun") + : task.status === "completed" + ? t("taskCenter.completedRun") + : t("taskCenter.awaitingRun")} +
+

{title}

+
+
+ + {isRunning + ? t("taskCenter.runningStatus") + : task.status === "completed" + ? t("taskCenter.completedStatus") + : t("taskCenter.waitingStatus")} + +
+ +
+ + + {t("taskCenter.stepProgress", { + completed: task.completed, + total: task.total, + })} + + {isRunning && task.turn_started_at ? ( + + + + + ) : ( + + + {formatServerDateTime(task.last_active || task.created_at)} + + )} +
+ + + + {active && task.status === "active" ? ( +
+ {t("taskCenter.currentStep")} + {active.content} +
+ ) : null} + +
+ +
+ + {isRunning ? ( + + ) : null} + +
+
+ ); +} + +export function TaskDetailDrawer({ + task, + open, + onClose, + onOpenThread, + onCancel, +}: { + task: ThreadTaskSummary | null; + open: boolean; + onClose: () => void; + onOpenThread: () => void; + onCancel: () => void; +}) { + const { t } = useTranslation(); + const percent = useMemo( + () => (task?.total ? Math.round((task.completed / task.total) * 100) : 0), + [task], + ); + return ( + + {task.turn_active + ? t("taskCenter.runningStatus") + : task.status === "completed" + ? t("taskCenter.completedStatus") + : t("taskCenter.waitingStatus")} + + ) : null + } + footer={ + task ? ( +
+ {task.turn_active ? ( + + ) : ( + + )} + +
+ ) : null + } + > + {task ? ( +
+
+
+ {t("taskCenter.progress")} + {percent}% +
+ +

+ {t("taskCenter.stepProgress", { + completed: task.completed, + total: task.total, + })} + {task.turn_active && task.turn_started_at ? ( + <> + {" · "} + {t("taskCenter.elapsed")}{" "} + + + ) : null} +

+
+
+ {t("taskCenter.executionPlan")} +
+ +
+ {t("taskCenter.threadId")} + {task.thread_id} +
+
+ ) : null} + + ); +} diff --git a/dashboard/src/pages/Tasks/index.module.less b/dashboard/src/pages/Tasks/index.module.less new file mode 100644 index 00000000..97775231 --- /dev/null +++ b/dashboard/src/pages/Tasks/index.module.less @@ -0,0 +1,484 @@ +.loading { + display: grid; + min-height: 260px; + place-items: center; +} + +.threadTasks { + display: flex; + flex-direction: column; + gap: 22px; + max-width: 1180px; +} + +.toolbar { + display: flex; + align-items: center; + justify-content: space-between; + gap: 16px; +} + +.toolbar > div:first-child { + display: flex; + flex-direction: column; + gap: 2px; +} + +.toolbar strong { + color: var(--fn-text-primary, #172033); + font-size: 15px; +} + +.toolbar span { + color: var(--fn-text-tertiary, #8c95a5); + font-size: 12px; +} + +.toolbarActions, +.liveRefreshToggle { + display: flex; + align-items: center; + gap: 10px; +} + +.liveRefreshToggle { + padding: 5px 10px; + border: 1px solid var(--fn-border-secondary, rgba(0, 0, 0, 0.08)); + border-radius: 999px; + background: var(--fn-bg-secondary, rgba(0, 0, 0, 0.018)); + color: var(--fn-text-secondary, #667085); + cursor: pointer; + font-size: 12px; +} + +.runStats { + display: grid; + grid-template-columns: repeat(3, minmax(0, 1fr)); + overflow: hidden; + border: 1px solid var(--fn-border-secondary, rgba(0, 0, 0, 0.07)); + border-radius: 14px; + background: var(--fn-bg-elevated, #fff); +} + +.runStats > div { + display: grid; + grid-template-columns: auto 1fr auto; + gap: 9px; + align-items: center; + padding: 14px 16px; + color: var(--fn-text-secondary, #667085); +} + +.runStats > div + div { + border-left: 1px solid var(--fn-border-secondary, rgba(0, 0, 0, 0.07)); +} + +.runStats svg { + color: #1677ff; +} + +.runStats span { + font-size: 12px; +} + +.runStats strong { + color: var(--fn-text-primary, #172033); + font-size: 18px; + font-variant-numeric: tabular-nums; +} + +.sectionLabel { + margin-bottom: 9px; + color: var(--fn-text-tertiary, #8c95a5); + font-size: 11px; + font-weight: 700; + letter-spacing: 0.08em; + text-transform: uppercase; +} + +.cardGrid { + display: grid; + grid-template-columns: repeat(auto-fit, minmax(min(100%, 380px), 1fr)); + gap: 14px; +} + +.runCard { + overflow: hidden; + border: 1px solid var(--fn-border-secondary, rgba(0, 0, 0, 0.08)); + border-radius: 16px; + background: var(--fn-bg-elevated, #fff); + box-shadow: 0 2px 10px rgba(16, 24, 40, 0.035); + transition: + border-color 0.18s ease, + box-shadow 0.18s ease, + transform 0.18s ease; +} + +.runCard:hover { + border-color: rgba(22, 119, 255, 0.28); + box-shadow: 0 8px 26px rgba(16, 24, 40, 0.08); + transform: translateY(-1px); +} + +.runCardFeatured { + border-color: rgba(22, 119, 255, 0.22); + background: radial-gradient( + circle at 100% 0, + rgba(22, 119, 255, 0.09), + transparent 34% + ), + var(--fn-bg-elevated, #fff); + box-shadow: 0 8px 30px rgba(22, 119, 255, 0.08); +} + +.runCardMain { + display: block; + width: 100%; + padding: 18px 20px 16px; + border: 0; + background: transparent; + color: inherit; + cursor: pointer; + text-align: left; +} + +.runCardFeatured .runCardMain { + padding: 22px 24px 18px; +} + +.runCardHeader, +.runIdentity, +.runMeta, +.runActions, +.drawerFooter { + display: flex; + align-items: center; +} + +.runCardHeader { + justify-content: space-between; + gap: 16px; +} + +.runIdentity { + min-width: 0; + gap: 11px; +} + +.liveDot { + position: relative; + flex: 0 0 auto; + width: 9px; + height: 9px; + border-radius: 50%; +} + +.liveDotRunning { + background: #1677ff; + box-shadow: 0 0 0 5px rgba(22, 119, 255, 0.1); +} + +.liveDotRunning::after { + position: absolute; + inset: -5px; + border: 1px solid rgba(22, 119, 255, 0.45); + border-radius: 50%; + animation: taskPulse 1.8s ease-out infinite; + content: ""; +} + +.liveDotWaiting { + background: #98a2b3; + box-shadow: 0 0 0 5px rgba(152, 162, 179, 0.12); +} + +@keyframes taskPulse { + from { + opacity: 0.7; + transform: scale(0.7); + } + to { + opacity: 0; + transform: scale(1.35); + } +} + +.runEyebrow { + margin-bottom: 2px; + color: var(--fn-text-tertiary, #8c95a5); + font-size: 10px; + font-weight: 700; + letter-spacing: 0.08em; + text-transform: uppercase; +} + +.runTitle { + overflow: hidden; + margin: 0; + color: var(--fn-text-primary, #172033); + font-size: 16px; + font-weight: 650; + line-height: 1.4; + text-overflow: ellipsis; + white-space: nowrap; +} + +.runCardFeatured .runTitle { + font-size: 19px; +} + +.runMeta { + gap: 18px; + margin-top: 14px; + color: var(--fn-text-tertiary, #8c95a5); + font-size: 11px; +} + +.runMeta span { + display: inline-flex; + gap: 6px; + align-items: center; +} + +.runProgress { + margin: 10px 0 12px; +} + +.currentStep { + display: grid; + grid-template-columns: auto minmax(0, 1fr); + gap: 9px; + align-items: baseline; + margin: 2px 0 12px; + padding: 9px 11px; + border-radius: 9px; + background: rgba(22, 119, 255, 0.06); +} + +.currentStep span { + color: #1677ff; + font-size: 10px; + font-weight: 700; + letter-spacing: 0.04em; + text-transform: uppercase; +} + +.currentStep strong { + overflow: hidden; + color: var(--fn-text-primary, #172033); + font-size: 12px; + font-weight: 550; + text-overflow: ellipsis; + white-space: nowrap; +} + +.timeline { + display: flex; + flex-direction: column; + gap: 0; + margin: 0; + padding: 0; + list-style: none; +} + +.timelineItem { + position: relative; + display: grid; + grid-template-columns: 20px minmax(0, 1fr) auto; + gap: 8px; + min-height: 34px; + color: var(--fn-text-secondary, #667085); +} + +.timelineItem:not(:last-child)::before { + position: absolute; + top: 18px; + bottom: -2px; + left: 7px; + width: 1px; + background: var(--fn-border-secondary, rgba(0, 0, 0, 0.09)); + content: ""; +} + +.timelineMarker { + z-index: 1; + display: inline-flex; + align-items: flex-start; + justify-content: center; + padding-top: 3px; + color: #98a2b3; + background: var(--fn-bg-elevated, #fff); +} + +.runCardFeatured .timelineMarker { + background: transparent; +} + +.timelineItemCurrent .timelineMarker, +.timelineItemCurrent .timelineStatus { + color: #1677ff; +} + +.timelineItemCurrent .timelineContent { + color: var(--fn-text-primary, #172033); + font-weight: 550; +} + +.timelineItemDone .timelineMarker { + color: #52c41a; +} + +.timelineItemDone .timelineContent { + color: var(--fn-text-tertiary, #98a2b3); + text-decoration: line-through; +} + +.timelineContent { + overflow: hidden; + padding-bottom: 10px; + font-size: 12px; + line-height: 1.5; + text-overflow: ellipsis; + white-space: nowrap; +} + +.timelineStatus { + padding-top: 1px; + color: var(--fn-text-tertiary, #98a2b3); + font-size: 10px; + white-space: nowrap; +} + +.timelineMore { + padding: 1px 0 2px 28px; + color: var(--fn-text-tertiary, #98a2b3); + font-size: 11px; +} + +.runActions { + justify-content: flex-end; + gap: 7px; + padding: 10px 14px; + border-top: 1px solid var(--fn-border-secondary, rgba(0, 0, 0, 0.065)); + background: var(--fn-bg-secondary, rgba(0, 0, 0, 0.012)); +} + +.emptyRunState { + display: grid; + min-height: 300px; + border: 1px dashed var(--fn-border-secondary, rgba(0, 0, 0, 0.12)); + border-radius: 16px; + background: var(--fn-bg-secondary, rgba(0, 0, 0, 0.012)); + place-items: center; +} + +.drawerBody { + display: flex; + flex-direction: column; + gap: 22px; +} + +.drawerSummary { + padding: 16px; + border: 1px solid var(--fn-border-secondary, rgba(0, 0, 0, 0.07)); + border-radius: 12px; + background: var(--fn-bg-secondary, rgba(0, 0, 0, 0.018)); +} + +.drawerSummary > div:first-child { + display: flex; + align-items: center; + justify-content: space-between; + margin-bottom: 8px; +} + +.drawerSummary span, +.drawerSummary p { + color: var(--fn-text-tertiary, #8c95a5); + font-size: 12px; +} + +.drawerSummary strong { + color: var(--fn-text-primary, #172033); + font-size: 20px; +} + +.drawerSummary p { + margin: 6px 0 0; +} + +.drawerSectionTitle { + color: var(--fn-text-primary, #172033); + font-size: 13px; + font-weight: 650; +} + +.drawerThreadMeta { + display: flex; + align-items: center; + justify-content: space-between; + gap: 12px; + padding-top: 16px; + border-top: 1px solid var(--fn-border-secondary, rgba(0, 0, 0, 0.07)); + color: var(--fn-text-tertiary, #8c95a5); + font-size: 11px; +} + +.drawerThreadMeta code { + overflow: hidden; + padding: 3px 7px; + border-radius: 6px; + background: var(--fn-bg-secondary, rgba(0, 0, 0, 0.035)); + color: var(--fn-text-secondary, #667085); + text-overflow: ellipsis; +} + +.drawerFooter { + justify-content: space-between; +} + +@media (max-width: 720px) { + .toolbar { + align-items: flex-start; + flex-direction: column; + } + + .toolbarActions { + justify-content: space-between; + width: 100%; + } + + .runStats { + grid-template-columns: 1fr; + } + + .runStats > div + div { + border-top: 1px solid var(--fn-border-secondary, rgba(0, 0, 0, 0.07)); + border-left: 0; + } + + .runCardMain, + .runCardFeatured .runCardMain { + padding: 16px; + } + + .runCardHeader { + align-items: flex-start; + } + + .runActions { + flex-wrap: wrap; + } + + .runActions :global(.ant-btn-text) { + display: none; + } + + .timelineStatus { + display: none; + } + + .timelineItem { + grid-template-columns: 20px minmax(0, 1fr); + } +} diff --git a/dashboard/src/pages/Tasks/index.tsx b/dashboard/src/pages/Tasks/index.tsx new file mode 100644 index 00000000..d872207a --- /dev/null +++ b/dashboard/src/pages/Tasks/index.tsx @@ -0,0 +1,277 @@ +import { useCallback, useEffect, useMemo, useRef, useState } from "react"; +import { Button, Card, Empty, Modal, Spin, Switch, Tabs, message } from "antd"; +import { History, RefreshCw, Sparkles, TimerReset } from "lucide-react"; +import { useTranslation } from "react-i18next"; +import { useNavigate } from "react-router-dom"; +import { + octopThreadsApi, + type ThreadTaskSummary, +} from "../../api/modules/octopThreads"; +import { useAgent } from "../../context/AgentContext"; +import PageShell from "../../layouts/PageShell"; +import CronJobsPage from "../Control/CronJobs"; +import { TaskDetailDrawer, TaskRunCard } from "./TaskRunCard"; +import styles from "./index.module.less"; + +function ThreadTasksPanel({ status }: { status: "active" | "completed" }) { + const { t } = useTranslation(); + const navigate = useNavigate(); + const { activeAgentId } = useAgent(); + const [tasks, setTasks] = useState([]); + const [loading, setLoading] = useState(false); + const [refreshing, setRefreshing] = useState(false); + const [autoRefresh, setAutoRefresh] = useState(status === "active"); + const [selectedTask, setSelectedTask] = useState( + null, + ); + const requestGeneration = useRef(0); + + const load = useCallback( + async (quiet = false) => { + const generation = ++requestGeneration.current; + if (!activeAgentId) { + setTasks([]); + return; + } + if (quiet) setRefreshing(true); + else setLoading(true); + try { + const rows = await octopThreadsApi.tasks(activeAgentId, status, 20); + if (generation !== requestGeneration.current) return; + const sorted = [...rows].sort( + (left, right) => + Number(right.turn_active) - Number(left.turn_active) || + right.last_active - left.last_active, + ); + setTasks(sorted); + setSelectedTask((selected) => + selected + ? sorted.find((row) => row.thread_id === selected.thread_id) || null + : null, + ); + } catch { + if (generation === requestGeneration.current && !quiet) setTasks([]); + } finally { + if (generation === requestGeneration.current) { + setLoading(false); + setRefreshing(false); + } + } + }, + [activeAgentId, status], + ); + + useEffect(() => { + void load(); + return () => { + requestGeneration.current += 1; + }; + }, [load]); + + useEffect(() => { + if (status !== "active" || !autoRefresh || !activeAgentId) return; + const timer = window.setInterval(() => { + if (document.visibilityState === "visible") void load(true); + }, 4000); + return () => window.clearInterval(timer); + }, [activeAgentId, autoRefresh, load, status]); + + const openThread = useCallback( + (task: ThreadTaskSummary) => { + if (!activeAgentId) return; + navigate(`/chat/${activeAgentId}/${task.thread_id}`); + }, + [activeAgentId, navigate], + ); + + const cancelTask = useCallback( + (task: ThreadTaskSummary) => { + if (!activeAgentId || !task.turn_active) return; + Modal.confirm({ + title: t("taskCenter.cancelConfirmTitle"), + content: t("taskCenter.cancelConfirmDesc"), + okText: t("taskCenter.stopRun"), + cancelText: t("common.cancel"), + okButtonProps: { danger: true }, + onOk: async () => { + const result = await octopThreadsApi.cancelTurn( + activeAgentId, + task.thread_id, + ); + if (result.cancelled) message.success(t("taskCenter.cancelled")); + await load(true); + }, + }); + }, + [activeAgentId, load, t], + ); + + const runningCount = tasks.filter((task) => task.turn_active).length; + const waitingCount = tasks.filter((task) => !task.turn_active).length; + const completedSteps = tasks.reduce((sum, task) => sum + task.completed, 0); + const featuredTask = tasks[0]; + const queuedTasks = tasks.slice(1); + + if (!activeAgentId) { + return ( + + + + ); + } + if (loading && tasks.length === 0) { + return ( +
+ +
+ ); + } + + return ( +
+
+
+ + {status === "active" + ? t("taskCenter.activeHeading") + : t("taskCenter.historyHeading")} + + {t("taskCenter.count", { count: tasks.length })} +
+
+ {status === "active" ? ( + + ) : null} + +
+
+ + {status === "active" && tasks.length > 0 ? ( +
+
+ + {t("taskCenter.runningNow")} + {runningCount} +
+
+ + {t("taskCenter.awaitingInput")} + {waitingCount} +
+
+ + {t("taskCenter.stepsCompleted")} + {completedSteps} +
+
+ ) : null} + + {tasks.length === 0 ? ( +
+ +
+ ) : ( + <> + {featuredTask ? ( +
+
+ {status === "active" + ? featuredTask.turn_active + ? t("taskCenter.currentRun") + : t("taskCenter.nextPlan") + : t("taskCenter.latestCompleted")} +
+ setSelectedTask(featuredTask)} + onOpenThread={() => openThread(featuredTask)} + onCancel={() => cancelTask(featuredTask)} + /> +
+ ) : null} + {queuedTasks.length > 0 ? ( +
+
+ {status === "active" + ? t("taskCenter.otherPlans") + : t("taskCenter.earlierRuns")} +
+
+ {queuedTasks.map((task) => ( + setSelectedTask(task)} + onOpenThread={() => openThread(task)} + onCancel={() => cancelTask(task)} + /> + ))} +
+
+ ) : null} + + )} + + setSelectedTask(null)} + onOpenThread={() => selectedTask && openThread(selectedTask)} + onCancel={() => selectedTask && cancelTask(selectedTask)} + /> +
+ ); +} + +export default function TasksPage() { + const { t } = useTranslation(); + const tabs = useMemo( + () => [ + { + key: "active", + label: t("taskCenter.tabs.active"), + children: , + }, + { + key: "scheduled", + label: t("taskCenter.tabs.scheduled"), + children: , + }, + { + key: "history", + label: t("taskCenter.tabs.history"), + children: , + }, + ], + [t], + ); + return ( + + + + ); +} diff --git a/dashboard/src/routes/index.tsx b/dashboard/src/routes/index.tsx index 98e49629..dd2dca11 100644 --- a/dashboard/src/routes/index.tsx +++ b/dashboard/src/routes/index.tsx @@ -3,7 +3,7 @@ import { Navigate, useLocation } from "react-router-dom"; // Lazy-loaded pages — Common const ExpertsPage = lazy(() => import("../pages/Experts")); -const CronJobsPage = lazy(() => import("../pages/Control/CronJobs")); +const TasksPage = lazy(() => import("../pages/Tasks")); const ConnectorsPage = lazy(() => import("../pages/Agent/Connectors")); const ACPPage = lazy(() => import("../pages/Agent/ACP")); const SkillPackagesPage = lazy(() => import("../pages/SkillPackages")); @@ -142,7 +142,7 @@ export const routeConfigs: RouteConfig[] = [ // Common { path: "/experts", element: }, - { path: "/tasks", element: }, + { path: "/tasks", element: }, { path: "/connectors", element: }, { path: "/skill-packages", element: }, { path: "/knowledge-bases", element: }, diff --git a/dashboard/src/routes/prefetch.ts b/dashboard/src/routes/prefetch.ts index 4a9cbb5e..e781c9ba 100644 --- a/dashboard/src/routes/prefetch.ts +++ b/dashboard/src/routes/prefetch.ts @@ -2,7 +2,7 @@ const ROUTE_PREFETCHERS: Record Promise> = { "/chat": () => import("../pages/Chat"), "/experts": () => import("../pages/Experts"), - "/tasks": () => import("../pages/Control/CronJobs"), + "/tasks": () => import("../pages/Tasks"), "/connectors": () => import("../pages/Agent/Connectors"), "/skill-packages": () => import("../pages/SkillPackages"), "/knowledge-bases": () => import("../pages/KnowledgeBases"), diff --git a/dashboard/src/utils/parseHarnessChunk.test.ts b/dashboard/src/utils/parseHarnessChunk.test.ts index 3e4865ec..bf535852 100644 --- a/dashboard/src/utils/parseHarnessChunk.test.ts +++ b/dashboard/src/utils/parseHarnessChunk.test.ts @@ -23,4 +23,19 @@ describe("parseHarnessChunk usage", () => { }, }); }); + + it("parses authoritative task plan updates", () => { + const chunk = parseHarnessChunk( + 'data: {"type":"task_plan_updated","thread_id":"thread-1","task_state":{"thread_id":"thread-1","available":true,"status":"active","items":[{"id":"1","content":"Ship","status":"in_progress"}],"completed":0,"total":1}}', + ); + + expect(chunk).toMatchObject({ + type: "task_plan_updated", + thread_id: "thread-1", + task_state: { + status: "active", + total: 1, + }, + }); + }); }); diff --git a/dashboard/src/utils/parseHarnessChunk.ts b/dashboard/src/utils/parseHarnessChunk.ts index b5c53cc0..5cd1d6dd 100644 --- a/dashboard/src/utils/parseHarnessChunk.ts +++ b/dashboard/src/utils/parseHarnessChunk.ts @@ -9,6 +9,8 @@ * ``parseHarnessChunk`` also accepts legacy ``data: …`` SSE lines. */ +import type { ThreadTaskState } from "../api/modules/octopThreads"; + export interface TokenChunk { type: "token"; /** Graph node that produced the chunk (e.g. ``agent``, ``tool``). */ @@ -97,6 +99,12 @@ export interface AttachmentChunk { filename?: string; } +export interface TaskPlanUpdatedChunk { + type: "task_plan_updated"; + thread_id: string; + task_state: ThreadTaskState; +} + export type HarnessChunk = | TokenChunk | ReasoningChunk @@ -110,7 +118,8 @@ export type HarnessChunk = | ErrorChunk | HitlRequiredChunk | SlashActionChunk - | AttachmentChunk; + | AttachmentChunk + | TaskPlanUpdatedChunk; /** * Parse one ``data: …`` SSE frame line into a typed chunk. @@ -237,6 +246,19 @@ export function parseHarnessChunk(line: string): HarnessChunk | null { kind: typeof obj.kind === "string" ? obj.kind : undefined, filename: typeof obj.filename === "string" ? obj.filename : undefined, }; + case "task_plan_updated": + if ( + !obj.task_state || + typeof obj.task_state !== "object" || + Array.isArray(obj.task_state) + ) { + return { type: "custom", data: raw }; + } + return { + type: "task_plan_updated", + thread_id: typeof obj.thread_id === "string" ? obj.thread_id : "", + task_state: obj.task_state as unknown as ThreadTaskState, + }; default: // Forward-compatible: keep the unrecognized payload around so // a debug toggle can render it instead of dropping it. diff --git a/src/octop/api/routers/chat/history.py b/src/octop/api/routers/chat/history.py index e85f6557..8e8e796e 100644 --- a/src/octop/api/routers/chat/history.py +++ b/src/octop/api/routers/chat/history.py @@ -2,8 +2,9 @@ from __future__ import annotations +import asyncio from pathlib import Path -from typing import Any +from typing import Any, Literal from fastapi import APIRouter, Depends, Request @@ -19,6 +20,7 @@ from octop.infra.agents.context_breakdown import SEGMENT_KEYS, compute_context_breakdown from octop.infra.agents.middleware.thread_artifacts import artifacts_for_response from octop.infra.agents.thread_fork import fork_dashboard_thread +from octop.infra.agents.thread_tasks import read_thread_task_state from octop.infra.agents.workspace_dir import agent_facing_workspace_dir_from_config from octop.infra.errors import ErrorCode, OctopError from octop.infra.gateway.hitl.coordinator import pending_hitl_payload @@ -191,6 +193,9 @@ async def get_thread_history( user_id=effective_uid, ) workspace_dir = _agent_facing_workspace_dir(server, agent_id) + task_state = await read_thread_task_state( + server.app_runtime.agent_registry, agent_id, thread_id + ) return { "thread_id": thread_id, "messages": messages, @@ -200,9 +205,88 @@ async def get_thread_history( "turn_active": server.app_runtime.gateway.ws_hub.is_turn_active(thread_id), "hitl_pending": hitl_pending, "artifacts": artifacts_for_response(row.artifacts, workspace_dir), + "task_state": task_state, } +@router.get( + "/agents/{agent_id}/threads/{thread_id}/task-state", + summary="Current thread task plan", +) +async def get_thread_task_state( + agent_id: str, + thread_id: str, + as_user: int | None = None, + user: Any = Depends(current_user), + server: Any = Depends(get_server), +) -> dict[str, Any]: + """Return the authoritative task projection from the thread checkpoint.""" + _require_thread(server, agent_id, thread_id, user, as_user) + return await read_thread_task_state(server.app_runtime.agent_registry, agent_id, thread_id) + + +@router.get("/agents/{agent_id}/thread-tasks", summary="List thread task plans") +async def list_thread_tasks( + agent_id: str, + status: Literal["active", "completed", "all"] = "all", + limit: int = 50, + as_user: int | None = None, + user: Any = Depends(current_user), + server: Any = Depends(get_server), +) -> list[dict[str, Any]]: + """List non-empty task plans for this user's recent agent threads.""" + require_agent_row(agent_id, user=user, as_user=as_user, server=server) + effective_uid = as_user if as_user is not None else user.id + rows = server.app_runtime.gateway.thread_registry.list_threads( + agent_id=agent_id, + user_id=effective_uid, + limit=max(1, min(limit, 100)), + ) + task_states = await asyncio.gather( + *( + read_thread_task_state(server.app_runtime.agent_registry, agent_id, row.thread_id) + for row in rows + ) + ) + output: list[dict[str, Any]] = [] + for row, task_state in zip(rows, task_states, strict=True): + if task_state["status"] == "idle": + continue + if status != "all" and task_state["status"] != status: + continue + output.append( + { + **task_state, + "agent_id": agent_id, + "title": row.title, + "last_active": row.last_active, + "created_at": row.created_at, + "turn_active": server.app_runtime.gateway.ws_hub.is_turn_active(row.thread_id), + "turn_started_at": server.app_runtime.gateway.ws_hub.turn_started_at(row.thread_id), + } + ) + return output + + +@router.post( + "/agents/{agent_id}/threads/{thread_id}/cancel", + summary="Cancel active thread turn", +) +async def cancel_thread_turn( + agent_id: str, + thread_id: str, + as_user: int | None = None, + user: Any = Depends(current_user), + server: Any = Depends(get_server), +) -> dict[str, Any]: + """Cancel the live harness stream for an owned thread, when one exists.""" + _require_thread(server, agent_id, thread_id, user, as_user) + active = server.app_runtime.gateway.ws_hub.is_turn_active(thread_id) + if active: + server.app_runtime.agent_registry.cancel_stream(agent_id, thread_id) + return {"thread_id": thread_id, "cancelled": active} + + @router.post( "/agents/{agent_id}/threads/{thread_id}/read", status_code=204, diff --git a/src/octop/infra/agents/thread_tasks.py b/src/octop/infra/agents/thread_tasks.py new file mode 100644 index 00000000..8cce4c02 --- /dev/null +++ b/src/octop/infra/agents/thread_tasks.py @@ -0,0 +1,129 @@ +"""Read-only projection of DeepAgents todos from persisted thread state.""" + +from __future__ import annotations + +import logging +from typing import Any, Literal, TypeGuard + +logger = logging.getLogger(__name__) + +TaskItemStatus = Literal["pending", "in_progress", "completed", "cancelled"] +TaskPlanStatus = Literal["idle", "active", "completed"] + + +def _is_task_item_status(value: str) -> TypeGuard[TaskItemStatus]: + return value in {"pending", "in_progress", "completed", "cancelled"} + + +def _normalize_item(raw: Any, index: int) -> dict[str, str] | None: + if isinstance(raw, str): + content = raw.strip() + status: TaskItemStatus = "pending" + elif isinstance(raw, dict): + content = str( + raw.get("content") + or raw.get("text") + or raw.get("title") + or raw.get("description") + or "" + ).strip() + raw_status = str(raw.get("status") or "pending").strip().lower() + aliases = { + "done": "completed", + "complete": "completed", + "running": "in_progress", + "active": "in_progress", + "canceled": "cancelled", + } + normalized = aliases.get(raw_status, raw_status) + status = normalized if _is_task_item_status(normalized) else "pending" + else: + return None + if not content: + return None + item_id = ( + str(raw.get("id") or raw.get("key") or index + 1) + if isinstance(raw, dict) + else str(index + 1) + ) + return {"id": item_id, "content": content, "status": status} + + +def project_thread_task_state( + thread_id: str, + todos: Any, + *, + available: bool = True, +) -> dict[str, Any]: + """Normalize LangGraph ``todos`` into the stable dashboard contract.""" + raw_items = todos if isinstance(todos, list) else [] + items = [ + item + for index, raw in enumerate(raw_items) + if (item := _normalize_item(raw, index)) is not None + ] + completed = sum(item["status"] == "completed" for item in items) + active = any(item["status"] in {"pending", "in_progress"} for item in items) + status: TaskPlanStatus = "active" if active else "completed" if items else "idle" + return { + "thread_id": thread_id, + "available": available, + "status": status, + "items": items, + "completed": completed, + "total": len(items), + } + + +def task_state_from_stream_chunk(thread_id: str, chunk: dict[str, Any]) -> dict[str, Any] | None: + """Project a harness state frame when it explicitly contains todos.""" + if chunk.get("type") not in {"state_update", "state_snapshot"}: + return None + data = chunk.get("data") + if not isinstance(data, dict) or "todos" not in data: + return None + return project_thread_task_state(thread_id, data.get("todos")) + + +async def read_thread_task_state( + agent_manager: Any, + agent_id: str, + thread_id: str, +) -> dict[str, Any]: + """Read the current todo list from the live harness checkpointer. + + Agent shutdown makes the live graph unavailable. That is represented in + the response instead of turning thread history and the task center into a + 500 response. + """ + try: + harness = agent_manager.get_agent(agent_id) + graph = getattr(harness, "graph", None) + aget_state = getattr(graph, "aget_state", None) + if aget_state is None: + return project_thread_task_state(thread_id, [], available=False) + state = await aget_state({"configurable": {"thread_id": thread_id}}) + values = getattr(state, "values", None) + todos = values.get("todos") if isinstance(values, dict) else [] + return project_thread_task_state(thread_id, todos) + except Exception: + logger.warning( + "failed to read task state for agent=%s thread=%s", + agent_id, + thread_id, + exc_info=True, + ) + return project_thread_task_state(thread_id, [], available=False) + + +def task_state_fingerprint(state: dict[str, Any]) -> tuple[Any, ...]: + """Return a compact equality key used to suppress duplicate stream frames.""" + return ( + state.get("available"), + state.get("status"), + tuple( + (item.get("id"), item.get("content"), item.get("status")) + for item in state.get("items", []) + if isinstance(item, dict) + ), + ) diff --git a/src/octop/infra/gateway/process/processor.py b/src/octop/infra/gateway/process/processor.py index 443bf137..d0c49dc7 100644 --- a/src/octop/infra/gateway/process/processor.py +++ b/src/octop/infra/gateway/process/processor.py @@ -19,6 +19,11 @@ from octop.i18n.domains.stream import format_stream_error from octop.infra.agents.providers.reasoning import reasoning_request_parameters +from octop.infra.agents.thread_tasks import ( + read_thread_task_state, + task_state_fingerprint, + task_state_from_stream_chunk, +) from octop.infra.gateway.hitl.coordinator import ( HitlChannelCoordinator, HitlSlashOutcome, @@ -585,6 +590,7 @@ async def iter_turn_chunks(self, msg: InboundMessage) -> AsyncIterator[dict[str, stream_ok = False harness_workspace = harness_workspace_for_agent(self._agent_manager, agent_id) usage_tracker = UsageTracker() + last_task_fingerprint: tuple[Any, ...] | None = None locale = resolve_user_locale( user_repo=self._user_repo, user_id=user_id, @@ -595,6 +601,16 @@ async def iter_turn_chunks(self, msg: InboundMessage) -> AsyncIterator[dict[str, try: async for chunk in self._agent_manager.stream(agent_id, request): usage_tracker.observe(chunk) + task_state = task_state_from_stream_chunk(thread_id, chunk) + if task_state is not None: + fingerprint = task_state_fingerprint(task_state) + if fingerprint != last_task_fingerprint: + last_task_fingerprint = fingerprint + yield { + "type": "task_plan_updated", + "thread_id": thread_id, + "task_state": task_state, + } if chunk.get("type") == "hitl_required": request_payload = chunk.get("request") if isinstance(request_payload, dict): @@ -640,6 +656,16 @@ async def iter_turn_chunks(self, msg: InboundMessage) -> AsyncIterator[dict[str, except Exception as exc: await self._record_stream_error(user_id=user_id, agent_id=agent_id, exc=exc) yield {"type": "error", "message": format_stream_error(exc, locale)} + if stream_ok and last_task_fingerprint is None: + final_task_state = await read_thread_task_state( + self._agent_manager, agent_id, thread_id + ) + if final_task_state.get("available") and final_task_state.get("total", 0) > 0: + yield { + "type": "task_plan_updated", + "thread_id": thread_id, + "task_state": final_task_state, + } if stream_ok: self._touch_thread_after_turn(thread_id, msg.text) self._record_turn_usage( diff --git a/src/octop/infra/gateway/ws/ws_hub.py b/src/octop/infra/gateway/ws/ws_hub.py index b9a276d0..08e2af0a 100644 --- a/src/octop/infra/gateway/ws/ws_hub.py +++ b/src/octop/infra/gateway/ws/ws_hub.py @@ -5,6 +5,7 @@ import asyncio import json import logging +import time from collections.abc import Awaitable, Callable from typing import Any @@ -36,6 +37,7 @@ def __init__(self) -> None: self._thread_subscribers: dict[str, set[str]] = {} self._conn_thread: dict[str, str] = {} self._active_turns: set[str] = set() + self._turn_started_at: dict[str, int] = {} def register(self, connection_id: str, send_fn: SendFn) -> None: self._connections[connection_id] = send_fn @@ -78,13 +80,20 @@ def mark_turn_active(self, thread_id: str) -> None: tid = thread_id.strip() if tid: self._active_turns.add(tid) + self._turn_started_at.setdefault(tid, int(time.time())) def mark_turn_idle(self, thread_id: str) -> None: - self._active_turns.discard(thread_id.strip()) + tid = thread_id.strip() + self._active_turns.discard(tid) + self._turn_started_at.pop(tid, None) def is_turn_active(self, thread_id: str) -> bool: return thread_id.strip() in self._active_turns + def turn_started_at(self, thread_id: str) -> int | None: + """Return epoch seconds for the current in-memory turn, if active.""" + return self._turn_started_at.get(thread_id.strip()) + async def push(self, connection_id: str, frame: dict[str, Any]) -> None: send_fn = self._connections.get(connection_id) if send_fn is None: diff --git a/tests/integration/test_chat_ws.py b/tests/integration/test_chat_ws.py index cac5a6df..5fcf59ea 100644 --- a/tests/integration/test_chat_ws.py +++ b/tests/integration/test_chat_ws.py @@ -6,8 +6,9 @@ import json from collections.abc import AsyncIterator from pathlib import Path +from types import SimpleNamespace from typing import Any -from unittest.mock import MagicMock +from unittest.mock import AsyncMock, MagicMock import httpx import pytest @@ -567,6 +568,48 @@ async def test_thread_history_reports_active_turn(env: Any) -> None: assert active.json()["turn_active"] is True +async def test_thread_task_summary_and_cancel_active_turn(env: Any) -> None: + c, srv, _fake, alice_auth, bob_auth, aid = env + create = await c.post(f"/api/agents/{aid}/threads", headers=alice_auth) + tid = create.json()["thread_id"] + agent = srv.app_runtime.agent_registry.get_agent(aid) + agent.graph.aget_state = AsyncMock( + return_value=SimpleNamespace( + values={ + "todos": [ + {"content": "Inspect", "status": "completed"}, + {"content": "Implement", "status": "in_progress"}, + ] + } + ) + ) + srv.app_runtime.gateway.ws_hub.mark_turn_active(tid) + + listed = await c.get( + f"/api/agents/{aid}/thread-tasks?status=active", + headers=alice_auth, + ) + assert listed.status_code == 200 + row = next(item for item in listed.json() if item["thread_id"] == tid) + assert row["turn_active"] is True + assert isinstance(row["turn_started_at"], int) + + cancel_spy = MagicMock() + srv.app_runtime.agent_registry.cancel_stream = cancel_spy + cancelled = await c.post( + f"/api/agents/{aid}/threads/{tid}/cancel", + headers=alice_auth, + ) + assert cancelled.json() == {"thread_id": tid, "cancelled": True} + cancel_spy.assert_called_once_with(aid, tid) + + denied = await c.post( + f"/api/agents/{aid}/threads/{tid}/cancel", + headers=bob_auth, + ) + assert denied.status_code in {403, 404} + + async def test_create_thread(env: Any) -> None: c, _srv, _fake, alice_auth, _bob_auth, aid = env r = await c.post(f"/api/agents/{aid}/threads", headers=alice_auth) diff --git a/tests/unit/agents/test_thread_tasks.py b/tests/unit/agents/test_thread_tasks.py new file mode 100644 index 00000000..6ec3aa82 --- /dev/null +++ b/tests/unit/agents/test_thread_tasks.py @@ -0,0 +1,85 @@ +"""Tests for the persisted thread task projection.""" + +from __future__ import annotations + +from types import SimpleNamespace +from typing import Any + +import pytest + +from octop.infra.agents.thread_tasks import ( + project_thread_task_state, + read_thread_task_state, + task_state_from_stream_chunk, +) + + +def test_project_thread_task_state_normalizes_items_and_status() -> None: + state = project_thread_task_state( + "thread-1", + [ + {"content": "Inspect", "status": "done"}, + {"id": "build", "content": "Build", "status": "running"}, + {"content": " "}, + ], + ) + + assert state == { + "thread_id": "thread-1", + "available": True, + "status": "active", + "items": [ + {"id": "1", "content": "Inspect", "status": "completed"}, + {"id": "build", "content": "Build", "status": "in_progress"}, + ], + "completed": 1, + "total": 2, + } + + +def test_stream_projection_only_emits_when_todos_are_explicit() -> None: + assert ( + task_state_from_stream_chunk( + "thread-1", {"type": "state_snapshot", "data": {"messages": []}} + ) + is None + ) + + state = task_state_from_stream_chunk( + "thread-1", + { + "type": "state_update", + "data": {"todos": [{"content": "Ship", "status": "pending"}]}, + }, + ) + assert state is not None + assert state["status"] == "active" + assert state["items"][0]["content"] == "Ship" + + +@pytest.mark.asyncio +async def test_read_thread_task_state_reads_harness_graph() -> None: + class Graph: + async def aget_state(self, config: dict[str, Any]) -> SimpleNamespace: + assert config == {"configurable": {"thread_id": "thread-1"}} + return SimpleNamespace(values={"todos": [{"content": "Done", "status": "completed"}]}) + + manager = SimpleNamespace(get_agent=lambda _agent_id: SimpleNamespace(graph=Graph())) + + state = await read_thread_task_state(manager, "agent-1", "thread-1") + + assert state["status"] == "completed" + assert state["completed"] == 1 + + +@pytest.mark.asyncio +async def test_read_thread_task_state_is_unavailable_when_agent_is_stopped() -> None: + def unavailable(_agent_id: str) -> None: + raise RuntimeError("stopped") + + state = await read_thread_task_state( + SimpleNamespace(get_agent=unavailable), "agent-1", "thread-1" + ) + + assert state["available"] is False + assert state["status"] == "idle" diff --git a/tests/unit/api/test_chat_history.py b/tests/unit/api/test_chat_history.py index b9fd2a89..4e27674f 100644 --- a/tests/unit/api/test_chat_history.py +++ b/tests/unit/api/test_chat_history.py @@ -162,6 +162,20 @@ async def test_get_thread_history_returns_has_more(monkeypatch: pytest.MonkeyPat "_load_thread_messages", AsyncMock(return_value=([{"role": "user", "content": "hi"}], True)), ) + monkeypatch.setattr( + history_mod, + "read_thread_task_state", + AsyncMock( + return_value={ + "thread_id": "thr_1", + "available": True, + "status": "active", + "items": [{"id": "1", "content": "Ship", "status": "pending"}], + "completed": 0, + "total": 1, + } + ), + ) out = await history_mod.get_thread_history( "agt_1", "thr_1", @@ -175,6 +189,7 @@ async def test_get_thread_history_returns_has_more(monkeypatch: pytest.MonkeyPat assert out["limit"] == HISTORY_DEFAULT_LIMIT assert out["offset"] == 0 assert out["messages"][0]["role"] == "user" + assert out["task_state"]["items"][0]["content"] == "Ship" @pytest.mark.asyncio diff --git a/tests/unit/gateway/test_dashboard_ws.py b/tests/unit/gateway/test_dashboard_ws.py index 253dab7e..f367b428 100644 --- a/tests/unit/gateway/test_dashboard_ws.py +++ b/tests/unit/gateway/test_dashboard_ws.py @@ -88,10 +88,13 @@ async def capture_b(frame: dict[str, Any]) -> None: def test_ws_hub_turn_active_flags() -> None: hub = WebSocketHub() assert hub.is_turn_active("t1") is False + assert hub.turn_started_at("t1") is None hub.mark_turn_active("t1") assert hub.is_turn_active("t1") is True + assert isinstance(hub.turn_started_at("t1"), int) hub.mark_turn_idle("t1") assert hub.is_turn_active("t1") is False + assert hub.turn_started_at("t1") is None @pytest.mark.asyncio