From 5394ba43142c451e3bc1c973de895ece8e8aa71f Mon Sep 17 00:00:00 2001 From: Astrid Gealer Date: Sat, 10 Oct 2026 09:54:17 +0000 Subject: [PATCH 01/25] feat: task graphs for parallel agent work across worktrees and machines Agents can plan a request as a graph of tasks. Each node runs as its own orchestration v2 thread in its own worktree, starting once its dependencies succeed; nodes with several dependencies merge their branches, and each end of the tree commits, pushes and opens a pull request. The proposing thread gets a report when the graph finishes. - TaskGraphService drives ThreadLaunchService and runStackedAction; graphs persist as one row each and recover after restarts. - task_graph_* MCP tools let agents create, list, edit (add, update, remove, cancel_branch, retry), run and cancel graphs; the web editor and mobile use the same edit operations. - taskGraphAutoRun and taskGraphMaxConcurrentNodes server settings; prompt wording overrides auto-run per graph. - Peer machines: pair another environment with a scoped bearer session and place nodes on whichever machine scores best with the shared load balancing scorer. - Web: graph card above the composer, lazy-loaded canvas editor, settings for auto-run, concurrency and peer machines. Mobile: graph sheet with run, cancel, cancel branch and retry. Co-Authored-By: Claude Opus 5.5 --- apps/mobile/src/Stack.tsx | 11 + .../src/features/threads/TaskGraphSheet.tsx | 403 ++++++++ .../features/threads/ThreadDetailScreen.tsx | 13 + .../threads/floating-working-control.tsx | 36 +- .../threads/task-graph-presentation.test.ts | 135 +++ .../threads/task-graph-presentation.ts | 164 ++++ apps/server/src/auth/RpcAuthorization.ts | 2 + apps/server/src/mcp/McpHttpServer.ts | 5 + .../src/mcp/toolkits/taskGraph/handlers.ts | 81 ++ .../src/mcp/toolkits/taskGraph/tools.ts | 106 +++ .../toolkits/worktree/registration.test.ts | 2 + .../src/observability/RpcInstrumentation.ts | 8 + .../src/orchestration-v2/runtimeLayer.ts | 16 + apps/server/src/persistence/Migrations.ts | 2 + .../persistence/Migrations/061_TaskGraphs.ts | 41 + apps/server/src/taskGraph/TaskGraphPeers.ts | 727 +++++++++++++++ .../src/taskGraph/TaskGraphService.test.ts | 341 +++++++ apps/server/src/taskGraph/TaskGraphService.ts | 877 ++++++++++++++++++ apps/server/src/ws.ts | 16 + apps/web/package.json | 2 + apps/web/src/components/ChatView.tsx | 45 +- apps/web/src/components/chat/ChatComposer.tsx | 2 +- .../web/src/components/chat/TaskGraphCard.tsx | 310 +++++++ .../src/components/chat/TaskGraphEditor.css | 14 + .../src/components/chat/TaskGraphEditor.tsx | 660 +++++++++++++ .../chat/taskGraphCanvasLayout.test.ts | 105 +++ .../components/chat/taskGraphCanvasLayout.ts | 115 +++ .../src/components/chat/taskGraphView.test.ts | 105 +++ apps/web/src/components/chat/taskGraphView.ts | 119 +++ .../components/chat/useTaskGraphCommands.ts | 74 ++ .../settings/ConnectionsSettings.tsx | 2 + .../settings/LoadBalancingSettings.tsx | 10 +- .../components/settings/SettingsPanels.tsx | 85 ++ .../settings/TaskGraphMachinesSettings.tsx | 263 ++++++ .../src/components/settings/settingsSearch.ts | 22 + docs/README.md | 1 + docs/internals/task-graphs.md | 50 + docs/user/task-graphs.md | 66 ++ packages/client-runtime/src/load-balancing.ts | 42 +- packages/client-runtime/src/rpc/client.ts | 2 + packages/client-runtime/src/state/server.ts | 44 + packages/contracts/src/baseSchemas.ts | 2 + .../contracts/src/clientRpcPermissions.ts | 8 + packages/contracts/src/index.ts | 1 + packages/contracts/src/rpc.ts | 81 ++ packages/contracts/src/settings.ts | 16 + packages/contracts/src/taskGraph.ts | 217 +++++ packages/shared/package.json | 8 + packages/shared/src/loadBalancing.ts | 43 + packages/shared/src/taskGraph.test.ts | 156 ++++ packages/shared/src/taskGraph.ts | 342 +++++++ pnpm-lock.yaml | 86 ++ vite.config.ts | 1 + 53 files changed, 6016 insertions(+), 69 deletions(-) create mode 100644 apps/mobile/src/features/threads/TaskGraphSheet.tsx create mode 100644 apps/mobile/src/features/threads/task-graph-presentation.test.ts create mode 100644 apps/mobile/src/features/threads/task-graph-presentation.ts create mode 100644 apps/server/src/mcp/toolkits/taskGraph/handlers.ts create mode 100644 apps/server/src/mcp/toolkits/taskGraph/tools.ts create mode 100644 apps/server/src/persistence/Migrations/061_TaskGraphs.ts create mode 100644 apps/server/src/taskGraph/TaskGraphPeers.ts create mode 100644 apps/server/src/taskGraph/TaskGraphService.test.ts create mode 100644 apps/server/src/taskGraph/TaskGraphService.ts create mode 100644 apps/web/src/components/chat/TaskGraphCard.tsx create mode 100644 apps/web/src/components/chat/TaskGraphEditor.css create mode 100644 apps/web/src/components/chat/TaskGraphEditor.tsx create mode 100644 apps/web/src/components/chat/taskGraphCanvasLayout.test.ts create mode 100644 apps/web/src/components/chat/taskGraphCanvasLayout.ts create mode 100644 apps/web/src/components/chat/taskGraphView.test.ts create mode 100644 apps/web/src/components/chat/taskGraphView.ts create mode 100644 apps/web/src/components/chat/useTaskGraphCommands.ts create mode 100644 apps/web/src/components/settings/TaskGraphMachinesSettings.tsx create mode 100644 docs/internals/task-graphs.md create mode 100644 docs/user/task-graphs.md create mode 100644 packages/contracts/src/taskGraph.ts create mode 100644 packages/shared/src/loadBalancing.ts create mode 100644 packages/shared/src/taskGraph.test.ts create mode 100644 packages/shared/src/taskGraph.ts diff --git a/apps/mobile/src/Stack.tsx b/apps/mobile/src/Stack.tsx index ec4b2565e174..39adef216c20 100644 --- a/apps/mobile/src/Stack.tsx +++ b/apps/mobile/src/Stack.tsx @@ -53,6 +53,7 @@ import { GitCommitSheet } from "./features/threads/git/GitCommitSheet"; import { GitConfirmSheet } from "./features/threads/git/GitConfirmSheet"; import { GitOverviewSheet } from "./features/threads/git/GitOverviewSheet"; import { ThreadAgentsSheet } from "./features/threads/ThreadAgentsSheet"; +import { TaskGraphSheet } from "./features/threads/TaskGraphSheet"; import { ThreadQueueSheet } from "./features/threads/ThreadQueueControl"; import { ThreadRouteScreen } from "./features/threads/ThreadRouteScreen"; import { McpAppFullscreenScreen } from "./features/threads/McpAppFullscreenScreen"; @@ -556,6 +557,7 @@ const WORKSPACE_OVERLAY_ROUTES = new Set([ "SettingsSheet", "ThreadAgents", "ThreadQueue", + "ThreadTaskGraphs", "ThreadReviewComment", "ThreadDevicePreview", "ThreadBrowserPreview", @@ -810,6 +812,15 @@ const RootStackConfig = createWorkspaceStackNavigator({ sheetGrabberVisible: true, }, }), + ThreadTaskGraphs: createNativeStackScreen({ + screen: TaskGraphSheet, + options: { + ...FORM_SHEET_PRESENTATION_OPTIONS, + headerShown: false, + sheetAllowedDetents: [0.6, 0.95], + sheetGrabberVisible: true, + }, + }), GitOverview: createNativeStackScreen({ screen: GitOverviewSheet, linking: `${THREAD_LINKING_PREFIX}/git`, diff --git a/apps/mobile/src/features/threads/TaskGraphSheet.tsx b/apps/mobile/src/features/threads/TaskGraphSheet.tsx new file mode 100644 index 000000000000..55955125f98e --- /dev/null +++ b/apps/mobile/src/features/threads/TaskGraphSheet.tsx @@ -0,0 +1,403 @@ +import { useAtomValue } from "@effect/atom-react"; +import { + isAtomCommandInterrupted, + squashAtomCommandFailure, + type AtomCommandResult, +} from "@t3tools/client-runtime/state/runtime"; +import type { + EnvironmentId, + TaskGraph, + TaskGraphEdit, + TaskGraphNode, + ThreadId, +} from "@t3tools/contracts"; +import { taskGraphLayers } from "@t3tools/shared/taskGraph"; +import { StackActions, useNavigation, type StaticScreenProps } from "@react-navigation/native"; +import * as Haptics from "expo-haptics"; +import { useState } from "react"; +import { Alert, Platform, Pressable, ScrollView, View } from "react-native"; +import { Screen, ScreenStack, ScreenStackHeaderConfig } from "react-native-screens"; +import { useSafeAreaInsets } from "react-native-safe-area-context"; + +import { AndroidSheetHeader } from "../../components/AndroidScreenHeader"; +import { AppText as Text } from "../../components/AppText"; +import { showConfirmDialog } from "../../components/ConfirmDialogHost"; +import { tryOpenExternalUrl } from "../../lib/openExternalUrl"; +import { useUniwindTheme } from "../../lib/useUniwindTheme"; +import { nativeHeaderScrollEdgeEffects } from "../../native/StackHeader"; +import { useEnvironmentQuery } from "../../state/query"; +import { serverEnvironment } from "../../state/server"; +import { useAtomCommand } from "../../state/use-atom-command"; +import { RequestActionButton } from "./RequestActionButton"; +import { SUBAGENT_TONE_TEXT_CLASS, SubagentStatusDot } from "./SubagentStatusDot"; +import { + sortTaskGraphs, + taskGraphActions, + taskGraphNodeActions, + taskGraphNodeStatusLabel, + taskGraphNodeTone, + taskGraphSummaryLabel, + taskGraphTone, +} from "./task-graph-presentation"; + +const HEADER_SCROLL_EDGE_EFFECTS = nativeHeaderScrollEdgeEffects(Platform.OS, Platform.Version); + +type TaskGraphTarget = { readonly environmentId: EnvironmentId; readonly threadId: ThreadId }; + +const useTaskGraphsQuery = (target: TaskGraphTarget) => + useEnvironmentQuery( + serverEnvironment.taskGraphsLive({ + environmentId: target.environmentId, + input: { threadId: target.threadId }, + }), + ); + +/** The thread's task graphs, live. Null until the first snapshot arrives. */ +export function useThreadTaskGraphs(target: TaskGraphTarget): ReadonlyArray | null { + return useTaskGraphsQuery(target).data?.graphs ?? null; +} + +function confirm(input: { + readonly title: string; + readonly message: string; + readonly confirmText: string; + readonly onConfirm: () => void; +}) { + if (Platform.OS === "ios") { + Alert.alert(input.title, input.message, [ + { text: "Keep", style: "cancel" }, + { text: input.confirmText, style: "destructive", onPress: input.onConfirm }, + ]); + return; + } + showConfirmDialog({ ...input, cancelText: "Keep", destructive: true }); +} + +/** + * Read-only view of a thread's task graphs, with the actions that move a + * graph between states: run a draft, cancel, cancel a branch, retry a node. + * Editing the graph itself happens on web and desktop. + */ +export function TaskGraphSheet({ route }: StaticScreenProps) { + const target = route.params; + const navigation = useNavigation(); + const insets = useSafeAreaInsets(); + const theme = useUniwindTheme(); + const query = useTaskGraphsQuery(target); + const graphs = query.data === null ? null : sortTaskGraphs(query.data.graphs); + const canEdit = useAtomValue( + serverEnvironment.editTaskGraph.permissionAtom(target.environmentId), + ); + const canRun = useAtomValue(serverEnvironment.runTaskGraph.permissionAtom(target.environmentId)); + const canCancel = useAtomValue( + serverEnvironment.cancelTaskGraph.permissionAtom(target.environmentId), + ); + const editGraph = useAtomCommand(serverEnvironment.editTaskGraph, { + label: "task graph edit", + reportFailure: false, + }); + const runGraph = useAtomCommand(serverEnvironment.runTaskGraph, { + label: "task graph run", + reportFailure: false, + }); + const cancelGraph = useAtomCommand(serverEnvironment.cancelTaskGraph, { + label: "task graph cancel", + reportFailure: false, + }); + // Keys of actions in flight, so a second tap cannot send the same edit twice. + const [pending, setPending] = useState>(() => new Set()); + + const perform = async ( + key: string, + failureTitle: string, + action: () => Promise>, + ) => { + setPending((current) => new Set(current).add(key)); + try { + const result = await action(); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + Alert.alert(failureTitle, String(squashAtomCommandFailure(result))); + } + } finally { + setPending((current) => { + const next = new Set(current); + next.delete(key); + return next; + }); + } + }; + + const edit = (graph: TaskGraph, change: TaskGraphEdit & { readonly key: string }) => + perform( + `${graph.id}:${change.key}`, + change.type === "retry_node" ? "Could not retry task" : "Could not cancel branch", + () => + editGraph({ + environmentId: target.environmentId, + input: { graphId: graph.id, edits: [change] }, + }), + ); + + const onRun = (graph: TaskGraph) => { + void Haptics.selectionAsync(); + void perform(graph.id, "Could not run graph", () => + runGraph({ environmentId: target.environmentId, input: { graphId: graph.id } }), + ); + }; + + const onCancel = (graph: TaskGraph) => + confirm({ + title: "Cancel task graph?", + message: `Every unfinished task in “${graph.title}” stops. Finished tasks and their pull requests stay.`, + confirmText: "Cancel graph", + onConfirm: () => + void perform(graph.id, "Could not cancel graph", () => + cancelGraph({ environmentId: target.environmentId, input: { graphId: graph.id } }), + ), + }); + + const onCancelBranch = (graph: TaskGraph, node: TaskGraphNode) => + confirm({ + title: "Cancel this branch?", + message: `“${node.title}” and every task that depends on it stop.`, + confirmText: "Cancel branch", + onConfirm: () => void edit(graph, { type: "cancel_branch", key: node.key }), + }); + + const onRetry = (graph: TaskGraph, node: TaskGraphNode) => { + void Haptics.selectionAsync(); + void edit(graph, { type: "retry_node", key: node.key }); + }; + + const openNodeThread = (node: TaskGraphNode) => { + if (node.threadId === null) return; + void Haptics.selectionAsync(); + // Replace rather than push, like the agents sheet: the node thread belongs + // in the workspace stack, not on top of a dismissed sheet. + navigation.dispatch( + StackActions.replace("Thread", { + environmentId: node.assignedEnvironmentId ?? target.environmentId, + threadId: node.threadId, + }), + ); + }; + + const openPullRequest = async (url: string) => { + if (!(await tryOpenExternalUrl(url, "pull-request"))) { + Alert.alert("Unable to open PR", "The pull request could not be opened."); + } + }; + + const content = ( + + {query.error !== null && graphs === null ? ( + {query.error} + ) : graphs === null ? ( + Loading task graphs… + ) : graphs.length === 0 ? ( + + No task graphs on this thread. + + ) : ( + graphs.map((graph) => { + const actions = taskGraphActions(graph); + const busy = pending.has(graph.id); + return ( + + + + + + {graph.title} + + + + {taskGraphSummaryLabel(graph)} · from {graph.baseRef} + + + {(actions.run && canRun) || (actions.cancel && canCancel) ? ( + + {actions.run && canRun ? ( + onRun(graph)} /> + ) : null} + {actions.cancel && canCancel ? ( + onCancel(graph)} + /> + ) : null} + + ) : null} + {taskGraphLayers(graph.nodes).map((layer, index) => ( + + + Step {index + 1} + + {layer.map((node) => { + const nodeActions = taskGraphNodeActions(graph, node); + return ( + openNodeThread(node)} + onOpenPullRequest={(url) => void openPullRequest(url)} + onCancelBranch={() => onCancelBranch(graph, node)} + onRetry={() => onRetry(graph, node)} + /> + ); + })} + + ))} + + ); + }) + )} + + ); + + if (Platform.OS === "ios") { + // A plain formSheet screen never renders a stack header, so it comes from + // a nested native stack inside the sheet (same shape as the agents sheet). + return ( + + + + {content} + + + + + ); + } + + return ( + + navigation.goBack()} /> + {content} + + ); +} + +function NodeRow(props: { + readonly node: TaskGraphNode; + readonly busy: boolean; + readonly canCancelBranch: boolean; + readonly canRetry: boolean; + readonly onOpen: () => void; + readonly onOpenPullRequest: (url: string) => void; + readonly onCancelBranch: () => void; + readonly onRetry: () => void; +}) { + const { node } = props; + const tone = taskGraphNodeTone(node.status); + const pullRequestUrl = node.pullRequestResult?.url ?? null; + const pullRequestError = + node.pullRequestResult?.status === "failed" ? node.pullRequestResult.error : null; + const detail = node.status === "succeeded" ? pullRequestError : (node.error ?? pullRequestError); + const hasActions = pullRequestUrl !== null || props.canCancelBranch || props.canRetry; + + const summary = ( + + + + + {node.title} + + + {taskGraphNodeStatusLabel(node.status)} + + + {node.branch !== null ? ( + + {node.branch} + + ) : null} + {detail ? ( + + {detail} + + ) : null} + + ); + + return ( + + {node.threadId === null ? ( + {summary} + ) : ( + + {summary} + + )} + {hasActions ? ( + + {pullRequestUrl !== null ? ( + props.onOpenPullRequest(pullRequestUrl)} /> + ) : null} + {props.canRetry ? ( + + ) : null} + {props.canCancelBranch ? ( + + ) : null} + + ) : null} + + ); +} + +function NodeAction(props: { + readonly label: string; + readonly disabled?: boolean; + readonly onPress: () => void; +}) { + return ( + + {props.label} + + ); +} diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 9229c117d8da..6e8b6d752782 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -144,6 +144,8 @@ import { import { ComposerPopoverHost } from "./ComposerPopoverHost"; import { ThreadFeed, type ThreadFeedHistoryControls } from "./ThreadFeed"; import { useThreadTurnSubagents } from "./ThreadAgentsSheet"; +import { useThreadTaskGraphs } from "./TaskGraphSheet"; +import { resolveTaskGraphSegment } from "./task-graph-presentation"; import { ComposerQueuedEditBanner } from "./ComposerQueuedEdit"; import { useThreadQueuedCount } from "./ThreadQueueControl"; import type { ThreadContentPresentation } from "./threadContentPresentation"; @@ -410,6 +412,9 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread threadId: props.selectedThread.id, }); const agentsSegment = resolveSubagentPillSegment(turnSubagents); + const taskGraphSegment = resolveTaskGraphSegment( + useThreadTaskGraphs({ environmentId: props.environmentId, threadId: props.selectedThread.id }), + ); const composerEditorRef = useRef(null); // A provider-native subagent shows status instead of a composer. const isProviderSubagent = isProviderNativeSubagentThread(props.selectedThread.source); @@ -1238,6 +1243,14 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread threadId: props.selectedThread.id, }); }} + taskGraph={taskGraphSegment} + onOpenTaskGraph={() => { + Keyboard.dismiss(); + navigation.navigate("ThreadTaskGraphs", { + environmentId: props.environmentId, + threadId: props.selectedThread.id, + }); + }} queuedCount={queuedCount} onOpenQueue={() => { Keyboard.dismiss(); diff --git a/apps/mobile/src/features/threads/floating-working-control.tsx b/apps/mobile/src/features/threads/floating-working-control.tsx index f21a83f09a17..c6c9434f2711 100644 --- a/apps/mobile/src/features/threads/floating-working-control.tsx +++ b/apps/mobile/src/features/threads/floating-working-control.tsx @@ -29,6 +29,8 @@ import { NATIVE_LIQUID_GLASS_SUPPORTED } from "../../native/native-glass"; import { BrowserPreviewButton } from "../browser/browser-preview-button"; import { DevicePreviewButton } from "../devices/device-preview-button"; import type { FloatingWorkingStatus } from "./floating-working-status"; +import { SubagentStatusDot } from "./SubagentStatusDot"; +import type { TaskGraphSegment } from "./task-graph-presentation"; import { ShimmeringWorkContent } from "./thread-work-log"; import { useVisibleSecondClock } from "./use-visible-second-clock"; @@ -74,6 +76,8 @@ export function FloatingWorkingControl(props: { readonly onScrollToEnd: () => void; readonly agents: SubagentPillSegment | null; readonly onOpenAgents: () => void; + readonly taskGraph: TaskGraphSegment | null; + readonly onOpenTaskGraph: () => void; readonly queuedCount: number; readonly onOpenQueue: () => void; /** Extra distance to rise above the anchor, e.g. an overlay card's coverage. */ @@ -83,11 +87,14 @@ export function FloatingWorkingControl(props: { const [overlayWidth, setOverlayWidth] = useState(windowWidth); const [queueWidth, setQueueWidth] = useState(0); const [agentsWidth, setAgentsWidth] = useState(0); + const [taskGraphWidth, setTaskGraphWidth] = useState(0); const hasQueue = props.queuedCount > 0; const hasPreview = props.devicePreview !== null || props.browserPreview !== null; const [previewWidth, setPreviewWidth] = useState(0); const agents = props.agents; const hasAgents = agents !== null; + const taskGraph = props.taskGraph; + const hasTaskGraph = taskGraph !== null; // Segments keep their measured width; only the status label absorbs the // remainder, so a long "Working 12m 04s" truncates before a count does. const labelWidth = Math.max( @@ -97,6 +104,7 @@ export function FloatingWorkingControl(props: { 32 - (hasQueue ? queueWidth : 0) - (hasAgents ? agentsWidth : 0) - + (hasTaskGraph ? taskGraphWidth : 0) - (hasPreview ? previewWidth : 0), ); const separationProgress = useSharedValue(props.showScrollToEnd ? 1 : 0); @@ -134,7 +142,7 @@ export function FloatingWorkingControl(props: { // Forget the width while no label is shown so the next one appears at its // own size instead of animating from the previous label's. const hasStatus = props.status !== null; - const hasCapsule = hasStatus || hasQueue || hasAgents || hasPreview; + const hasCapsule = hasStatus || hasQueue || hasAgents || hasTaskGraph || hasPreview; useEffect(() => { if (!hasStatus) { measuredWidthRef.current = null; @@ -158,11 +166,13 @@ export function FloatingWorkingControl(props: { const previewCompact = hasStatus || hasAgents || + hasTaskGraph || hasQueue || (props.devicePreview !== null && props.browserPreview !== null); - // The queue, agents, and reconnect labels have separate tap targets. + // The queue, agents, task graph, and reconnect labels have separate tap targets. const statusInteractive = props.status?.kind === "connection"; - const capsuleInteractive = statusInteractive || hasQueue || hasAgents || hasPreview; + const capsuleInteractive = + statusInteractive || hasQueue || hasAgents || hasTaskGraph || hasPreview; // The host stays centered on the capsule, but its measurement constraint // comes from the overlay, independent of the capsule's current width. const statusContent = @@ -223,6 +233,24 @@ export function FloatingWorkingControl(props: { ) : null} + {taskGraph !== null ? ( + setTaskGraphWidth(event.nativeEvent.layout.width)} + className="h-11 flex-row items-center gap-1.5 px-3 active:opacity-70" + > + {hasStatus || hasPreview || hasAgents ? ( + + ) : null} + + + {taskGraph.label} + + + ) : null} {hasQueue ? ( - {hasStatus || hasPreview || hasAgents ? ( + {hasStatus || hasPreview || hasAgents || hasTaskGraph ? ( ) : null} diff --git a/apps/mobile/src/features/threads/task-graph-presentation.test.ts b/apps/mobile/src/features/threads/task-graph-presentation.test.ts new file mode 100644 index 000000000000..c1daa1a16ac0 --- /dev/null +++ b/apps/mobile/src/features/threads/task-graph-presentation.test.ts @@ -0,0 +1,135 @@ +import type { TaskGraph, TaskGraphNode } from "@t3tools/contracts"; +import { describe, expect, it } from "vite-plus/test"; + +import { + resolveTaskGraphSegment, + sortTaskGraphs, + taskGraphActions, + taskGraphNodeActions, + taskGraphSummaryLabel, +} from "./task-graph-presentation"; + +const node = (key: string, status: TaskGraphNode["status"], dependsOn: string[] = []) => + ({ key, status, dependsOn, title: key }) as unknown as TaskGraphNode; + +const graph = ( + status: TaskGraph["status"], + nodes: TaskGraphNode[], + overrides: Partial = {}, +) => + ({ + id: `graph-${status}`, + title: "Ship auth", + status, + nodes, + updatedAt: "2026-10-01T00:00:00.000Z", + ...overrides, + }) as unknown as TaskGraph; + +describe("taskGraphSummaryLabel", () => { + it("counts finished tasks while running", () => { + expect( + taskGraphSummaryLabel( + graph("running", [node("a", "succeeded"), node("b", "running"), node("c", "pending")]), + ), + ).toBe("Running · 1 of 3 done"); + }); + + it("names failures on a failed graph", () => { + expect( + taskGraphSummaryLabel(graph("failed", [node("a", "failed"), node("b", "skipped", ["a"])])), + ).toBe("Failed · 1 failed"); + }); + + it("counts tasks on a draft", () => { + expect(taskGraphSummaryLabel(graph("draft", [node("a", "pending")]))).toBe("Draft · 1 task"); + }); +}); + +describe("resolveTaskGraphSegment", () => { + it("is absent without graphs", () => { + expect(resolveTaskGraphSegment(null)).toBeNull(); + expect(resolveTaskGraphSegment([])).toBeNull(); + }); + + it("prefers an open graph over a newer finished one", () => { + const segment = resolveTaskGraphSegment([ + graph("succeeded", [node("a", "succeeded")], { updatedAt: "2026-10-09T00:00:00.000Z" }), + graph("running", [node("a", "succeeded"), node("b", "running")]), + ]); + expect(segment).toMatchObject({ label: "Graph 1/2", tone: "working" }); + }); + + it("stays reachable once every graph has finished", () => { + expect(resolveTaskGraphSegment([graph("failed", [node("a", "failed")])])).toMatchObject({ + label: "Graph failed", + tone: "failed", + }); + }); + + it("counts several open graphs", () => { + expect( + resolveTaskGraphSegment([ + graph("draft", [node("a", "pending")]), + graph("running", [node("a", "running")]), + ])?.label, + ).toBe("2 graphs"); + }); +}); + +describe("sortTaskGraphs", () => { + it("puts running before draft before finished, newest first within each", () => { + const sorted = sortTaskGraphs([ + graph("succeeded", [], { id: "old" as never, updatedAt: "2026-10-01T00:00:00.000Z" }), + graph("draft", [], { id: "draft" as never }), + graph("failed", [], { id: "new" as never, updatedAt: "2026-10-05T00:00:00.000Z" }), + graph("running", [], { id: "running" as never }), + ]); + expect(sorted.map((entry) => entry.id)).toEqual(["running", "draft", "new", "old"]); + }); +}); + +describe("taskGraphActions", () => { + it("runs only drafts and cancels only open graphs", () => { + expect(taskGraphActions({ status: "draft" })).toEqual({ run: true, cancel: true }); + expect(taskGraphActions({ status: "running" })).toEqual({ run: false, cancel: true }); + expect(taskGraphActions({ status: "failed" })).toEqual({ run: false, cancel: false }); + }); +}); + +describe("taskGraphNodeActions", () => { + it("cancels the branch of unfinished nodes in an open graph", () => { + const nodes = [node("a", "running"), node("b", "pending", ["a"])]; + expect(taskGraphNodeActions(graph("running", nodes), nodes[0]!)).toEqual({ + cancelBranch: true, + retry: false, + }); + expect(taskGraphNodeActions(graph("running", nodes), nodes[1]!).cancelBranch).toBe(true); + }); + + it("retries failed nodes, including in a finished graph", () => { + const nodes = [node("a", "failed")]; + expect(taskGraphNodeActions(graph("failed", nodes), nodes[0]!)).toEqual({ + cancelBranch: false, + retry: true, + }); + }); + + it("offers no retry for a skipped node whose dependency still blocks it", () => { + const nodes = [node("a", "failed"), node("b", "skipped", ["a"])]; + expect(taskGraphNodeActions(graph("failed", nodes), nodes[1]!).retry).toBe(false); + }); + + it("retries a skipped node once its dependency has succeeded", () => { + const nodes = [node("a", "succeeded"), node("b", "skipped", ["a"])]; + expect(taskGraphNodeActions(graph("failed", nodes), nodes[1]!).retry).toBe(true); + }); + + it("takes no edits on a cancelled graph", () => { + const nodes = [node("a", "cancelled")]; + expect(taskGraphNodeActions(graph("cancelled", nodes), nodes[0]!)).toEqual({ + cancelBranch: false, + retry: false, + }); + }); +}); diff --git a/apps/mobile/src/features/threads/task-graph-presentation.ts b/apps/mobile/src/features/threads/task-graph-presentation.ts new file mode 100644 index 000000000000..0f55c50b9909 --- /dev/null +++ b/apps/mobile/src/features/threads/task-graph-presentation.ts @@ -0,0 +1,164 @@ +import type { + TaskGraph, + TaskGraphNode, + TaskGraphNodeStatus, + TaskGraphStatus, +} from "@t3tools/contracts"; +import { isActiveTaskGraphNodeStatus } from "@t3tools/shared/taskGraph"; + +import type { SubagentRowTone } from "./threadAgentsPresentation"; + +const BLOCKING_STATUSES: ReadonlySet = new Set([ + "failed", + "cancelled", + "skipped", +]); + +export const isOpenTaskGraph = (graph: Pick): boolean => + graph.status === "draft" || graph.status === "running"; + +export function taskGraphNodeTone(status: TaskGraphNodeStatus): SubagentRowTone { + if (isActiveTaskGraphNodeStatus(status)) return "working"; + if (status === "succeeded") return "completed"; + if (status === "failed") return "failed"; + return "stopped"; +} + +export function taskGraphTone(status: TaskGraphStatus): SubagentRowTone { + if (status === "running") return "working"; + if (status === "succeeded") return "completed"; + if (status === "failed") return "failed"; + return "stopped"; +} + +export function taskGraphNodeStatusLabel(status: TaskGraphNodeStatus): string { + switch (status) { + case "pending": + return "Waiting"; + case "running": + return "Running"; + case "delivering": + return "Opening PR"; + case "succeeded": + return "Done"; + case "failed": + return "Failed"; + case "cancelled": + return "Cancelled"; + case "skipped": + return "Skipped"; + } +} + +export function taskGraphStatusLabel(status: TaskGraphStatus): string { + switch (status) { + case "draft": + return "Draft"; + case "running": + return "Running"; + case "succeeded": + return "Done"; + case "failed": + return "Failed"; + case "cancelled": + return "Cancelled"; + } +} + +/** One-line status for a graph, such as "Running · 2 of 5 done". */ +export function taskGraphSummaryLabel(graph: Pick): string { + const total = graph.nodes.length; + const tasks = `${total} ${total === 1 ? "task" : "tasks"}`; + if (graph.status === "draft" || graph.status === "succeeded") { + return `${taskGraphStatusLabel(graph.status)} · ${tasks}`; + } + if (graph.status === "running") { + const done = graph.nodes.filter((node) => node.status === "succeeded").length; + return `Running · ${done} of ${total} done`; + } + const failed = graph.nodes.filter((node) => node.status === "failed").length; + return failed > 0 + ? `${taskGraphStatusLabel(graph.status)} · ${failed} failed` + : `${taskGraphStatusLabel(graph.status)} · ${tasks}`; +} + +/** Open graphs first (running before draft), then the most recently changed. */ +export function sortTaskGraphs(graphs: ReadonlyArray): ReadonlyArray { + const rank = (graph: TaskGraph) => + graph.status === "running" ? 0 : graph.status === "draft" ? 1 : 2; + return graphs.toSorted( + (left, right) => rank(left) - rank(right) || right.updatedAt.localeCompare(left.updatedAt), + ); +} + +export interface TaskGraphSegment { + readonly label: string; + readonly accessibilityLabel: string; + readonly tone: SubagentRowTone; +} + +/** + * The thread's task graph entry point. Finished graphs keep it visible so a + * failed node can still be retried and its pull requests reached. + */ +export function resolveTaskGraphSegment( + graphs: ReadonlyArray | null, +): TaskGraphSegment | null { + if (graphs === null || graphs.length === 0) return null; + const open = graphs.filter(isOpenTaskGraph); + if (open.length > 1) { + return { + label: `${open.length} graphs`, + accessibilityLabel: `${open.length} open task graphs`, + tone: open.some((graph) => graph.status === "running") ? "working" : "stopped", + }; + } + const graph = sortTaskGraphs(graphs)[0]!; + const label = + graph.status === "running" + ? `Graph ${graph.nodes.filter((node) => node.status === "succeeded").length}/${graph.nodes.length}` + : `Graph ${taskGraphStatusLabel(graph.status).toLowerCase()}`; + return { + label, + accessibilityLabel: `Task graph ${graph.title}, ${taskGraphSummaryLabel(graph)}`, + tone: taskGraphTone(graph.status), + }; +} + +export interface TaskGraphActions { + readonly run: boolean; + readonly cancel: boolean; +} + +export const taskGraphActions = (graph: Pick): TaskGraphActions => ({ + run: graph.status === "draft", + cancel: isOpenTaskGraph(graph), +}); + +export interface TaskGraphNodeActions { + readonly cancelBranch: boolean; + readonly retry: boolean; +} + +/** + * What a node offers. A cancelled graph takes no edits. Retrying a skipped + * node whose dependency still has not succeeded would only skip it again, so + * that retry belongs on the dependency instead. + */ +export function taskGraphNodeActions( + graph: Pick, + node: Pick, +): TaskGraphNodeActions { + if (graph.status === "cancelled") return { cancelBranch: false, retry: false }; + const cancelBranch = + isOpenTaskGraph(graph) && + (node.status === "pending" || isActiveTaskGraphNodeStatus(node.status)); + if (node.status === "failed" || node.status === "cancelled") return { cancelBranch, retry: true }; + if (node.status !== "skipped") return { cancelBranch, retry: false }; + const statusByKey = new Map(graph.nodes.map((other) => [other.key, other.status])); + const blocked = node.dependsOn.some((key) => { + const status = statusByKey.get(key); + return status !== undefined && BLOCKING_STATUSES.has(status); + }); + return { cancelBranch, retry: !blocked }; +} diff --git a/apps/server/src/auth/RpcAuthorization.ts b/apps/server/src/auth/RpcAuthorization.ts index a33d2f3317aa..23b675f0dd19 100644 --- a/apps/server/src/auth/RpcAuthorization.ts +++ b/apps/server/src/auth/RpcAuthorization.ts @@ -111,6 +111,8 @@ export const RPC_REQUIRED_SCOPES = { [WS_METHODS.serverGetBackgroundPolicy]: AuthOrchestrationReadScope, [WS_METHODS.scheduledTasksList]: AuthOrchestrationReadScope, [WS_METHODS.scheduledTasksSubscribe]: AuthOrchestrationReadScope, + [WS_METHODS.taskGraphsSubscribe]: AuthOrchestrationReadScope, + [WS_METHODS.taskGraphPeersSubscribe]: AuthOrchestrationReadScope, [WS_METHODS.secretsAnswerRequest]: AuthOrchestrationOperateScope, // Delivery logs hold request bodies, so they need the same scope as the URL. [WS_METHODS.scheduledTasksListWebhookDeliveries]: AuthOrchestrationOperateScope, diff --git a/apps/server/src/mcp/McpHttpServer.ts b/apps/server/src/mcp/McpHttpServer.ts index 0e2b52696712..26d7181d5d6a 100644 --- a/apps/server/src/mcp/McpHttpServer.ts +++ b/apps/server/src/mcp/McpHttpServer.ts @@ -47,6 +47,8 @@ import { } from "./toolkits/preview/tools.ts"; import * as WorktreeHandlers from "./toolkits/worktree/handlers.ts"; import { WorktreeToolkit } from "./toolkits/worktree/tools.ts"; +import * as TaskGraphHandlers from "./toolkits/taskGraph/handlers.ts"; +import { TaskGraphToolkit } from "./toolkits/taskGraph/tools.ts"; import * as WorktreeMcpService from "./WorktreeMcpService.ts"; import * as PullRequestsHandlers from "./toolkits/pullRequests/handlers.ts"; import { PullRequestsToolkit } from "./toolkits/pullRequests/tools.ts"; @@ -816,6 +818,8 @@ export const layerOrchestratorToolkit = toolkitRegistration( export const layerThreadToolkit = toolkitRegistration(ThreadToolkit, ThreadHandlers.layer); +export const layerTaskGraphToolkit = toolkitRegistration(TaskGraphToolkit, TaskGraphHandlers.layer); + const layerWorktreeToolkitRegistration = toolkitRegistration( WorktreeToolkit, WorktreeHandlers.layer, @@ -870,6 +874,7 @@ export const layer = Layer.mergeAll( layerPreviewToolkit, layerOrchestratorToolkit, layerThreadToolkit, + layerTaskGraphToolkit, layerAttachmentToolkit, layerProjectRegistration, layerEnvironmentToolkit, diff --git a/apps/server/src/mcp/toolkits/taskGraph/handlers.ts b/apps/server/src/mcp/toolkits/taskGraph/handlers.ts new file mode 100644 index 000000000000..d31da830c790 --- /dev/null +++ b/apps/server/src/mcp/toolkits/taskGraph/handlers.ts @@ -0,0 +1,81 @@ +import { OrchestratorMcpFailure, type TaskGraphId, type ThreadId } from "@t3tools/contracts"; +import * as Effect from "effect/Effect"; + +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as McpToolAccess from "../../McpToolAccess.ts"; +import * as TaskGraphService from "../../../taskGraph/TaskGraphService.ts"; +import { TaskGraphToolkit } from "./tools.ts"; + +const failure = (code: "invalid_request" | "orchestration_error", message: string) => + new OrchestratorMcpFailure({ code, message }); + +/** The calling thread; the access gate already refused callers without one. */ +const callerThread = Effect.gen(function* () { + const scope = yield* McpInvocationContext.McpInvocationContext; + if (scope.thread === undefined) { + return yield* failure( + "invalid_request", + "Task graphs need an agent running inside a T3 thread.", + ); + } + return scope.thread.threadId; +}); + +const orchestrationError = (error: { readonly message: string }) => + failure("orchestration_error", error.message); + +/** A graph the calling thread created; agents only steer their own graphs. */ +const ownGraph = (graphId: TaskGraphId, threadId: ThreadId) => + Effect.gen(function* () { + const service = yield* TaskGraphService.TaskGraphService; + const graph = yield* service.get(graphId).pipe(Effect.mapError(orchestrationError)); + if (graph.threadId !== threadId) { + return yield* failure("invalid_request", "That task graph belongs to another thread."); + } + return service; + }); + +const handlers = { + task_graph_create: McpToolAccess.actsAsCaller((input) => + Effect.gen(function* () { + const threadId = yield* callerThread; + const service = yield* TaskGraphService.TaskGraphService; + const graph = yield* service + .create({ ...input, threadId }) + .pipe(Effect.mapError(orchestrationError)); + return { graph }; + }), + ), + task_graph_list: McpToolAccess.readsAsCaller(() => + Effect.gen(function* () { + const threadId = yield* callerThread; + const service = yield* TaskGraphService.TaskGraphService; + const graphs = yield* service + .listForThread(threadId) + .pipe(Effect.mapError(orchestrationError)); + return { graphs }; + }), + ), + task_graph_edit: McpToolAccess.actsAsCaller(({ graphId, edits }) => + Effect.gen(function* () { + const service = yield* ownGraph(graphId, yield* callerThread); + return { + graph: yield* service.edit(graphId, edits).pipe(Effect.mapError(orchestrationError)), + }; + }), + ), + task_graph_run: McpToolAccess.actsAsCaller(({ graphId }) => + Effect.gen(function* () { + const service = yield* ownGraph(graphId, yield* callerThread); + return { graph: yield* service.run(graphId).pipe(Effect.mapError(orchestrationError)) }; + }), + ), + task_graph_cancel: McpToolAccess.actsAsCaller(({ graphId }) => + Effect.gen(function* () { + const service = yield* ownGraph(graphId, yield* callerThread); + return { graph: yield* service.cancel(graphId).pipe(Effect.mapError(orchestrationError)) }; + }), + ), +} satisfies McpToolAccess.Handlers; + +export const layer = McpToolAccess.toLayer(TaskGraphToolkit, handlers); diff --git a/apps/server/src/mcp/toolkits/taskGraph/tools.ts b/apps/server/src/mcp/toolkits/taskGraph/tools.ts new file mode 100644 index 000000000000..2e54aa818310 --- /dev/null +++ b/apps/server/src/mcp/toolkits/taskGraph/tools.ts @@ -0,0 +1,106 @@ +import { + MAX_TASK_GRAPH_NODES, + OrchestratorMcpFailure, + TaskGraph, + TaskGraphEdit, + TaskGraphId, + TaskGraphNodeInput, + TrimmedNonEmptyString, +} from "@t3tools/contracts"; +import { Tool, Toolkit } from "effect/ai"; +import * as Schema from "effect/Schema"; + +import * as McpInvocationContext from "../../McpInvocationContext.ts"; +import * as ThreadManagementService from "../../../orchestration-v2/ThreadManagementService.ts"; +import * as TaskGraphService from "../../../taskGraph/TaskGraphService.ts"; + +const dependencies = [ + McpInvocationContext.McpInvocationContext, + ThreadManagementService.ThreadManagementService, + TaskGraphService.TaskGraphService, +]; + +const GraphResult = Schema.Struct({ graph: TaskGraph }); +const GraphTarget = Schema.Struct({ graphId: TaskGraphId }); + +export const TaskGraphCreateTool = Tool.make("task_graph_create", { + description: + "Needs an agent running inside a T3 thread. Plan work as a graph of agent tasks and run it. Each node runs as its own T3 thread in its own git worktree, in parallel where dependencies allow. A node with no dependencies branches from baseRef (default: this thread's branch); a node with one dependency branches from that dependency's branch; a node with several starts from the first one's branch and merges the others before its task, and receives every dependency's final summary. When a node succeeds its work is committed; nodes nothing depends on also push and open a pull request (override per node with pullRequest), so each branch end of the tree becomes one PR. When the whole graph ends, this thread gets a report message. Use it when a request splits into independent pieces (audits of separate areas, features with sub-parts, a merge step that combines them). Write each prompt so an agent with no access to this conversation can do it. RUNNING: if the user asked to see, review or edit the plan first, pass run=false so it waits as a draft they can edit in the chat; if they asked to just do it, pass run=true; otherwise omit run to follow their setting. Change a graph later with task_graph_edit.", + parameters: Schema.Struct({ + title: TrimmedNonEmptyString.check(Schema.isMaxLength(120)), + nodes: Schema.Array(TaskGraphNodeInput).check( + Schema.isMinLength(1), + Schema.isMaxLength(MAX_TASK_GRAPH_NODES), + ), + baseRef: Schema.optional(TrimmedNonEmptyString), + run: Schema.optional(Schema.Boolean), + }), + success: GraphResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Create a task graph") + .annotate(Tool.Destructive, false) + .annotate(Tool.OpenWorld, true); + +const TaskGraphListTool = Tool.make("task_graph_list", { + description: + "Needs an agent running inside a T3 thread. List the task graphs this thread created, with every node's status, thread, branch, summary and pull request.", + success: Schema.Struct({ graphs: Schema.Array(TaskGraph) }), + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "List task graphs") + .annotate(Tool.Readonly, true) + .annotate(Tool.Destructive, false) + .annotate(Tool.Idempotent, true); + +const TaskGraphEditTool = Tool.make("task_graph_edit", { + description: + "Change a task graph from this thread, running or not. Edits apply in order and all or nothing. add_node adds work (it starts as soon as its dependencies succeeded); update_node and remove_node change nodes that have not started; cancel_branch cuts a node and everything below it, stopping any that are running; retry_node reruns a node that failed or was cancelled, along with what was skipped below it.", + parameters: Schema.Struct({ + graphId: TaskGraphId, + edits: Schema.Array(TaskGraphEdit).check(Schema.isMinLength(1)), + }), + success: GraphResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Edit a task graph") + .annotate(Tool.Destructive, true); + +const TaskGraphRunTool = Tool.make("task_graph_run", { + description: + "Start a draft task graph from this thread. Only call this when the user asks to run it.", + parameters: GraphTarget, + success: GraphResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Run a task graph") + .annotate(Tool.Destructive, false) + .annotate(Tool.OpenWorld, true); + +const TaskGraphCancelTool = Tool.make("task_graph_cancel", { + description: + "Cancel a whole task graph from this thread, stopping every running node. Finished nodes and their pull requests stay.", + parameters: GraphTarget, + success: GraphResult, + failure: OrchestratorMcpFailure, + failureMode: "return", + dependencies, +}) + .annotate(Tool.Title, "Cancel a task graph") + .annotate(Tool.Destructive, true); + +export const TaskGraphToolkit = Toolkit.make( + TaskGraphCreateTool, + TaskGraphListTool, + TaskGraphEditTool, + TaskGraphRunTool, + TaskGraphCancelTool, +); diff --git a/apps/server/src/mcp/toolkits/worktree/registration.test.ts b/apps/server/src/mcp/toolkits/worktree/registration.test.ts index 91257c2fdb80..691879d02cd2 100644 --- a/apps/server/src/mcp/toolkits/worktree/registration.test.ts +++ b/apps/server/src/mcp/toolkits/worktree/registration.test.ts @@ -1,4 +1,5 @@ import * as Orchestrator from "../../../orchestration-v2/Orchestrator.ts"; +import * as TaskGraphService from "../../../taskGraph/TaskGraphService.ts"; import * as ProjectionStore from "../../../orchestration-v2/ProjectionStore.ts"; import * as DeviceService from "../../../device/DeviceService.ts"; import * as ServerConfig from "../../../config.ts"; @@ -36,6 +37,7 @@ import * as PreviewBrowser from "../../../preview/PreviewBrowser.ts"; const layerStubServices = Layer.mergeAll( Layer.mock(Orchestrator.OrchestratorV2)({}), + Layer.mock(TaskGraphService.TaskGraphService)({}), Layer.mock(ProjectionStore.ProjectionStoreV2)({}), Layer.mock(DeviceService.DeviceService)({}), Layer.mock(ThreadManagementService.ThreadManagementService)({}), diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts index b1abaf8d465c..0d4df7308080 100644 --- a/apps/server/src/observability/RpcInstrumentation.ts +++ b/apps/server/src/observability/RpcInstrumentation.ts @@ -80,6 +80,14 @@ const RPC_AGGREGATES = { [WS_METHODS.serverReportClientActivity]: "server", [WS_METHODS.serverReportHostPowerState]: "server", [WS_METHODS.serverGetBackgroundPolicy]: "server", + [WS_METHODS.taskGraphsSubscribe]: "taskGraphs", + [WS_METHODS.taskGraphsEdit]: "taskGraphs", + [WS_METHODS.taskGraphsRun]: "taskGraphs", + [WS_METHODS.taskGraphsCancel]: "taskGraphs", + [WS_METHODS.taskGraphPeersSubscribe]: "taskGraphs", + [WS_METHODS.taskGraphPeersAdd]: "taskGraphs", + [WS_METHODS.taskGraphPeersRemove]: "taskGraphs", + [WS_METHODS.taskGraphPeersSetWeight]: "taskGraphs", [WS_METHODS.scheduledTasksList]: "scheduledTasks", [WS_METHODS.scheduledTasksSubscribe]: "scheduledTasks", [WS_METHODS.scheduledTasksUpsert]: "scheduledTasks", diff --git a/apps/server/src/orchestration-v2/runtimeLayer.ts b/apps/server/src/orchestration-v2/runtimeLayer.ts index 2442a4fa010c..a1a256ad63ae 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.ts @@ -50,6 +50,8 @@ import * as ThreadLifecycleService from "./ThreadLifecycleService.ts"; import * as ThreadForkService from "./ThreadForkService.ts"; import * as TurnItemPositionStore from "./TurnItemPositionStore.ts"; import * as ScheduledTaskService from "../scheduledTasks/ScheduledTaskService.ts"; +import * as TaskGraphPeers from "../taskGraph/TaskGraphPeers.ts"; +import * as TaskGraphService from "../taskGraph/TaskGraphService.ts"; import * as SecretRequests from "../secrets/SecretRequests.ts"; /** The shared application event log and its command receipts. */ @@ -275,6 +277,18 @@ const layerScheduledTaskProvided = ScheduledTaskService.layer.pipe( ), ), ); +const layerTaskGraphPeersProvided = TaskGraphPeers.layer.pipe(Layer.provide(layerProjectService)); +const layerTaskGraphProvided = TaskGraphService.layer.pipe( + Layer.provide( + Layer.mergeAll( + layerTaskGraphPeersProvided, + layerThreadLaunchProvided, + layerThreadManagementProvided, + layerOrchestratorProvided, + layerProjectService, + ), + ), +); const layerProviderContinuationWorkerProvided = ProviderContinuationService.layer.pipe( Layer.provide( Layer.mergeAll( @@ -348,6 +362,8 @@ export const layerProduction = Layer.mergeAll( layerThreadLaunchProvided, layerThreadLifecycleProvided, layerScheduledTaskProvided, + layerTaskGraphProvided, + layerTaskGraphPeersProvided, layerSecretRequestsProvided, UsageLimitRecoveryWorker.layer.pipe( Layer.provide(Layer.mergeAll(ProjectionStore.layer, layerThreadManagementProvided)), diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 35c1a3fdce4b..9c0e0a74c616 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -74,6 +74,7 @@ import Migration0057 from "./Migrations/057_ScheduledTaskWebhooks.ts"; import Migration0058 from "./Migrations/058_WebhookRelayDeliveries.ts"; import Migration0059 from "./Migrations/059_McpAppModelContext.ts"; import Migration0060 from "./Migrations/060_ThreadSnapshotWindowIndexes.ts"; +import Migration0061 from "./Migrations/061_TaskGraphs.ts"; /** * Migration loader with all migrations defined inline. @@ -148,6 +149,7 @@ export const migrationEntries = [ [58, "WebhookRelayDeliveries", Migration0058], [59, "McpAppModelContext", Migration0059], [60, "ThreadSnapshotWindowIndexes", Migration0060], + [61, "TaskGraphs", Migration0061], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/061_TaskGraphs.ts b/apps/server/src/persistence/Migrations/061_TaskGraphs.ts new file mode 100644 index 000000000000..ca07eef0126c --- /dev/null +++ b/apps/server/src/persistence/Migrations/061_TaskGraphs.ts @@ -0,0 +1,41 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + + // A graph is written whole on every change; its nodes live in graph_json. + // Graphs are small (at most 48 nodes) and always read as a unit. + yield* sql` + CREATE TABLE IF NOT EXISTS task_graphs ( + graph_id TEXT PRIMARY KEY, + project_id TEXT NOT NULL, + thread_id TEXT NOT NULL, + status TEXT NOT NULL, + graph_json TEXT NOT NULL, + created_at TEXT NOT NULL, + updated_at TEXT NOT NULL + ) + `; + yield* sql` + CREATE INDEX IF NOT EXISTS idx_task_graphs_thread + ON task_graphs(thread_id, created_at) + `; + yield* sql` + CREATE INDEX IF NOT EXISTS idx_task_graphs_running + ON task_graphs(status) + WHERE status = 'running' + `; + + // Peer machines this server may place nodes on. The session token for each + // lives in the secret store, never in this table. + yield* sql` + CREATE TABLE IF NOT EXISTS task_graph_peers ( + environment_id TEXT PRIMARY KEY, + label TEXT NOT NULL, + http_base_url TEXT NOT NULL, + weight INTEGER NOT NULL, + added_at TEXT NOT NULL + ) + `; +}); diff --git a/apps/server/src/taskGraph/TaskGraphPeers.ts b/apps/server/src/taskGraph/TaskGraphPeers.ts new file mode 100644 index 000000000000..cfcd661d5faa --- /dev/null +++ b/apps/server/src/taskGraph/TaskGraphPeers.ts @@ -0,0 +1,727 @@ +import * as NodeSocket from "@effect/platform-node/NodeSocket"; +import { + CommandId, + EnvironmentId, + ORCHESTRATION_V2_WS_METHODS, + repositoryGroupingKeyOf, + TaskGraphError, + WS_METHODS, + WsRpcGroup, + type HostResourcesSnapshot, + type ModelSelection, + type OrchestrationProjectShell, + type OrchestrationV2ShellStreamItem, + type OrchestrationV2ThreadShell, + type ProjectId, + type RuntimeMode, + type TaskGraph, + type TaskGraphNode, + type TaskGraphPeer, + type TaskGraphPeerListResult, + type ThreadId, +} from "@t3tools/contracts"; +import { resolveRemotePairingTarget } from "@t3tools/shared/remote"; +import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as FiberMap from "effect/FiberMap"; +import * as FetchHttpClient from "effect/http/FetchHttpClient"; +import * as HttpClient from "effect/http/HttpClient"; +import * as HttpClientRequest from "effect/http/HttpClientRequest"; +import * as Layer from "effect/Layer"; +import * as Option from "effect/Option"; +import * as PubSub from "effect/PubSub"; +import * as Queue from "effect/Queue"; +import * as Schedule from "effect/Schedule"; +import * as Schema from "effect/Schema"; +import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/sql/SqlClient"; +import * as RpcClient from "effect/rpc/RpcClient"; +import * as RpcSerialization from "effect/rpc/RpcSerialization"; +import * as Socket from "effect/socket/Socket"; + +import * as ServerSecretStore from "../auth/ServerSecretStore.ts"; +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import type { NodeOutcome } from "./TaskGraphService.ts"; + +/** How a finished peer node's branch leaves the peer: always pushed, plus a PR at a branch end. */ +export type PeerDelivery = "commit_push" | "commit_push_pr"; + +export interface PeerNodeStart { + readonly environmentId: EnvironmentId; + readonly graph: TaskGraph; + readonly node: TaskGraphNode; + readonly threadId: ThreadId; + readonly baseRef: string; + readonly prompt: string; + readonly modelSelection: ModelSelection; + readonly runtimeMode: RuntimeMode; + readonly delivery: PeerDelivery; +} + +export interface PeerCandidate { + readonly environmentId: EnvironmentId; + readonly resources: HostResourcesSnapshot | null; + readonly receivedAt: number; + readonly weight: number; +} + +/** + * Other machines this environment may run task graph nodes on. The graph + * service asks for placement candidates, hands over nodes placed on a peer, + * and hears back through `completions` when one ends. + */ +export class TaskGraphPeers extends Context.Service< + TaskGraphPeers, + { + /** Whether any peer is paired; branches are then pushed so other machines can reach them. */ + readonly hasPeers: Effect.Effect; + /** Connected peers hosting the same repository, with fresh load, under `maxNodesPerPeer`. */ + readonly candidates: (input: { + readonly projectId: ProjectId; + readonly maxNodesPerPeer: number; + }) => Effect.Effect>; + readonly startNode: (input: PeerNodeStart) => Effect.Effect; + /** Watches a node started before a restart. */ + readonly track: (input: { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + readonly delivery: PeerDelivery; + }) => Effect.Effect; + readonly interruptNode: (input: { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; + }) => Effect.Effect; + readonly completions: Stream.Stream<{ + readonly threadId: ThreadId; + readonly outcome: NodeOutcome; + }>; + readonly subscribe: Stream.Stream; + readonly add: (input: { + readonly pairingUrl: string; + readonly label?: string | undefined; + }) => Effect.Effect; + readonly remove: ( + environmentId: EnvironmentId, + ) => Effect.Effect; + readonly setWeight: ( + environmentId: EnvironmentId, + weight: number, + ) => Effect.Effect; + } +>()("t3/taskGraph/TaskGraphPeers") {} + +/** No peers: every node runs on this machine. Used by tests. */ +export const layerNone = Layer.succeed( + TaskGraphPeers, + TaskGraphPeers.of({ + hasPeers: Effect.succeed(false), + candidates: () => Effect.succeed([]), + startNode: () => Effect.fail(new TaskGraphError({ message: "No peer machines are paired." })), + track: () => Effect.void, + interruptNode: () => Effect.void, + completions: Stream.never, + subscribe: Stream.make({ peers: [] }), + add: () => Effect.fail(new TaskGraphError({ message: "Peers are unavailable." })), + remove: () => Effect.succeed({ peers: [] }), + setWeight: () => Effect.succeed({ peers: [] }), + }), +); + +/** + * What this server asks a peer for. It can run, watch and stop threads and + * push their branches; it cannot read files, open terminals, change settings + * or manage access. + */ +export const PEER_SCOPES = ["orchestration:read", "orchestration:operate", "source-control:write"]; + +const DEFAULT_PEER_WEIGHT = 50; +const tokenSecretName = (environmentId: string) => `task-graph-peer-token:${environmentId}`; + +const peerError = (message: string, cause?: unknown) => + new TaskGraphError({ message, ...(cause === undefined ? {} : { cause }) }); + +const TERMINAL_THREAD_STATUSES = new Set([ + "completed", + "failed", + "cancelled", + "interrupted", + "rolled_back", +]); + +/** The outcome a finished peer thread implies, before delivery; null while it still runs. */ +export function peerThreadOutcome( + thread: Pick, +): "succeeded" | "failed" | "cancelled" | null { + if (thread.activeRunId !== null || !TERMINAL_THREAD_STATUSES.has(thread.status)) return null; + if (thread.status === "completed") return "succeeded"; + return thread.status === "cancelled" || thread.status === "interrupted" ? "cancelled" : "failed"; +} + +const EnvironmentDescriptorJson = Schema.Struct({ + environmentId: EnvironmentId, + label: Schema.String, +}); +const TokenResponse = Schema.Struct({ + access_token: Schema.String, + token_type: Schema.String, +}); +const decodeDescriptor = Schema.decodeUnknownEffect(EnvironmentDescriptorJson); +const decodeToken = Schema.decodeUnknownEffect(TokenResponse); + +const toWsUrl = (httpBaseUrl: string) => { + const url = new URL("/ws", httpBaseUrl); + url.protocol = url.protocol === "https:" ? "wss:" : "ws:"; + url.searchParams.set("orchestrationProtocol", "2"); + return url.toString(); +}; + +const protocolLayer = (httpBaseUrl: string, token: string) => + RpcClient.layerProtocolSocket().pipe( + Layer.provide( + Socket.layerWebSocket(toWsUrl(httpBaseUrl)).pipe( + Layer.provide( + Layer.succeed( + Socket.WebSocketConstructor, + (url, protocols) => + new NodeSocket.NodeWS.WebSocket(url, protocols as string | string[] | undefined, { + headers: { authorization: `Bearer ${token}` }, + }) as unknown as globalThis.WebSocket, + ), + ), + ), + ), + Layer.provide(RpcSerialization.layerJson), + ); + +const makeClient = RpcClient.make(WsRpcGroup); +type PeerClient = Effect.Success; + +interface PeerRow { + readonly environment_id: string; + readonly label: string; + readonly http_base_url: string; + readonly weight: number; + readonly added_at: string; +} + +interface PeerLive { + status: TaskGraphPeer["status"]; + error: string | null; + client: PeerClient | null; + readonly projects: Map; + readonly threads: Map; +} + +interface Tracked { + readonly environmentId: EnvironmentId; + readonly delivery: PeerDelivery; + delivering: boolean; +} + +const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const secrets = yield* ServerSecretStore.ServerSecretStore; + const environment = yield* ServerEnvironment.ServerEnvironment; + const projects = yield* ProjectService.ProjectService; + const localEnvironmentId = yield* environment.getEnvironmentId; + const http = (yield* HttpClient.HttpClient).pipe(HttpClient.filterStatusOk); + + /** Reads a JSON body from a peer during pairing; any failure means the link is unusable. */ + const fetchJson = (request: HttpClientRequest.HttpClientRequest) => + http.execute(request).pipe( + Effect.flatMap((response) => response.json), + Effect.timeout("10 seconds"), + Effect.mapError((cause) => + peerError(`Could not reach ${new URL(request.url).origin}.`, cause), + ), + ); + + const live = new Map(); + const tracked = new Map(); + const connections = yield* FiberMap.make(); + const changes = yield* PubSub.sliding(1); + const completions = yield* Queue.unbounded<{ threadId: ThreadId; outcome: NodeOutcome }>(); + const notify = PubSub.publish(changes, undefined).pipe(Effect.asVoid); + + const readRows = sql` + SELECT environment_id, label, http_base_url, weight, added_at + FROM task_graph_peers ORDER BY added_at + `.pipe(Effect.mapError((cause) => peerError("Could not read peer machines.", cause))); + + const list = readRows.pipe( + Effect.map((rows) => ({ + peers: rows.map((row): TaskGraphPeer => { + const state = live.get(row.environment_id); + return { + environmentId: EnvironmentId.make(row.environment_id), + label: row.label, + httpBaseUrl: row.http_base_url, + weight: row.weight, + status: state?.status ?? "connecting", + error: state?.error ?? null, + addedAt: row.added_at, + }; + }), + })), + ); + + const setStatus = ( + environmentId: string, + status: TaskGraphPeer["status"], + error: string | null, + ) => + Effect.suspend(() => { + const state = live.get(environmentId); + if (state === undefined || (state.status === status && state.error === error)) + return Effect.void; + state.status = status; + state.error = error; + return notify; + }); + + // --- Remote node lifecycle ---------------------------------------------------- + + /** Commits and pushes a finished peer thread's branch there, then reports the node. */ + const finishTracked = (environmentId: string, thread: OrchestrationV2ThreadShell) => + Effect.gen(function* () { + const entry = tracked.get(thread.id); + const state = live.get(environmentId); + const outcome = peerThreadOutcome(thread); + if (entry === undefined || entry.delivering || outcome === null || state?.client == null) + return; + entry.delivering = true; + if (outcome !== "succeeded") { + tracked.delete(thread.id); + yield* Queue.offer(completions, { + threadId: thread.id, + outcome: { type: outcome, error: `The node's run on ${environmentId} ${thread.status}.` }, + }); + return; + } + const summary = + thread.latestVisibleMessage?.role === "assistant" ? thread.latestVisibleMessage.text : null; + const delivered = + thread.worktreePath === null + ? Option.none<{ readonly url: string | null }>() + : Option.some( + yield* state.client[WS_METHODS.gitRunStackedAction]({ + actionId: `task-graph:${thread.id}`, + cwd: thread.worktreePath, + action: entry.delivery, + threadId: thread.id, + projectId: thread.projectId, + }).pipe( + Stream.runCollect, + Effect.flatMap((events) => { + for (const event of events) { + if (event.kind === "action_failed") + return Effect.fail(peerError(event.message)); + if (event.kind === "action_finished") { + return Effect.succeed({ url: event.result.pr.url ?? null }); + } + } + return Effect.fail(peerError("The peer did not report how delivery went.")); + }), + ), + ); + tracked.delete(thread.id); + yield* Queue.offer(completions, { + threadId: thread.id, + outcome: { + type: "succeeded", + summary, + branch: thread.branch, + ...(entry.delivery === "commit_push_pr" && Option.isSome(delivered) + ? { pullRequestUrl: delivered.value.url } + : {}), + }, + }); + }).pipe( + Effect.catchCause((cause) => { + tracked.delete(thread.id); + return Queue.offer(completions, { + threadId: thread.id, + outcome: { + type: "failed", + error: `Could not push the node's branch from the peer: ${Cause.pretty(cause)}`, + }, + }).pipe(Effect.asVoid); + }), + ); + + const handleShellItem = ( + environmentId: string, + state: PeerLive, + item: OrchestrationV2ShellStreamItem, + ) => + Effect.gen(function* () { + const changed: OrchestrationV2ThreadShell[] = []; + switch (item.kind) { + case "snapshot": + state.projects.clear(); + state.threads.clear(); + for (const project of item.snapshot.projects) state.projects.set(project.id, project); + for (const thread of item.snapshot.threads) { + state.threads.set(thread.id, thread); + changed.push(thread); + } + break; + case "project.updated": + state.projects.set(item.project.id, item.project); + break; + case "project.removed": + state.projects.delete(item.projectId); + break; + case "thread.updated": + state.threads.set(item.thread.id, item.thread); + changed.push(item.thread); + break; + case "thread.removed": { + state.threads.delete(item.threadId); + const entry = tracked.get(item.threadId); + if (entry !== undefined && entry.environmentId === environmentId) { + tracked.delete(item.threadId); + yield* Queue.offer(completions, { + threadId: item.threadId, + outcome: { type: "failed", error: "The node's thread was deleted on the peer." }, + }); + } + break; + } + default: + break; + } + for (const thread of changed) { + if (tracked.get(thread.id)?.environmentId === environmentId) { + // Delivery waits on the peer's git work; do not hold up the shell stream. + yield* finishTracked(environmentId, thread).pipe(Effect.forkDetach); + } + } + }); + + /** One connection to a peer, until it drops. */ + const connectOnce = (row: PeerRow, token: string, state: PeerLive) => + Effect.gen(function* () { + const client = yield* makeClient; + const config = yield* client[WS_METHODS.serverGetConfig]({}); + if (config.environment.environmentId !== row.environment_id) { + return yield* peerError( + `The machine at ${row.http_base_url} is now a different environment. Pair it again.`, + ); + } + state.client = client; + yield* setStatus(row.environment_id, "connected", null); + yield* client[ORCHESTRATION_V2_WS_METHODS.subscribeShell]({}).pipe( + Stream.runForEach((item) => handleShellItem(row.environment_id, state, item)), + ); + }).pipe( + Effect.scoped, + Effect.provide(protocolLayer(row.http_base_url, token)), + Effect.ensuring(Effect.sync(() => (state.client = null))), + ); + + const connect = (row: PeerRow) => + Effect.gen(function* () { + const token = yield* secrets.get(tokenSecretName(row.environment_id)); + if (Option.isNone(token)) { + return yield* setStatus( + row.environment_id, + "unauthorized", + "The peer's credential is missing. Pair it again.", + ); + } + const state = live.get(row.environment_id)!; + const tokenText = new TextDecoder().decode(token.value); + yield* connectOnce(row, tokenText, state).pipe( + Effect.catchCause((cause) => { + const message = Cause.pretty(cause); + return setStatus( + row.environment_id, + /\b401\b|unauthori[sz]ed/i.test(message) ? "unauthorized" : "unreachable", + message.split("\n")[0] ?? message, + ); + }), + Effect.repeat(Schedule.spaced("15 seconds")), + ); + }).pipe(Effect.ignoreCause); + + const startConnection = (row: PeerRow) => + Effect.gen(function* () { + live.set(row.environment_id, { + status: "connecting", + error: null, + client: null, + projects: new Map(), + threads: new Map(), + }); + yield* FiberMap.run(connections, row.environment_id, connect(row)); + }); + + yield* readRows.pipe( + Effect.flatMap((rows) => Effect.forEach(rows, startConnection, { discard: true })), + Effect.catchCause((cause) => + Effect.logWarning("Could not start task graph peer connections", { + cause: Cause.pretty(cause), + }), + ), + ); + + // --- Service ------------------------------------------------------------------ + + const repositoryKeyOf = (projectId: ProjectId) => + projects.getShell(projectId).pipe( + Effect.map((shell) => + Option.flatMap(shell, (project) => + Option.fromNullishOr(project.repositoryIdentity).pipe( + Option.map(repositoryGroupingKeyOf), + ), + ), + ), + Effect.orElseSucceed(() => Option.none()), + ); + + const peerProjectFor = (state: PeerLive, key: string) => + [...state.projects.values()].find( + (project) => + project.repositoryIdentity != null && + repositoryGroupingKeyOf(project.repositoryIdentity) === key, + ); + + const candidates: TaskGraphPeers["Service"]["candidates"] = ({ projectId, maxNodesPerPeer }) => + Effect.gen(function* () { + const key = yield* repositoryKeyOf(projectId); + if (Option.isNone(key)) return []; + const rows = yield* readRows.pipe(Effect.orElseSucceed(() => [])); + const usable = rows.flatMap((row) => { + const state = live.get(row.environment_id); + const running = [...tracked.values()].filter( + (entry) => entry.environmentId === row.environment_id, + ).length; + return state?.client != null && + row.weight > 0 && + running < maxNodesPerPeer && + peerProjectFor(state, key.value) !== undefined + ? [{ row, client: state.client }] + : []; + }); + return yield* Effect.forEach( + usable, + ({ row, client }) => + client[WS_METHODS.serverGetHostResources]({}).pipe( + Effect.timeout("5 seconds"), + Effect.orElseSucceed(() => null), + Effect.flatMap((resources) => + Clock.currentTimeMillis.pipe( + Effect.map((receivedAt) => ({ + environmentId: EnvironmentId.make(row.environment_id), + resources, + receivedAt, + weight: row.weight, + })), + ), + ), + ), + { concurrency: "unbounded" }, + ); + }); + + const startNode: TaskGraphPeers["Service"]["startNode"] = (input) => + Effect.gen(function* () { + const state = live.get(input.environmentId); + const client = state?.client; + if (state === undefined || client == null) { + return yield* peerError(`The peer ${input.environmentId} is not connected.`); + } + const key = yield* repositoryKeyOf(input.graph.projectId); + const project = Option.isNone(key) ? undefined : peerProjectFor(state, key.value); + if (project === undefined) { + return yield* peerError("The peer has no project for this repository."); + } + tracked.set(input.threadId, { + environmentId: input.environmentId, + delivery: input.delivery, + delivering: false, + }); + yield* client[ORCHESTRATION_V2_WS_METHODS.launchThread]({ + commandId: CommandId.make(`task-graph-launch:${input.threadId}`), + creationSource: "server", + threadId: input.threadId, + reuseExistingThread: true, + projectId: project.id, + title: input.node.title, + modelSelection: input.modelSelection, + runtimeMode: input.runtimeMode, + interactionMode: "default", + // Dependency branches were pushed by whichever machine ran them. + workspaceStrategy: { type: "worktree", baseRef: input.baseRef, startFromOrigin: true }, + initialMessage: { text: input.prompt, attachments: [] }, + }).pipe( + Effect.tapError(() => Effect.sync(() => tracked.delete(input.threadId))), + Effect.mapError((cause) => peerError("The peer could not start the node.", cause)), + ); + }); + + const track: TaskGraphPeers["Service"]["track"] = ({ environmentId, threadId, delivery }) => + Effect.gen(function* () { + tracked.set(threadId, { environmentId, delivery, delivering: false }); + const thread = live.get(environmentId)?.threads.get(threadId); + if (thread !== undefined) yield* finishTracked(environmentId, thread).pipe(Effect.forkDetach); + }); + + const interruptNode: TaskGraphPeers["Service"]["interruptNode"] = ({ environmentId, threadId }) => + Effect.gen(function* () { + tracked.delete(threadId); + const state = live.get(environmentId); + const runId = state?.threads.get(threadId)?.activeRunId; + if (state?.client == null || runId == null) return; + yield* state.client[ORCHESTRATION_V2_WS_METHODS.dispatchCommand]({ + type: "run.interrupt", + commandId: CommandId.make(`task-graph-stop:${threadId}`), + threadId, + runId, + reason: "The task graph branch was cancelled.", + }); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Could not stop a peer task graph node", { + environmentId, + threadId, + cause: Cause.pretty(cause), + }), + ), + ); + + const add: TaskGraphPeers["Service"]["add"] = (input) => + Effect.gen(function* () { + const target = yield* Effect.try({ + try: () => resolveRemotePairingTarget({ pairingUrl: input.pairingUrl }), + catch: (cause) => peerError("That is not a pairing link.", cause), + }); + const descriptor = yield* fetchJson( + HttpClientRequest.get( + new URL("/.well-known/t3/environment", target.httpBaseUrl).toString(), + ), + ).pipe( + Effect.flatMap((body) => + decodeDescriptor(body).pipe( + Effect.mapError((cause) => + peerError("The link does not point at a T3 Code server.", cause), + ), + ), + ), + ); + if (descriptor.environmentId === localEnvironmentId) { + return yield* peerError("That link points at this machine."); + } + const label = input.label ?? (descriptor.label.trim() || target.httpBaseUrl); + const token = yield* fetchJson( + HttpClientRequest.post(new URL("/oauth/token", target.httpBaseUrl).toString()).pipe( + HttpClientRequest.bodyUrlParams({ + grant_type: "urn:ietf:params:oauth:grant-type:token-exchange", + subject_token: target.credential, + subject_token_type: "urn:t3:params:oauth:token-type:environment-bootstrap", + requested_token_type: "urn:ietf:params:oauth:token-type:access_token", + scope: PEER_SCOPES.join(" "), + client_label: "T3 Code task graphs", + client_device_type: "bot", + }), + ), + ).pipe( + Effect.flatMap((body) => + decodeToken(body).pipe( + Effect.mapError((cause) => peerError("The peer refused the link.", cause)), + ), + ), + ); + if (token.token_type.toLowerCase() !== "bearer") { + return yield* peerError("The peer issued a credential this server cannot use."); + } + yield* secrets + .set( + tokenSecretName(descriptor.environmentId), + new TextEncoder().encode(token.access_token), + ) + .pipe( + Effect.mapError((cause) => peerError("Could not store the peer's credential.", cause)), + ); + const addedAt = DateTime.formatIso(yield* DateTime.now); + const row: PeerRow = { + environment_id: descriptor.environmentId, + label, + http_base_url: target.httpBaseUrl, + weight: DEFAULT_PEER_WEIGHT, + added_at: addedAt, + }; + yield* sql` + INSERT INTO task_graph_peers (environment_id, label, http_base_url, weight, added_at) + VALUES (${row.environment_id}, ${row.label}, ${row.http_base_url}, ${row.weight}, ${row.added_at}) + ON CONFLICT (environment_id) DO UPDATE SET + label = excluded.label, + http_base_url = excluded.http_base_url + `.pipe(Effect.mapError((cause) => peerError("Could not save the peer.", cause))); + yield* startConnection(row); + yield* notify; + return yield* list; + }); + + const remove: TaskGraphPeers["Service"]["remove"] = (environmentId) => + Effect.gen(function* () { + yield* FiberMap.remove(connections, environmentId); + live.delete(environmentId); + yield* sql`DELETE FROM task_graph_peers WHERE environment_id = ${environmentId}`.pipe( + Effect.mapError((cause) => peerError("Could not remove the peer.", cause)), + ); + // Forgetting the token is what unpairs on this side; the peer lists the + // session under its connected clients until someone revokes it there. + yield* secrets.remove(tokenSecretName(environmentId)).pipe(Effect.ignore); + for (const [threadId, entry] of tracked) { + if (entry.environmentId !== environmentId) continue; + tracked.delete(threadId); + yield* Queue.offer(completions, { + threadId, + outcome: { type: "failed", error: "The peer running this node was removed." }, + }); + } + yield* notify; + return yield* list; + }); + + const setWeight: TaskGraphPeers["Service"]["setWeight"] = (environmentId, weight) => + sql`UPDATE task_graph_peers SET weight = ${weight} WHERE environment_id = ${environmentId}`.pipe( + Effect.mapError((cause) => peerError("Could not update the peer.", cause)), + Effect.andThen(notify), + Effect.andThen(list), + ); + + const subscribe = Stream.unwrap( + Effect.gen(function* () { + const subscription = yield* PubSub.subscribe(changes); + return Stream.concat( + Stream.fromEffect(list), + Stream.fromSubscription(subscription).pipe(Stream.mapEffect(() => list)), + ); + }), + ); + + return TaskGraphPeers.of({ + hasPeers: readRows.pipe( + Effect.map((rows) => rows.length > 0), + Effect.orElseSucceed(() => false), + ), + candidates, + startNode, + track, + interruptNode, + completions: Stream.fromQueue(completions), + subscribe, + add, + remove, + setWeight, + }); +}); + +export const layer = Layer.effect(TaskGraphPeers, make).pipe(Layer.provide(FetchHttpClient.layer)); diff --git a/apps/server/src/taskGraph/TaskGraphService.test.ts b/apps/server/src/taskGraph/TaskGraphService.test.ts new file mode 100644 index 000000000000..fc68bbc3bc78 --- /dev/null +++ b/apps/server/src/taskGraph/TaskGraphService.test.ts @@ -0,0 +1,341 @@ +import * as NodeCrypto from "@effect/platform-node/NodeCrypto"; +import { assert, it } from "@effect/vitest"; +import { + EnvironmentId, + type GitRunStackedActionInput, + type OrchestrationV2DomainEvent, + type OrchestrationV2ThreadProjection, + ThreadId, +} from "@t3tools/contracts"; +import * as Deferred from "effect/Deferred"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as Queue from "effect/Queue"; +import * as Stream from "effect/Stream"; + +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as GitWorkflow from "../git/GitWorkflowService.ts"; +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as ThreadLaunch from "../orchestration-v2/ThreadLaunchService.ts"; +import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; +import * as SqlitePersistence from "../persistence/Sqlite.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import * as HostResources from "../resourceTelemetry/HostResources.ts"; +import * as Scheduler from "../scheduling/Scheduler.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as TaskGraphPeers from "./TaskGraphPeers.ts"; +import * as TaskGraphService from "./TaskGraphService.ts"; + +const PARENT = ThreadId.make("thread-parent"); + +interface Harness { + readonly launches: Queue.Queue; + readonly gitActions: Array; + readonly interrupted: Array; + readonly report: Deferred.Deferred; + /** Marks a node thread's run finished with a reply on a branch, then emits the run event. */ + readonly finish: ( + launch: ThreadLaunch.ThreadLaunchInput, + result: { readonly reply: string; readonly branch: string }, + ) => Effect.Effect; +} + +const projection = (input: { + readonly id: ThreadId; + readonly branch: string | null; + readonly worktreePath: string | null; + readonly runStatus?: string; + readonly reply?: string; +}) => + ({ + thread: { + id: input.id, + projectId: "project-1", + modelSelection: { instanceId: "codex", model: "gpt-5" }, + runtimeMode: "full-access", + branch: input.branch, + worktreePath: input.worktreePath, + }, + runs: input.runStatus === undefined ? [] : [{ status: input.runStatus, ordinal: 1 }], + messages: input.reply === undefined ? [] : [{ role: "assistant", text: input.reply }], + }) as unknown as OrchestrationV2ThreadProjection; + +/** Builds the service against in-memory SQLite and recording fakes for everything it drives. */ +const withService = ( + body: ( + service: TaskGraphService.TaskGraphService["Service"], + harness: Harness, + ) => Effect.Effect, + peers: Layer.Layer = TaskGraphPeers.layerNone, +) => + Effect.gen(function* () { + const launches = yield* Queue.unbounded(); + const events = yield* Queue.unbounded(); + const report = yield* Deferred.make(); + const projections = new Map([ + [PARENT, projection({ id: PARENT, branch: "main", worktreePath: null })], + ]); + const gitActions: Array = []; + const interrupted: Array = []; + + const harness: Harness = { + launches, + gitActions, + interrupted, + report, + finish: (launch, result) => + Effect.gen(function* () { + const threadId = launch.threadId!; + projections.set( + threadId, + projection({ + id: threadId, + branch: result.branch, + worktreePath: `/worktrees/${result.branch}`, + runStatus: "completed", + reply: result.reply, + }), + ); + yield* Queue.offer(events, { + type: "run.updated", + threadId, + payload: { status: "completed" }, + } as unknown as OrchestrationV2DomainEvent); + }), + }; + + const layer = TaskGraphService.layer.pipe( + Layer.provide( + Layer.mergeAll( + Layer.mock(ThreadLaunch.ThreadLaunchService)({ + launch: (input) => + Queue.offer(launches, input).pipe( + Effect.as({ + threadId: input.threadId!, + projection: projection({ id: input.threadId!, branch: null, worktreePath: null }), + resumed: false, + }), + ), + }), + Layer.mock(ThreadManagement.ThreadManagementService)({ + getThreadProjection: (threadId) => + Effect.suspend(() => { + const found = projections.get(threadId); + return found === undefined + ? Effect.die(`no thread ${threadId}`) + : Effect.succeed(found); + }), + streamDomainEvents: Stream.fromQueue(events), + sendToThread: (input) => + Deferred.succeed(report, input.text).pipe(Effect.as({} as never)), + interruptThread: (input) => + Effect.sync(() => { + interrupted.push(input.threadId); + return { type: "no_active_run" as const }; + }), + }), + Layer.mock(GitWorkflow.GitWorkflowService)({ + runStackedAction: (input) => + Effect.sync(() => { + gitActions.push(input); + return { + pr: + input.action === "commit_push_pr" + ? { status: "created", url: `https://pr/${input.cwd}` } + : {}, + } as never; + }), + }), + Layer.mock(HostResources.HostResources)({ + read: Effect.succeed({ + sampledAt: 0, + cpuUtilization: 0.1, + cpuCount: 8, + availableMemoryBytes: 8e9, + totalMemoryBytes: 16e9, + }), + }), + Layer.mock(ServerEnvironment.ServerEnvironment)({ + getEnvironmentId: Effect.succeed(EnvironmentId.make("env-local")), + }), + // Linking the created PR is best effort; these fakes make it log and move on. + Layer.mock(Orchestrator.OrchestratorV2)({}), + Layer.mock(ProjectService.ProjectService)({}), + ServerSettings.layerTest({}), + peers, + Scheduler.layer, + NodeCrypto.layer, + ), + ), + ); + return yield* TaskGraphService.TaskGraphService.pipe( + Effect.flatMap((service) => body(service, harness)), + Effect.provide(layer), + ); + }).pipe(Effect.provide(SqlitePersistence.layerMemory)); + +const node = (key: string, dependsOn: ReadonlyArray = []) => ({ + key, + title: key, + prompt: `Do ${key}.`, + dependsOn, +}); + +it.effect("fans out, merges, opens a PR at the end, and reports to the proposing thread", () => + withService((service, harness) => + Effect.gen(function* () { + yield* service.create({ + threadId: PARENT, + title: "Security audit", + nodes: [node("a"), node("b", ["a"]), node("c", ["a"]), node("d", ["b", "c"])], + run: true, + }); + + const a = yield* Queue.take(harness.launches); + assert.equal(a.title, "a"); + assert.deepEqual(a.workspaceStrategy, { type: "worktree", baseRef: "main" }); + yield* harness.finish(a, { reply: "a found two issues", branch: "t3/a" }); + + // Both branches of the fan-out start from a's branch once a succeeds. + const fanOut = [yield* Queue.take(harness.launches), yield* Queue.take(harness.launches)]; + assert.deepEqual(fanOut.map((launch) => launch.title).toSorted(), ["b", "c"]); + for (const launch of fanOut) { + assert.deepEqual(launch.workspaceStrategy, { type: "worktree", baseRef: "t3/a" }); + assert.include(launch.initialMessage!.text, "a found two issues"); + yield* harness.finish(launch, { + reply: `${launch.title} fixed`, + branch: `t3/${launch.title}`, + }); + } + + const d = yield* Queue.take(harness.launches); + assert.equal(d.title, "d"); + assert.deepEqual(d.workspaceStrategy, { type: "worktree", baseRef: "t3/b" }); + assert.include(d.initialMessage!.text, "- t3/c (from 'c')"); + assert.include(d.initialMessage!.text, "b fixed"); + assert.include(d.initialMessage!.text, "c fixed"); + yield* harness.finish(d, { reply: "merged", branch: "t3/d" }); + + const report = yield* Deferred.await(harness.report); + assert.include(report, 'Task graph "Security audit" finished: succeeded.'); + assert.include(report, "PR: https://pr//worktrees/t3/d"); + // Inner nodes commit so dependents can branch from them; only the end opens a PR. + assert.deepEqual( + harness.gitActions.map((action) => [action.cwd, action.action]), + [ + ["/worktrees/t3/a", "commit"], + ...fanOut + .map((launch) => [`/worktrees/t3/${launch.title}`, "commit"]) + .toSorted(([left], [right]) => left!.localeCompare(right!)), + ["/worktrees/t3/d", "commit_push_pr"], + ].toSorted(([left], [right]) => left!.localeCompare(right!)), + ); + }), + ), +); + +it.effect("cutting off a running branch stops its thread and skips what depends on it", () => + withService((service, harness) => + Effect.gen(function* () { + const graph = yield* service.create({ + threadId: PARENT, + title: "Features", + nodes: [node("a"), node("b", ["a"])], + run: true, + }); + const a = yield* Queue.take(harness.launches); + + const edited = yield* service.edit(graph.id, [{ type: "cancel_branch", key: "a" }]); + + assert.deepEqual(harness.interrupted, [a.threadId]); + assert.deepEqual( + edited.nodes.map((graphNode) => graphNode.status), + ["cancelled", "cancelled"], + ); + assert.equal(edited.status, "cancelled"); + assert.include(yield* Deferred.await(harness.report), "finished: cancelled"); + }), + ), +); + +it.effect("a draft waits for run, and edits before it starts change what launches", () => + withService((service, harness) => + Effect.gen(function* () { + const draft = yield* service.create({ + threadId: PARENT, + title: "Plan", + nodes: [node("a")], + run: false, + }); + assert.equal(draft.status, "draft"); + assert.equal(yield* Queue.size(harness.launches), 0); + + yield* service.edit(draft.id, [ + { type: "update_node", key: "a", prompt: "Do a differently." }, + { type: "add_node", node: node("b") }, + ]); + yield* service.run(draft.id); + + const launched = [yield* Queue.take(harness.launches), yield* Queue.take(harness.launches)]; + assert.deepEqual(launched.map((launch) => launch.title).toSorted(), ["a", "b"]); + assert.include( + launched.find((launch) => launch.title === "a")!.initialMessage!.text, + "Do a differently.", + ); + }), + ), +); + +it.effect("places a node on a peer with more free capacity and pushes local branches for it", () => + Effect.gen(function* () { + const started = yield* Queue.unbounded(); + const peers = Layer.succeed( + TaskGraphPeers.TaskGraphPeers, + TaskGraphPeers.TaskGraphPeers.of({ + hasPeers: Effect.succeed(true), + // A much bigger idle machine: it wins the load-balancing score. + candidates: () => + Effect.succeed([ + { + environmentId: EnvironmentId.make("env-peer"), + resources: { + sampledAt: 0, + cpuUtilization: 0, + cpuCount: 64, + availableMemoryBytes: 100e9, + totalMemoryBytes: 128e9, + }, + // it.effect runs on a test clock that starts at 0. + receivedAt: 0, + weight: 50, + }, + ]), + startNode: (input) => Queue.offer(started, input).pipe(Effect.asVoid), + track: () => Effect.void, + interruptNode: () => Effect.void, + completions: Stream.never, + subscribe: Stream.never, + add: () => Effect.die("unused"), + remove: () => Effect.die("unused"), + setWeight: () => Effect.die("unused"), + }), + ); + yield* withService( + (service) => + Effect.gen(function* () { + yield* service.create({ + threadId: PARENT, + title: "Spread", + nodes: [node("a")], + run: true, + }); + const start = yield* Queue.take(started); + assert.equal(start.environmentId, "env-peer"); + assert.equal(start.baseRef, "main"); + // A branch end opens its PR from the peer. + assert.equal(start.delivery, "commit_push_pr"); + }), + peers, + ); + }), +); diff --git a/apps/server/src/taskGraph/TaskGraphService.ts b/apps/server/src/taskGraph/TaskGraphService.ts new file mode 100644 index 000000000000..7cb8fd139b3e --- /dev/null +++ b/apps/server/src/taskGraph/TaskGraphService.ts @@ -0,0 +1,877 @@ +import { + CommandId, + EnvironmentId, + MessageId, + TaskGraph, + TaskGraphError, + TaskGraphId, + ThreadId, + type HostResourcesSnapshot, + type TaskGraphCreateInput, + type TaskGraphEdit, + type TaskGraphListResult, + type TaskGraphNode, +} from "@t3tools/contracts"; +import { + applyTaskGraphEdits, + buildTaskGraphNodePrompt, + deriveTaskGraphStatus, + isActiveTaskGraphNodeStatus, + isTerminalTaskGraphNodeStatus, + newTaskGraphNode, + readyTaskGraphNodes, + skipUnreachableTaskGraphNodes, + taskGraphNodeOpensPullRequest, + taskGraphNodeSummary, + validateTaskGraphNodes, +} from "@t3tools/shared/taskGraph"; +import { chooseLoadBalancedEnvironment } from "@t3tools/shared/loadBalancing"; +import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; +import * as Context from "effect/Context"; +import * as Crypto from "effect/Crypto"; +import * as DateTime from "effect/DateTime"; +import * as Effect from "effect/Effect"; +import * as Exit from "effect/Exit"; +import * as Layer from "effect/Layer"; +import * as PubSub from "effect/PubSub"; +import * as Schema from "effect/Schema"; +import * as Semaphore from "effect/Semaphore"; +import * as Stream from "effect/Stream"; +import * as SqlClient from "effect/sql/SqlClient"; + +import * as ServerEnvironment from "../environment/ServerEnvironment.ts"; +import * as GitWorkflow from "../git/GitWorkflowService.ts"; +import { linkCreatedPullRequest } from "../git/linkCreatedPullRequest.ts"; +import * as Orchestrator from "../orchestration-v2/Orchestrator.ts"; +import * as ThreadLaunch from "../orchestration-v2/ThreadLaunchService.ts"; +import * as ThreadManagement from "../orchestration-v2/ThreadManagementService.ts"; +import * as ProjectService from "../project/ProjectService.ts"; +import * as HostResources from "../resourceTelemetry/HostResources.ts"; +import * as Scheduler from "../scheduling/Scheduler.ts"; +import { forkParked } from "../serverActivation.ts"; +import * as ServerSettings from "../serverSettings.ts"; +import * as TaskGraphPeers from "./TaskGraphPeers.ts"; + +/** + * Runs task graphs: plans of agent tasks with dependencies, where each node is + * its own thread in its own worktree. The orchestrator stays unaware of graphs; + * this service launches node threads through `ThreadLaunchService`, reads their + * outcome from the domain event stream, and opens a pull request per branch end. + * + * Every graph change goes through one lock and is written whole, so a crash + * leaves either the old graph or the new one. Thread launches and git work run + * outside the lock against state already recorded as `running`/`delivering`, + * and are idempotent by command id so recovery can repeat them. + */ +export class TaskGraphService extends Context.Service< + TaskGraphService, + { + readonly create: (input: TaskGraphCreateInput) => Effect.Effect; + readonly get: (graphId: TaskGraphId) => Effect.Effect; + readonly listForThread: ( + threadId: ThreadId, + ) => Effect.Effect, TaskGraphError>; + /** Emits the thread's graphs on subscribe and again after each change to one of them. */ + readonly subscribeThread: ( + threadId: ThreadId, + ) => Stream.Stream; + readonly edit: ( + graphId: TaskGraphId, + edits: ReadonlyArray, + ) => Effect.Effect; + readonly run: (graphId: TaskGraphId) => Effect.Effect; + readonly cancel: (graphId: TaskGraphId) => Effect.Effect; + /** A node that ran on a peer finished there; recorded like a local node's end. */ + readonly completeRemoteNode: (input: { + readonly threadId: ThreadId; + readonly outcome: NodeOutcome; + }) => Effect.Effect; + } +>()("t3/taskGraph/TaskGraphService") {} + +export type NodeOutcome = + | { + readonly type: "succeeded"; + readonly summary: string | null; + readonly branch: string | null; + readonly pullRequestUrl?: string | null; + } + | { readonly type: "failed" | "cancelled"; readonly error: string }; + +/** Above this the host is too busy to take another node; matches client load balancing. */ +const MAX_CPU_UTILIZATION = 0.95; +const MIN_AVAILABLE_MEMORY_FRACTION = 0.05; + +export function hostHasCapacity(snapshot: HostResourcesSnapshot): boolean { + if (snapshot.cpuUtilization !== null && snapshot.cpuUtilization >= MAX_CPU_UTILIZATION) { + return false; + } + return ( + snapshot.totalMemoryBytes <= 0 || + snapshot.availableMemoryBytes / snapshot.totalMemoryBytes > MIN_AVAILABLE_MEMORY_FRACTION + ); +} + +/** This machine's weight against its peers' (peers default to the same). */ +const LOCAL_WEIGHT = 50; + +const peerDelivery = (graph: TaskGraph, node: TaskGraphNode): TaskGraphPeers.PeerDelivery => + taskGraphNodeOpensPullRequest(graph.nodes, node) ? "commit_push_pr" : "commit_push"; + +const decodeGraphJson = Schema.decodeUnknownEffect(Schema.fromJsonString(TaskGraph)); +const encodeGraphJson = Schema.encodeSync(Schema.fromJsonString(TaskGraph)); + +interface TaskGraphRow { + readonly graph_json: string; +} + +const graphError = (message: string, graphId?: TaskGraphId, cause?: unknown) => + new TaskGraphError({ + message, + ...(graphId === undefined ? {} : { graphId }), + ...(cause === undefined ? {} : { cause }), + }); + +function errorMessage(error: unknown): string { + if (Cause.isCause(error)) return Cause.pretty(error); + if (error instanceof Error) return error.message; + return String(error); +} + +const nowIso = DateTime.now.pipe(Effect.map((now) => DateTime.formatIso(now))); + +/** The message the proposing thread receives when its graph ends. */ +export function taskGraphReport(graph: TaskGraph): string { + const lines = graph.nodes.map((node) => { + const pr = node.pullRequestResult; + const prText = + pr === null + ? "" + : pr.status === "opened" + ? ` — PR: ${pr.url ?? "opened"}` + : ` — PR failed: ${pr.error ?? "unknown error"}`; + const detail = node.status === "succeeded" ? "" : node.error === null ? "" : ` (${node.error})`; + return `- ${node.key} "${node.title}": ${node.status}${detail}${prText}`; + }); + return [`Task graph "${graph.title}" finished: ${graph.status}.`, ...lines].join("\n"); +} + +const make = Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const crypto = yield* Crypto.Crypto; + const launcher = yield* ThreadLaunch.ThreadLaunchService; + const threads = yield* ThreadManagement.ThreadManagementService; + const git = yield* GitWorkflow.GitWorkflowService; + const hostResources = yield* HostResources.HostResources; + const settings = yield* ServerSettings.ServerSettingsService; + const scheduler = yield* Scheduler.Scheduler; + const environment = yield* ServerEnvironment.ServerEnvironment; + const peers = yield* TaskGraphPeers.TaskGraphPeers; + const orchestrator = yield* Orchestrator.OrchestratorV2; + const projects = yield* ProjectService.ProjectService; + const localEnvironmentId = yield* environment.getEnvironmentId; + + const lock = yield* Semaphore.make(1); + const locked = lock.withPermits(1); + const changes = yield* PubSub.unbounded(); + /** Node threads still running locally, so run events map back to their node. */ + const nodeByThread = new Map(); + + const readGraph = (graphId: TaskGraphId) => + sql`SELECT graph_json FROM task_graphs WHERE graph_id = ${graphId}`.pipe( + Effect.flatMap((rows) => + rows[0] === undefined ? Effect.succeed(null) : decodeGraphJson(rows[0].graph_json), + ), + Effect.mapError((cause) => graphError("Could not read task graph.", graphId, cause)), + ); + + const requireGraph = (graphId: TaskGraphId) => + readGraph(graphId).pipe( + Effect.flatMap((graph) => + graph === null + ? Effect.fail(graphError("Task graph not found.", graphId)) + : Effect.succeed(graph), + ), + ); + + const readGraphs = (rows: ReadonlyArray) => + Effect.forEach(rows, (row) => decodeGraphJson(row.graph_json)).pipe( + Effect.mapError((cause) => graphError("Could not read task graphs.", undefined, cause)), + ); + + const runningGraphs = sql` + SELECT graph_json FROM task_graphs WHERE status = 'running' ORDER BY created_at + `.pipe( + Effect.mapError((cause) => graphError("Could not list task graphs.", undefined, cause)), + Effect.flatMap(readGraphs), + ); + + const indexGraph = (graph: TaskGraph) => { + for (const node of graph.nodes) { + if (node.threadId === null) continue; + if (isActiveTaskGraphNodeStatus(node.status)) { + nodeByThread.set(node.threadId, { graphId: graph.id, key: node.key }); + } else { + nodeByThread.delete(node.threadId); + } + } + }; + + /** Writes a graph after settling skips and status. Returns the stored graph. */ + const writeGraph = (graph: TaskGraph) => + Effect.gen(function* () { + const now = yield* nowIso; + const nodes = skipUnreachableTaskGraphNodes(graph.nodes, now); + const stored: TaskGraph = { + ...graph, + nodes, + status: deriveTaskGraphStatus(graph.status, nodes), + updatedAt: now, + }; + yield* sql` + INSERT INTO task_graphs (graph_id, project_id, thread_id, status, graph_json, created_at, updated_at) + VALUES (${stored.id}, ${stored.projectId}, ${stored.threadId}, ${stored.status}, + ${encodeGraphJson(stored)}, ${stored.createdAt}, ${stored.updatedAt}) + ON CONFLICT (graph_id) DO UPDATE SET + status = excluded.status, + graph_json = excluded.graph_json, + updated_at = excluded.updated_at + `.pipe(Effect.mapError((cause) => graphError("Could not save task graph.", graph.id, cause))); + indexGraph(stored); + yield* PubSub.publish(changes, stored.threadId); + return stored; + }); + + const updateNode = ( + graphId: TaskGraphId, + key: string, + update: (node: TaskGraphNode, graph: TaskGraph) => TaskGraphNode, + ) => + locked( + Effect.gen(function* () { + const graph = yield* requireGraph(graphId); + const before = graph.status; + const stored = yield* writeGraph({ + ...graph, + nodes: graph.nodes.map((node) => (node.key === key ? update(node, graph) : node)), + }); + return { stored, finished: before === "running" && stored.status !== "running" }; + }), + ).pipe(Effect.tap(({ stored, finished }) => (finished ? reportFinished(stored) : Effect.void))); + + /** Tells the proposing thread how its graph ended, queued behind any turn in progress. */ + const reportFinished = (graph: TaskGraph) => + threads + .sendToThread({ + projectId: graph.projectId, + commandId: CommandId.make(`task-graph-report:${graph.id}:${graph.updatedAt}`), + threadId: graph.threadId, + messageId: MessageId.make(`task-graph-report:${graph.id}:${graph.updatedAt}`), + text: taskGraphReport(graph), + attachments: [], + mode: "queue", + createdBy: "system", + creationSource: "server", + }) + .pipe( + Effect.asVoid, + Effect.catchCause((cause) => + Effect.logWarning("Could not report finished task graph", { + graphId: graph.id, + cause: Cause.pretty(cause), + }), + ), + ); + + // --- Finishing nodes ------------------------------------------------------- + + /** Commits the node's work so dependents can branch from it, and opens its PR. */ + const deliver = (graph: TaskGraph, node: TaskGraphNode, worktreePath: string) => + Effect.gen(function* () { + const opensPullRequest = taskGraphNodeOpensPullRequest(graph.nodes, node); + // With peers paired, a dependent may run elsewhere and needs the branch on origin. + const push = yield* peers.hasPeers; + const result = yield* git.runStackedAction({ + actionId: `task-graph:${graph.id}:${node.key}:${node.threadId}`, + cwd: worktreePath, + action: opensPullRequest ? "commit_push_pr" : push ? "commit_push" : "commit", + ...(node.threadId === null ? {} : { threadId: node.threadId }), + projectId: graph.projectId, + }); + if (opensPullRequest && node.threadId !== null) { + yield* linkCreatedPullRequest({ + threadId: node.threadId, + result, + commandId: Effect.succeed(CommandId.make(`task-graph-pr-link:${node.threadId}`)), + }).pipe( + Effect.provideService(Orchestrator.OrchestratorV2, orchestrator), + Effect.provideService(ProjectService.ProjectService, projects), + ); + } + return opensPullRequest + ? ({ status: "opened", url: result.pr.url ?? null, error: null } as const) + : null; + }); + + const finishLocalNode = (graphId: TaskGraphId, key: string, threadId: ThreadId) => + Effect.gen(function* () { + const projection = yield* threads.getThreadProjection(threadId); + // A follow-up the user sent to the node thread keeps the node running. + if (ThreadManagement.latestActiveRun(projection) !== undefined) return; + const run = ThreadManagement.latestRun(projection); + nodeByThread.delete(threadId); + if (run === undefined || run.status !== "completed") { + const status = + run?.status === "cancelled" || run?.status === "interrupted" ? "cancelled" : "failed"; + yield* updateNode(graphId, key, (node, _graph) => + node.status === "running" + ? { + ...node, + status, + error: + run === undefined + ? "The node's thread has no run." + : `The node's run ${run.status}.`, + } + : node, + ).pipe(Effect.tap(() => advance)); + return; + } + const lastReply = projection.messages.findLast( + (message) => message.role === "assistant" && message.text.trim().length > 0, + ); + const { stored } = yield* updateNode(graphId, key, (node) => + node.status === "running" + ? { + ...node, + status: "delivering", + branch: projection.thread.branch, + summary: lastReply === undefined ? null : taskGraphNodeSummary(lastReply.text), + } + : node, + ); + const node = stored.nodes.find((candidate) => candidate.key === key); + if (node === undefined || node.status !== "delivering") return; + const worktreePath = projection.thread.worktreePath; + const delivered = + worktreePath === null + ? Exit.succeed(null) + : yield* Effect.exit(deliver(stored, node, worktreePath)); + const completedAt = yield* nowIso; + yield* updateNode(graphId, key, (current) => + current.status !== "delivering" + ? current + : Exit.isSuccess(delivered) + ? { ...current, status: "succeeded", pullRequestResult: delivered.value, completedAt } + : taskGraphNodeOpensPullRequest(stored.nodes, current) + ? { + // The work itself is done; only the PR failed, so dependents still run. + ...current, + status: "succeeded", + pullRequestResult: { + status: "failed", + url: null, + error: errorMessage(delivered.cause), + }, + completedAt, + } + : { + ...current, + status: "failed", + error: `Could not commit the node's work: ${errorMessage(delivered.cause)}`, + completedAt, + }, + ); + yield* advance; + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Could not finish task graph node", { + graphId, + key, + cause: Cause.pretty(cause), + }), + ), + ); + + const completeRemoteNode: TaskGraphService["Service"]["completeRemoteNode"] = ({ + threadId, + outcome, + }) => + Effect.gen(function* () { + const target = nodeByThread.get(threadId); + if (target === undefined) return; + nodeByThread.delete(threadId); + const completedAt = yield* nowIso; + yield* updateNode(target.graphId, target.key, (node) => { + if (!isActiveTaskGraphNodeStatus(node.status)) return node; + if (outcome.type !== "succeeded") { + return { ...node, status: outcome.type, error: outcome.error, completedAt }; + } + return { + ...node, + status: "succeeded", + summary: outcome.summary === null ? null : taskGraphNodeSummary(outcome.summary), + branch: outcome.branch, + pullRequestResult: + outcome.pullRequestUrl === undefined + ? null + : { status: "opened", url: outcome.pullRequestUrl, error: null }, + completedAt, + }; + }); + yield* advance; + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Could not record remote task graph node", { + threadId, + cause: Cause.pretty(cause), + }), + ), + ); + + // --- Starting nodes -------------------------------------------------------- + + /** + * Where a ready node runs: its pinned machine, or whichever of this machine + * and its peers has the most free capacity, scored like client load balancing. + * Null when nowhere has room; the node waits for the next pass. + */ + const placeNode = ( + graph: TaskGraph, + node: TaskGraphNode, + local: { readonly resources: HostResourcesSnapshot; readonly hasRoom: boolean }, + maxNodes: number, + ): Effect.Effect => + Effect.gen(function* () { + if (node.environmentId === localEnvironmentId) { + return local.hasRoom ? localEnvironmentId : null; + } + const remote = yield* peers.candidates({ + projectId: graph.projectId, + maxNodesPerPeer: maxNodes, + }); + if (node.environmentId !== null) { + return remote.some((candidate) => candidate.environmentId === node.environmentId) + ? node.environmentId + : null; + } + if (remote.length === 0) return local.hasRoom ? localEnvironmentId : null; + const now = yield* Clock.currentTimeMillis; + const chosen = chooseLoadBalancedEnvironment( + [ + ...(local.hasRoom + ? [ + { + environmentId: localEnvironmentId, + resources: local.resources, + receivedAt: now, + weight: LOCAL_WEIGHT, + }, + ] + : []), + ...remote, + ], + now, + ); + // The scorer skips a machine without a CPU sample yet; this one can still take work. + if (chosen === null) return local.hasRoom ? localEnvironmentId : null; + return EnvironmentId.make(chosen); + }); + + const launchNode = (graph: TaskGraph, node: TaskGraphNode) => + Effect.gen(function* () { + const threadId = node.threadId!; + const parent = yield* threads.getThreadProjection(graph.threadId); + const firstDependency = graph.nodes.find((candidate) => candidate.key === node.dependsOn[0]); + const baseRef = firstDependency?.branch ?? graph.baseRef; + const prompt = buildTaskGraphNodePrompt(graph, node); + const modelSelection = node.modelSelection ?? parent.thread.modelSelection; + if (node.assignedEnvironmentId !== localEnvironmentId) { + yield* peers.startNode({ + environmentId: node.assignedEnvironmentId!, + graph, + node, + threadId, + baseRef, + prompt, + modelSelection, + runtimeMode: parent.thread.runtimeMode, + delivery: peerDelivery(graph, node), + }); + return; + } + // A dependency that ran on a peer pushed its branch; it exists here only on origin. + const fromPeer = + firstDependency !== undefined && + firstDependency.assignedEnvironmentId !== localEnvironmentId; + yield* launcher.launch({ + commandId: CommandId.make(`task-graph-launch:${threadId}`), + threadId, + reuseExistingThread: true, + projectId: graph.projectId, + title: node.title, + modelSelection, + runtimeMode: parent.thread.runtimeMode, + interactionMode: "default", + workspaceStrategy: { + type: "worktree", + baseRef, + ...(fromPeer ? { startFromOrigin: true } : {}), + }, + initialMessage: { + messageId: MessageId.make(`task-graph-message:${threadId}`), + senderThreadId: graph.threadId, + text: prompt, + attachments: [], + }, + createdBy: "agent", + creationSource: "server", + }); + }).pipe( + Effect.catchCause((cause) => + Effect.gen(function* () { + const completedAt = yield* nowIso; + nodeByThread.delete(node.threadId!); + yield* updateNode(graph.id, node.key, (current) => + current.status === "running" + ? { + ...current, + status: "failed", + error: `Could not start the node: ${errorMessage(cause)}`, + completedAt, + } + : current, + ); + }), + ), + ); + + /** + * Starts every node that is ready and fits: under the environment's + * concurrency limit, and on a machine that is not overloaded. Nodes that do + * not fit wait for the next scheduler tick or the next node to finish. + */ + const advance: Effect.Effect = Effect.gen(function* () { + const { taskGraphMaxConcurrentNodes } = yield* settings.getSettings; + const toLaunch = yield* locked( + Effect.gen(function* () { + const graphs = yield* runningGraphs; + let active = graphs + .flatMap((graph) => graph.nodes) + .filter( + (node) => + isActiveTaskGraphNodeStatus(node.status) && + node.assignedEnvironmentId === localEnvironmentId, + ).length; + const host = yield* hostResources.read; + const hostFree = hostHasCapacity(host); + const started: Array<{ graph: TaskGraph; node: TaskGraphNode }> = []; + for (const graph of graphs) { + const ready = readyTaskGraphNodes( + skipUnreachableTaskGraphNodes(graph.nodes, yield* nowIso), + ); + const assigned = new Map(); + for (const node of ready) { + const environmentId = yield* placeNode( + graph, + node, + { resources: host, hasRoom: hostFree && active < taskGraphMaxConcurrentNodes }, + taskGraphMaxConcurrentNodes, + ); + if (environmentId === null) continue; + if (environmentId === localEnvironmentId) active += 1; + assigned.set(node.key, { + ...node, + status: "running", + assignedEnvironmentId: environmentId, + threadId: ThreadId.make(`task-graph-node:${yield* crypto.randomUUIDv4}`), + startedAt: yield* nowIso, + }); + } + // Write even with nothing to start: skips may have settled the graph. + const stored = yield* writeGraph({ + ...graph, + nodes: graph.nodes.map((node) => assigned.get(node.key) ?? node), + }); + if (stored.status !== "running") yield* reportFinished(stored).pipe(Effect.forkDetach); + for (const node of assigned.values()) started.push({ graph: stored, node }); + } + return started; + }), + ); + yield* Effect.forEach(toLaunch, ({ graph, node }) => launchNode(graph, node), { + concurrency: "unbounded", + discard: true, + }); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Could not advance task graphs", { cause: Cause.pretty(cause) }), + ), + ); + + const stopNodes = (graph: TaskGraph, nodes: ReadonlyArray) => + Effect.forEach( + nodes, + (node) => { + if (node.threadId === null) return Effect.void; + nodeByThread.delete(node.threadId); + if ( + node.assignedEnvironmentId !== localEnvironmentId && + node.assignedEnvironmentId !== null + ) { + return peers.interruptNode({ + environmentId: node.assignedEnvironmentId, + threadId: node.threadId, + }); + } + return threads + .interruptThread({ + projectId: graph.projectId, + commandId: CommandId.make(`task-graph-stop:${node.threadId}`), + threadId: node.threadId, + reason: "The task graph branch was cancelled.", + }) + .pipe(Effect.asVoid); + }, + { discard: true }, + ).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Could not stop cancelled task graph nodes", { + graphId: graph.id, + cause: Cause.pretty(cause), + }), + ), + ); + + // --- Public API ------------------------------------------------------------- + + const create: TaskGraphService["Service"]["create"] = (input) => + Effect.gen(function* () { + const parent = yield* threads + .getThreadProjection(input.threadId) + .pipe( + Effect.mapError((cause) => graphError("Could not read the thread.", undefined, cause)), + ); + const nodes = input.nodes.map(newTaskGraphNode); + const invalid = validateTaskGraphNodes(nodes); + if (invalid !== null) return yield* graphError(invalid); + const { taskGraphAutoRun } = yield* settings.getSettings.pipe( + Effect.mapError((cause) => graphError("Could not read settings.", undefined, cause)), + ); + const now = yield* nowIso; + const uuid = yield* crypto.randomUUIDv4.pipe( + Effect.mapError((cause) => graphError("Could not create task graph id.", undefined, cause)), + ); + const graph = yield* locked( + writeGraph({ + id: TaskGraphId.make(`task-graph:${uuid}`), + projectId: parent.thread.projectId, + threadId: parent.thread.id, + title: input.title, + baseRef: input.baseRef ?? parent.thread.branch ?? "HEAD", + status: (input.run ?? taskGraphAutoRun) ? "running" : "draft", + nodes, + createdAt: now, + updatedAt: now, + }), + ); + if (graph.status === "running") yield* advance; + return yield* requireGraph(graph.id); + }); + + const edit: TaskGraphService["Service"]["edit"] = (graphId, edits) => + Effect.gen(function* () { + const { graph, stopped, finished } = yield* locked( + Effect.gen(function* () { + const current = yield* requireGraph(graphId); + if (current.status === "cancelled") { + return yield* graphError("This task graph was cancelled.", graphId); + } + const result = applyTaskGraphEdits(current.nodes, edits, yield* nowIso); + if (!result.ok) return yield* graphError(result.error, graphId); + const stopped = current.nodes.filter((node) => { + const next = result.nodes.find((candidate) => candidate.key === node.key); + return isActiveTaskGraphNodeStatus(node.status) && next?.status === "cancelled"; + }); + // Retrying a node in a finished graph puts the graph back to work. + const reopened = result.nodes.some((node) => node.status === "pending"); + const status = + current.status === "draft" ? "draft" : reopened ? "running" : current.status; + const graph = yield* writeGraph({ ...current, status, nodes: result.nodes }); + return { + graph, + stopped, + finished: current.status === "running" && graph.status !== "running", + }; + }), + ); + yield* stopNodes(graph, stopped); + if (finished) yield* reportFinished(graph); + if (graph.status === "running") yield* advance; + return yield* requireGraph(graphId); + }); + + const run: TaskGraphService["Service"]["run"] = (graphId) => + Effect.gen(function* () { + yield* locked( + Effect.gen(function* () { + const graph = yield* requireGraph(graphId); + if (graph.status !== "draft") { + return yield* graphError("Only a draft task graph can be run.", graphId); + } + yield* writeGraph({ ...graph, status: "running" }); + }), + ); + yield* advance; + return yield* requireGraph(graphId); + }); + + const cancel: TaskGraphService["Service"]["cancel"] = (graphId) => + Effect.gen(function* () { + const { graph, stopped } = yield* locked( + Effect.gen(function* () { + const current = yield* requireGraph(graphId); + const completedAt = yield* nowIso; + const nodes = current.nodes.map((node) => + isTerminalTaskGraphNodeStatus(node.status) + ? node + : { ...node, status: "cancelled" as const, error: "Cancelled.", completedAt }, + ); + const graph = yield* writeGraph({ ...current, status: "cancelled", nodes }); + return { + graph, + stopped: current.nodes.filter((node) => isActiveTaskGraphNodeStatus(node.status)), + }; + }), + ); + yield* stopNodes(graph, stopped); + return graph; + }); + + const listForThread: TaskGraphService["Service"]["listForThread"] = (threadId) => + sql` + SELECT graph_json FROM task_graphs WHERE thread_id = ${threadId} ORDER BY created_at + `.pipe( + Effect.mapError((cause) => graphError("Could not list task graphs.", undefined, cause)), + Effect.flatMap(readGraphs), + ); + + const subscribeThread: TaskGraphService["Service"]["subscribeThread"] = (threadId) => + Stream.unwrap( + Effect.gen(function* () { + // Subscribe before the snapshot so a change between the two is not lost. + const subscription = yield* PubSub.subscribe(changes); + return Stream.concat( + Stream.fromEffect(listForThread(threadId)), + Stream.fromSubscription(subscription).pipe( + Stream.filter((changed) => changed === threadId), + Stream.mapEffect(() => listForThread(threadId)), + ), + ).pipe(Stream.map((graphs) => ({ graphs }))); + }), + ); + + // --- Startup ---------------------------------------------------------------- + + // Rebuild the thread index, then catch up on node runs that ended while the + // server was down. Launches are idempotent by command id, so a node that was + // marked running but never launched is launched again. + yield* Effect.gen(function* () { + const graphs = yield* runningGraphs; + for (const graph of graphs) indexGraph(graph); + yield* Effect.forEach( + graphs.flatMap((graph) => + graph.nodes + .filter( + (node) => + node.status === "running" && node.assignedEnvironmentId === localEnvironmentId, + ) + .map((node) => ({ graph, node })), + ), + ({ graph, node }) => + threads.getThreadProjection(node.threadId!).pipe( + Effect.matchEffect({ + onFailure: () => launchNode(graph, node), + onSuccess: () => finishLocalNode(graph.id, node.key, node.threadId!), + }), + ), + { discard: true }, + ); + yield* Effect.forEach( + graphs.flatMap((graph) => + graph.nodes + .filter( + (node) => + isActiveTaskGraphNodeStatus(node.status) && + node.assignedEnvironmentId !== null && + node.assignedEnvironmentId !== localEnvironmentId, + ) + .map((node) => ({ graph, node })), + ), + ({ graph, node }) => + peers.track({ + environmentId: node.assignedEnvironmentId!, + threadId: node.threadId!, + delivery: peerDelivery(graph, node), + }), + { discard: true }, + ); + // A node mid-delivery when the server stopped redelivers; the git action is idempotent. + yield* Effect.forEach( + graphs.flatMap((graph) => + graph.nodes + .filter( + (node) => + node.status === "delivering" && node.assignedEnvironmentId === localEnvironmentId, + ) + .map((node) => ({ graph, node })), + ), + ({ graph, node }) => + updateNode(graph.id, node.key, (current) => ({ ...current, status: "running" })).pipe( + Effect.andThen(finishLocalNode(graph.id, node.key, node.threadId!)), + ), + { discard: true }, + ); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Could not recover task graphs", { cause: Cause.pretty(cause) }), + ), + ); + + yield* forkParked( + Stream.runForEach(threads.streamDomainEvents, (event) => { + if ( + event.type !== "run.updated" || + !ThreadManagement.isTerminalRunStatus(event.payload.status) + ) { + return Effect.void; + } + const target = nodeByThread.get(event.threadId); + return target === undefined + ? Effect.void + : finishLocalNode(target.graphId, target.key, event.threadId); + }).pipe( + Effect.catchCause((cause) => + Effect.logWarning("Task graph event stream failed", { cause: Cause.pretty(cause) }), + ), + ), + ); + yield* forkParked(Stream.runForEach(peers.completions, completeRemoteNode)); + + // Retries nodes held back by load or the concurrency limit, and picks up peer capacity. + yield* scheduler.register("task-graphs", advance); + + return TaskGraphService.of({ + create, + get: requireGraph, + listForThread, + subscribeThread, + edit, + run, + cancel, + completeRemoteNode, + }); +}); + +export const layer = Layer.effect(TaskGraphService, make); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 8e7a1210a507..d135a13c95fa 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -123,6 +123,8 @@ import * as ThreadLaunchService from "./orchestration-v2/ThreadLaunchService.ts" import * as ThreadMessageIntake from "./orchestration-v2/ThreadMessageIntake.ts"; import * as IdAllocator from "@t3tools/provider-core/server/IdAllocator"; import * as ScheduledTasks from "./scheduledTasks/ScheduledTaskService.ts"; +import * as TaskGraphPeers from "./taskGraph/TaskGraphPeers.ts"; +import * as TaskGraphs from "./taskGraph/TaskGraphService.ts"; import * as SecretRequests from "./secrets/SecretRequests.ts"; import { archivedShellStreamItemFromThreadShell, @@ -1221,6 +1223,8 @@ const layerWsRpc = ( const threadLaunch = yield* ThreadLaunchService.ThreadLaunchService; const providerSessionManager = yield* ProviderSessionManager.ProviderSessionManagerV2; const scheduledTasks = yield* ScheduledTasks.ScheduledTaskService; + const taskGraphs = yield* TaskGraphs.TaskGraphService; + const taskGraphPeers = yield* TaskGraphPeers.TaskGraphPeers; const secretRequests = yield* SecretRequests.SecretRequests; const pullRequests = yield* PullRequestService.PullRequestService; const pullRequestSync = yield* PullRequestSyncReactor.PullRequestSyncReactor; @@ -2041,6 +2045,18 @@ const layerWsRpc = ( [WS_METHODS.scheduledTasksSubscribe]: (_input) => scheduledTasks.subscribeList().pipe(Stream.map(withVisibleWebhookUrls)), [WS_METHODS.scheduledTasksUpsert]: (input) => scheduledTasks.upsert(input), + [WS_METHODS.taskGraphsSubscribe]: (input) => taskGraphs.subscribeThread(input.threadId), + [WS_METHODS.taskGraphsEdit]: (input) => + taskGraphs.edit(input.graphId, input.edits).pipe(Effect.map((graph) => ({ graph }))), + [WS_METHODS.taskGraphsRun]: (input) => + taskGraphs.run(input.graphId).pipe(Effect.map((graph) => ({ graph }))), + [WS_METHODS.taskGraphsCancel]: (input) => + taskGraphs.cancel(input.graphId).pipe(Effect.map((graph) => ({ graph }))), + [WS_METHODS.taskGraphPeersSubscribe]: (_input) => taskGraphPeers.subscribe, + [WS_METHODS.taskGraphPeersAdd]: (input) => taskGraphPeers.add(input), + [WS_METHODS.taskGraphPeersRemove]: (input) => taskGraphPeers.remove(input.environmentId), + [WS_METHODS.taskGraphPeersSetWeight]: (input) => + taskGraphPeers.setWeight(input.environmentId, input.weight), [WS_METHODS.scheduledTasksSetEnabled]: (input) => Effect.annotateCurrentSpan({ "scheduled_task.id": input.id }).pipe( Effect.andThen(scheduledTasks.setEnabled(input)), diff --git a/apps/web/package.json b/apps/web/package.json index 1d2d93ed36a9..fdc83e81b72d 100644 --- a/apps/web/package.json +++ b/apps/web/package.json @@ -15,6 +15,7 @@ "@base-ui/react": "^1.4.1", "@clerk/electron": "catalog:", "@clerk/react": "catalog:", + "@dagrejs/dagre": "^3.1.1", "@daypicker/react": "^10.0.1", "@dnd-kit/core": "^6.3.1", "@dnd-kit/modifiers": "^9.0.0", @@ -50,6 +51,7 @@ "@tiptap/pm": "^3.31.3", "@tiptap/react": "^3.31.3", "@tiptap/starter-kit": "^3.31.3", + "@xyflow/react": "^12.12.0", "class-variance-authority": "^0.7.1", "culori": "^4.0.2", "dompurify": "^3.4.16", diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index c8c202f5cadc..d2fea6b2d8cb 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -485,6 +485,7 @@ import { type QueuedRunsControlHandle, type EditQueuedRunRequest, } from "./chat/QueuedRunsControl"; +import { TaskGraphCards } from "./chat/TaskGraphCard"; import { useLinkedThreadPullRequest } from "./ThreadStatusIndicators"; import type { ComposerBannerStackItem } from "./chat/ComposerBannerStack"; import { ThreadStatusLine } from "./chat/ThreadStatusLine"; @@ -11635,25 +11636,31 @@ export default function ChatView(props: ChatViewProps) { isPreparingWorktree={isPreparingWorktree} queuedRunsControl={ isServerThread && activeThread ? ( - + <> + + + ) : null } bannerItems={composerBannerItems} diff --git a/apps/web/src/components/chat/ChatComposer.tsx b/apps/web/src/components/chat/ChatComposer.tsx index cc4031710158..1ac904eec7fb 100644 --- a/apps/web/src/components/chat/ChatComposer.tsx +++ b/apps/web/src/components/chat/ChatComposer.tsx @@ -1648,7 +1648,7 @@ export interface ChatComposerProps { onPageScrollKeyUp: (key: string) => void; onPageScrollRelease: () => void; - // Queued runs strip rendered above the composer (v2 queue/steer). + // Attachments rendered above the composer: task graphs and the queued runs strip. queuedRunsControl?: ReactNode; // Queued-message edit mode: attachments already stored on the message being // edited. Rendered in the attachment strip with a remove control; removal is diff --git a/apps/web/src/components/chat/TaskGraphCard.tsx b/apps/web/src/components/chat/TaskGraphCard.tsx new file mode 100644 index 000000000000..944204203af0 --- /dev/null +++ b/apps/web/src/components/chat/TaskGraphCard.tsx @@ -0,0 +1,310 @@ +import type { EnvironmentId, TaskGraph, TaskGraphNode, ThreadId } from "@t3tools/contracts"; +import { taskGraphLayers } from "@t3tools/shared/taskGraph"; +import { Link } from "@tanstack/react-router"; +import { + ChevronDownIcon, + Maximize2Icon, + MessageSquareTextIcon, + PlayIcon, + SquareIcon, + WorkflowIcon, +} from "lucide-react"; +import { lazy, Suspense, useId, useState } from "react"; +import { create } from "zustand"; + +import { cn } from "~/lib/utils"; +import { readLocalApi } from "~/localApi"; +import { useEnvironmentQuery } from "../../state/query"; +import { serverEnvironment } from "../../state/server"; +import { Button } from "../ui/button"; +import { PullRequestGlyph } from "../pullRequest/pullRequestIcons"; +import { Dialog, DialogPopup } from "../ui/dialog"; +import { Spinner } from "../ui/spinner"; +import { Tooltip, TooltipPopup, TooltipTrigger } from "../ui/tooltip"; +import { ComposerBanner } from "./ComposerBanner"; +import { + isTaskGraphFinished, + pullRequestLabel, + TASK_GRAPH_NODE_STATUS_DOT_CLASS, + TASK_GRAPH_NODE_STATUS_LABEL, + TASK_GRAPH_STATUS_LABEL, + taskGraphProgressLabel, + taskGraphPullRequestLinks, +} from "./taskGraphView"; +import { useTaskGraphCommands } from "./useTaskGraphCommands"; + +// The canvas, dagre and xyflow load only when someone opens the editor. +const loadTaskGraphEditor = () => import("./TaskGraphEditor"); +const TaskGraphEditor = lazy(loadTaskGraphEditor); + +const EMPTY_GRAPHS: ReadonlyArray = []; + +/** Finished graphs hidden from the chat for the rest of this session. */ +const useDismissedTaskGraphs = create<{ + readonly ids: ReadonlySet; + readonly dismiss: (id: string) => void; +}>((set) => ({ + ids: new Set(), + dismiss: (id) => set((state) => ({ ids: new Set(state.ids).add(id) })), +})); + +const openExternal = (url: string) => void readLocalApi()?.shell.openExternal(url); + +/** + * The task graphs a thread proposed, docked above its composer. Drafts and + * running graphs start expanded; finished ones collapse to a summary line and + * can be dismissed. Fed by the live subscription, so status needs no refresh. + */ +export function TaskGraphCards(props: { + readonly environmentId: EnvironmentId; + readonly threadId: ThreadId; +}) { + const graphsQuery = useEnvironmentQuery( + serverEnvironment.taskGraphsLive({ + environmentId: props.environmentId, + input: { threadId: props.threadId }, + }), + ); + const dismissed = useDismissedTaskGraphs((state) => state.ids); + const dismiss = useDismissedTaskGraphs((state) => state.dismiss); + // A load error renders nothing: servers from before task graphs reject the + // subscription, and every thread on them would otherwise carry an error row. + const graphs = (graphsQuery.data?.graphs ?? EMPTY_GRAPHS).filter( + (graph) => !(dismissed.has(graph.id) && isTaskGraphFinished(graph.status)), + ); + return graphs.map((graph) => ( + dismiss(graph.id)} + /> + )); +} + +function TaskGraphCard(props: { + readonly environmentId: EnvironmentId; + readonly graph: TaskGraph; + readonly onDismiss: () => void; +}) { + const { graph } = props; + const finished = isTaskGraphFinished(graph.status); + const [expandedOverride, setExpanded] = useState(null); + const expanded = expandedOverride ?? !finished; + const [editorOpen, setEditorOpen] = useState(false); + const commands = useTaskGraphCommands(props.environmentId, graph); + const listId = useId(); + const pullRequests = taskGraphPullRequestLinks(graph.nodes); + const layers = expanded ? taskGraphLayers(graph.nodes) : []; + + return ( + + + + + + + + {graph.title} + + + {TASK_GRAPH_STATUS_LABEL[graph.status]} · {taskGraphProgressLabel(graph)} + + {!expanded + ? pullRequests.map((pullRequest) => ( + + )) + : null} + + + {graph.status === "draft" ? ( + + ) : null} + + void loadTaskGraphEditor()} + onFocus={() => void loadTaskGraphEditor()} + onClick={() => setEditorOpen(true)} + /> + } + > + + + {graph.status === "draft" ? "Edit graph" : "Open graph"} + + {!finished ? ( + + void commands.cancel()} + /> + } + > + + + + {graph.status === "draft" ? "Discard draft" : "Cancel graph"} + + + ) : null} + + {finished ? ( + + ) : null} + + + {commands.error !== null ? ( + +

+ {commands.error} +

+ + ) : null} + + + {layers.map((layer, index) => ( +
+ {layers.length > 1 ? ( + + + Step {index + 1} + + + ) : null} + {layer.map((node) => ( + + ))} +
+ ))} + + + + + + {editorOpen ? ( + + + + } + > + + + ) : null} + + + + ); +} + +function TaskGraphNodeRow(props: { + readonly environmentId: EnvironmentId; + readonly node: TaskGraphNode; +}) { + const { node } = props; + const pullRequestUrl = node.pullRequestResult?.url ?? null; + return ( + + + + + + {node.title} + + {TASK_GRAPH_NODE_STATUS_LABEL[node.status]} + + {node.branch !== null ? ( + {node.branch} + ) : null} + + + {node.threadId !== null ? ( + + + } + /> + } + > + + + Open thread + + ) : null} + {pullRequestUrl !== null ? ( + + ) : null} + + + ); +} diff --git a/apps/web/src/components/chat/TaskGraphEditor.css b/apps/web/src/components/chat/TaskGraphEditor.css new file mode 100644 index 000000000000..dd67dba72a6f --- /dev/null +++ b/apps/web/src/components/chat/TaskGraphEditor.css @@ -0,0 +1,14 @@ +/* + * xyflow draws edges, handles, the connection line and the dot grid itself, + * so its theme variables map to ours here instead of in classes on elements + * we render. Loaded with the editor chunk only. + */ +.react-flow[data-task-graph-canvas] { + --xy-edge-stroke: var(--muted-foreground); + --xy-edge-stroke-selected: var(--ring); + --xy-edge-stroke-width: 1.5; + --xy-connectionline-stroke: var(--ring); + --xy-handle-background-color: var(--muted-foreground); + --xy-background-pattern-dots-color: var(--border); + --xy-attribution-background-color: transparent; +} diff --git a/apps/web/src/components/chat/TaskGraphEditor.tsx b/apps/web/src/components/chat/TaskGraphEditor.tsx new file mode 100644 index 000000000000..c4a8da338c6b --- /dev/null +++ b/apps/web/src/components/chat/TaskGraphEditor.tsx @@ -0,0 +1,660 @@ +import "@xyflow/react/dist/base.css"; +import "./TaskGraphEditor.css"; + +import type { EnvironmentId, TaskGraph, TaskGraphNode } from "@t3tools/contracts"; +import { + isActiveTaskGraphNodeStatus, + isTerminalTaskGraphNodeStatus, + taskGraphNodeOpensPullRequest, +} from "@t3tools/shared/taskGraph"; +import { Link } from "@tanstack/react-router"; +import { + Background, + Handle, + Position, + ReactFlow, + useReactFlow, + type Connection, + type NodeProps, +} from "@xyflow/react"; +import { + MessageSquareTextIcon, + PlayIcon, + PlusIcon, + RotateCcwIcon, + ScissorsIcon, + SquareIcon, + Trash2Icon, + UnlinkIcon, +} from "lucide-react"; +import { useEffect, useId, useState } from "react"; + +import { cn } from "~/lib/utils"; +import { readLocalApi } from "~/localApi"; +import { Button } from "../ui/button"; +import { DialogDescription, DialogHeader, DialogTitle } from "../ui/dialog"; +import { Input } from "../ui/input"; +import { Label } from "../ui/label"; +import { ScrollArea } from "../ui/scroll-area"; +import { Textarea } from "../ui/textarea"; +import { Toggle, ToggleGroup } from "../ui/toggle-group"; +import { PullRequestGlyph } from "../pullRequest/pullRequestIcons"; +import { + layoutTaskGraph, + taskGraphCanvasEdgeId, + taskGraphCanvasElements, + taskGraphLayoutSignature, + type TaskGraphCanvasNode, + type TaskGraphCanvasPoint, +} from "./taskGraphCanvasLayout"; +import { + isTaskGraphFinished, + pullRequestLabel, + TASK_GRAPH_NODE_STATUS_DOT_CLASS, + TASK_GRAPH_NODE_STATUS_LABEL, + TASK_GRAPH_STATUS_LABEL, + taskGraphNodeKeyFromTitle, + taskGraphProgressLabel, +} from "./taskGraphView"; +import type { TaskGraphCommands } from "./useTaskGraphCommands"; + +type Selection = + | { readonly kind: "none" } + | { readonly kind: "node"; readonly key: string } + | { readonly kind: "edge"; readonly dependency: string; readonly dependent: string } + | { readonly kind: "new" }; + +const NO_SELECTION: Selection = { kind: "none" }; + +// One editor is open at a time; keeping its last layout lets status updates +// from a running graph skip dagre until nodes or dependencies change. +let cachedLayout: { + readonly signature: string; + readonly positions: ReadonlyMap; +} | null = null; + +function graphLayout(nodes: ReadonlyArray) { + const signature = taskGraphLayoutSignature(nodes); + if (cachedLayout?.signature !== signature) { + cachedLayout = { signature, positions: layoutTaskGraph(nodes) }; + } + return cachedLayout; +} + +const openExternal = (url: string) => void readLocalApi()?.shell.openExternal(url); + +function TaskCanvasNode({ data, selected }: NodeProps) { + const { node, opensPullRequest } = data; + return ( +
+ + + + {node.title} + + + {node.key} + {TASK_GRAPH_NODE_STATUS_LABEL[node.status]} + {opensPullRequest ? ( + + ) : null} + + +
+ ); +} + +const NODE_TYPES = { task: TaskCanvasNode }; + +/** Fits the view on mount; keyed by layout so it refits on structure changes, not status updates. */ +function FitViewOnMount() { + const { fitView } = useReactFlow(); + useEffect(() => { + void fitView({ maxZoom: 1 }); + }, [fitView]); + return null; +} + +/** + * Canvas editor for one task graph, loaded only when the card opens it. The + * subscription's graph is the only copy: every change is sent as an edit and + * the canvas redraws from what the server reports back. + */ +export default function TaskGraphEditor(props: { + readonly environmentId: EnvironmentId; + readonly graph: TaskGraph; + readonly commands: TaskGraphCommands; +}) { + const { graph, commands } = props; + const [selectionState, setSelection] = useState(NO_SELECTION); + const layout = graphLayout(graph.nodes); + const selection = resolveSelection(selectionState, graph); + const { nodes, edges } = taskGraphCanvasElements(graph.nodes, layout.positions, { + nodeKey: selection.kind === "node" ? selection.key : null, + edgeId: + selection.kind === "edge" + ? taskGraphCanvasEdgeId(selection.dependency, selection.dependent) + : null, + }); + const canEdit = commands.canEdit && !commands.busy; + const finished = isTaskGraphFinished(graph.status); + + const pendingNode = (key: string) => + graph.nodes.find((node) => node.key === key && node.status === "pending"); + + const connect = (connection: Connection) => { + const dependent = pendingNode(connection.target); + if (dependent === undefined || dependent.dependsOn.includes(connection.source)) return; + void commands.edit([ + { + type: "update_node", + key: dependent.key, + dependsOn: [...dependent.dependsOn, connection.source], + }, + ]); + }; + + return ( + <> + +
+
+ {graph.title} + + {TASK_GRAPH_STATUS_LABEL[graph.status]} · {taskGraphProgressLabel(graph)} · from{" "} + {graph.baseRef} + +
+
+ + {!finished ? ( + + ) : null} + {graph.status === "draft" ? ( + + ) : null} +
+
+ {commands.error !== null ? ( +

+ {commands.error} +

+ ) : null} + +
+
+ + connection.source !== connection.target && + pendingNode(connection.target) !== undefined + } + onConnect={connect} + onNodeClick={(_event, node) => setSelection({ kind: "node", key: node.id })} + onEdgeClick={(_event, edge) => + setSelection({ kind: "edge", dependency: edge.source, dependent: edge.target }) + } + onPaneClick={() => setSelection(NO_SELECTION)} + data-task-graph-canvas + > + + + +
+
+ +
+ +
+ +
+
+ + ); +} + +/** Drops a selection the server has since removed, such as a deleted node. */ +function resolveSelection(selection: Selection, graph: TaskGraph): Selection { + if (selection.kind === "node") { + return graph.nodes.some((node) => node.key === selection.key) ? selection : NO_SELECTION; + } + if (selection.kind === "edge") { + const dependent = graph.nodes.find((node) => node.key === selection.dependent); + return dependent?.dependsOn.includes(selection.dependency) ? selection : NO_SELECTION; + } + return selection; +} + +function SelectionPanel(props: { + readonly environmentId: EnvironmentId; + readonly graph: TaskGraph; + readonly commands: TaskGraphCommands; + readonly selection: Selection; + readonly onSelect: (selection: Selection) => void; +}) { + const { graph, commands, selection } = props; + const canEdit = commands.canEdit && !commands.busy; + + if (selection.kind === "new") { + return ( + + taskGraphNodeKeyFromTitle( + title, + graph.nodes.map((node) => node.key), + ) + } + onCancel={() => props.onSelect(NO_SELECTION)} + onSubmit={async (values) => { + const key = taskGraphNodeKeyFromTitle( + values.title, + graph.nodes.map((node) => node.key), + ); + const accepted = await commands.edit([ + { + type: "add_node", + node: { + key, + title: values.title, + prompt: values.prompt, + dependsOn: [], + ...(values.pullRequest === null ? {} : { pullRequest: values.pullRequest }), + }, + }, + ]); + if (accepted) props.onSelect({ kind: "node", key }); + }} + /> + ); + } + + if (selection.kind === "edge") { + const dependent = graph.nodes.find((node) => node.key === selection.dependent); + const dependency = graph.nodes.find((node) => node.key === selection.dependency); + if (dependent === undefined || dependency === undefined) return null; + return ( +
+

Dependency

+

+ {dependent.title} starts after{" "} + {dependency.title} succeeds. +

+ {dependent.status === "pending" ? ( + + ) : ( +

+ {dependent.title} has started, so its dependencies are fixed. +

+ )} +
+ ); + } + + if (selection.kind === "node") { + const node = graph.nodes.find((candidate) => candidate.key === selection.key); + if (node === undefined) return null; + return ( + + ); + } + + return ( +
+

Graph

+

+ Each task runs as its own thread in its own worktree. Drag from the right edge of a task to + the left edge of another to make the second wait for the first. +

+

Tasks that nothing depends on open a pull request unless you turn it off.

+

Select a task or dependency to change it.

+
+ ); +} + +function NodePanel(props: { + readonly environmentId: EnvironmentId; + readonly graph: TaskGraph; + readonly node: TaskGraphNode; + readonly commands: TaskGraphCommands; +}) { + const { graph, node, commands } = props; + const canEdit = commands.canEdit && !commands.busy; + const opensByDefault = taskGraphNodeOpensPullRequest(graph.nodes, { + ...node, + pullRequest: null, + }); + const canCancelBranch = + isActiveTaskGraphNodeStatus(node.status) || + (node.status === "pending" && graph.status !== "draft"); + const canRetry = isTerminalTaskGraphNodeStatus(node.status) && node.status !== "succeeded"; + + const actions = ( +
+ {canCancelBranch ? ( + + ) : null} + {canRetry ? ( + + ) : null} +
+ ); + + if (node.status === "pending") { + return ( +
+ { + await commands.edit([ + { + type: "update_node", + key: node.key, + ...(values.title === node.title ? {} : { title: values.title }), + ...(values.prompt === node.prompt ? {} : { prompt: values.prompt }), + ...(values.pullRequest === node.pullRequest + ? {} + : { pullRequest: values.pullRequest }), + }, + ]); + }} + /> +
+ + {actions} +
+
+ ); + } + + const pullRequest = node.pullRequestResult; + const pullRequestUrl = pullRequest?.url ?? null; + return ( +
+
+

{node.title}

+

+ + {TASK_GRAPH_NODE_STATUS_LABEL[node.status]} + {node.key} +

+
+ {node.error !== null ?

{node.error}

: null} + {node.branch !== null ? ( +

{node.branch}

+ ) : null} +
+ {node.threadId !== null ? ( + + ) : null} + {pullRequestUrl !== null ? ( + + ) : null} +
+ {pullRequest?.status === "failed" && pullRequest.error !== null ? ( +

Pull request failed: {pullRequest.error}

+ ) : null} + {actions} +
+

Prompt

+

{node.prompt}

+
+ {node.summary !== null ? ( +
+

Result

+

{node.summary}

+
+ ) : null} +
+ ); +} + +interface TaskNodeValues { + readonly title: string; + readonly prompt: string; + /** Null follows the default: a pull request only when nothing depends on the task. */ + readonly pullRequest: boolean | null; +} + +const PULL_REQUEST_CHOICES = ["auto", "on", "off"] as const; +type PullRequestChoice = (typeof PULL_REQUEST_CHOICES)[number]; + +const toChoice = (value: boolean | null): PullRequestChoice => + value === null ? "auto" : value ? "on" : "off"; +const fromChoice = (choice: PullRequestChoice): boolean | null => + choice === "auto" ? null : choice === "on"; + +/** Form text stays local until submitted; everything else comes from the server. */ +function TaskNodeForm(props: { + readonly heading: string; + readonly initial: TaskNodeValues; + readonly opensByDefault: boolean; + readonly disabled: boolean; + readonly submitLabel: string; + readonly keyFor?: (title: string) => string; + readonly onSubmit: (values: TaskNodeValues) => Promise; + readonly onCancel?: () => void; +}) { + const id = useId(); + const [title, setTitle] = useState(props.initial.title); + const [prompt, setPrompt] = useState(props.initial.prompt); + const [pullRequest, setPullRequest] = useState(props.initial.pullRequest); + const values = { title: title.trim(), prompt: prompt.trim(), pullRequest }; + const dirty = + values.title !== props.initial.title || + values.prompt !== props.initial.prompt || + values.pullRequest !== props.initial.pullRequest; + const valid = values.title.length > 0 && values.title.length <= 120 && values.prompt.length > 0; + + return ( +
{ + event.preventDefault(); + if (dirty && valid && !props.disabled) void props.onSubmit(values); + }} + > +

{props.heading}

+
+ + setTitle(event.target.value)} + /> + {props.keyFor && values.title.length > 0 ? ( +

+ Key {props.keyFor(values.title)} +

+ ) : null} +
+
+ +