From d1b6821376446a74ed4af4c04fcebf870e5b0f31 Mon Sep 17 00:00:00 2001 From: Jan Jaap Date: Wed, 28 Oct 2026 22:59:08 +0100 Subject: [PATCH 01/77] feat: pipelines chain agent steps across providers A pipeline is a saved sequence of agent steps (provider, model, options, permissions, skills, prompt template). Starting one creates a parent thread that shows progress and one child thread per step; each step's final answer feeds the next step's prompt. Steps auto-advance, can pause before a step, and can be continued, retried, skipped or cancelled. Includes a shipped "Feature delivery" starter, web and mobile editors and run views, and sidebar nesting of step threads under their pipeline. Co-Authored-By: Claude Opus 5.5 --- apps/mobile/src/Stack.tsx | 35 + apps/mobile/src/features/home/HomeScreen.tsx | 3 + .../src/features/keyboard/CommandPalette.tsx | 7 + .../settings/PipelinePickerScreens.tsx | 92 ++ .../settings/SettingsPipelinesRouteScreen.tsx | 826 +++++++++++++ .../features/settings/SettingsRouteScreen.tsx | 1 + .../SettingsScheduledTasksRouteScreen.tsx | 6 +- .../components/settings-page-targets.ts | 1 + .../src/features/settings/pipeline-editor.ts | 126 ++ .../src/features/settings/settings-search.ts | 1 + .../features/threads/NewTaskDraftScreen.tsx | 169 ++- .../src/features/threads/PipelineRunPanel.tsx | 473 ++++++++ .../src/features/threads/ThreadComposer.tsx | 22 +- .../features/threads/ThreadDetailScreen.tsx | 156 ++- .../threads/ThreadNavigationSidebar.tsx | 3 + .../threads/pipeline-run-view.test.ts | 44 + .../src/features/threads/pipeline-run-view.ts | 51 + .../features/threads/thread-list-v2-items.tsx | 131 ++- .../src/features/threads/threadListV2.test.ts | 89 ++ .../src/features/threads/threadListV2.ts | 205 ++-- .../features/threads/wait-for-thread-shell.ts | 33 + apps/server/src/auth/RpcAuthorization.ts | 10 + .../src/environment/ServerEnvironment.ts | 1 + .../src/observability/RpcInstrumentation.ts | 10 + .../Orchestrator.control-reads.test.ts | 71 ++ .../src/orchestration-v2/Orchestrator.ts | 18 + .../src/orchestration-v2/ProjectionStore.ts | 2 + .../orchestration-v2/ThreadLaunchService.ts | 4 + .../ThreadManagementService.test.ts | 7 + .../ThreadManagementService.ts | 9 +- .../orchestration-v2/ThreadMessageIntake.ts | 1 + .../src/orchestration-v2/runtimeLayer.ts | 14 + apps/server/src/persistence/Migrations.ts | 2 + .../persistence/Migrations/073_Pipelines.ts | 69 ++ .../pipelines/PipelineService.runner.test.ts | 522 +++++++++ .../src/pipelines/PipelineService.test.ts | 208 ++++ apps/server/src/pipelines/PipelineService.ts | 265 +++++ .../src/pipelines/PipelineStore.test.ts | 316 +++++ apps/server/src/pipelines/PipelineStore.ts | 516 ++++++++ apps/server/src/pipelines/pipelineRunner.ts | 1043 +++++++++++++++++ .../src/pipelines/pipelineTemplate.test.ts | 100 ++ apps/server/src/pipelines/pipelineTemplate.ts | 63 + apps/server/src/ws.test.ts | 2 + apps/server/src/ws.ts | 41 + apps/web/src/components/ChatView.tsx | 259 ++-- .../components/CommandPalette.logic.test.ts | 48 +- .../src/components/CommandPalette.logic.ts | 55 + apps/web/src/components/CommandPalette.tsx | 83 ++ .../components/CommandPaletteProjectItems.tsx | 21 + apps/web/src/components/Sidebar.logic.ts | 11 + apps/web/src/components/Sidebar.tsx | 345 ++++-- apps/web/src/components/chat/ChatComposer.tsx | 53 +- .../components/chat/ComposerControlsBar.tsx | 40 +- .../chat/ComposerPipelinePicker.tsx | 99 ++ .../src/components/chat/MessagesTimeline.tsx | 6 +- .../chat/PipelineRunView.logic.test.ts | 57 + .../components/chat/PipelineRunView.logic.ts | 55 + .../src/components/chat/PipelineRunView.tsx | 464 ++++++++ apps/web/src/components/chat/chatViewSend.ts | 28 +- .../chat/composerSend.logic.test.ts | 47 +- .../src/components/chat/composerSend.logic.ts | 65 +- .../restingComposerControlsMeasurement.ts | 9 +- .../src/components/chat/useComposerSend.ts | 65 +- .../src/components/chat/useDraftPipelines.ts | 36 + .../src/components/chat/usePipelineSend.ts | 143 +++ .../components/settings/PipelinesSettings.tsx | 925 +++++++++++++++ .../settings/SettingsSidebarNav.tsx | 2 + .../src/components/settings/settingsSearch.ts | 12 + .../components/sidebar/SidebarThreadRow.tsx | 72 +- apps/web/src/providerModels.ts | 26 +- apps/web/src/routeTree.gen.ts | 21 + .../src/routes/_chat.settings.pipelines.tsx | 7 + packages/client-runtime/package.json | 12 + .../client-runtime/src/operations/commands.ts | 62 +- .../client-runtime/src/pipelineEditor.test.ts | 345 ++++++ packages/client-runtime/src/pipelineEditor.ts | 333 ++++++ .../src/pipelineRunPresentation.test.ts | 288 +++++ .../src/pipelineRunPresentation.ts | 119 ++ packages/client-runtime/src/rpc/client.ts | 2 + .../src/state/commandPermissions.test.ts | 37 +- packages/client-runtime/src/state/models.ts | 3 + .../src/state/pipelineNesting.test.ts | 197 ++++ .../src/state/pipelineNesting.ts | 187 +++ packages/client-runtime/src/state/server.ts | 48 + packages/contracts/src/baseSchemas.ts | 9 + .../contracts/src/clientRpcPermissions.ts | 8 + packages/contracts/src/environment.ts | 2 + packages/contracts/src/index.ts | 1 + .../contracts/src/orchestrationV2.test.ts | 76 +- packages/contracts/src/orchestrationV2.ts | 13 + packages/contracts/src/pipeline.test.ts | 135 +++ packages/contracts/src/pipeline.ts | 296 +++++ packages/contracts/src/rpc.ts | 98 ++ packages/shared/package.json | 4 + packages/shared/src/model.ts | 21 + packages/shared/src/pipelineStarters.test.ts | 24 + packages/shared/src/pipelineStarters.ts | 178 +++ 97 files changed, 10805 insertions(+), 515 deletions(-) create mode 100644 apps/mobile/src/features/settings/PipelinePickerScreens.tsx create mode 100644 apps/mobile/src/features/settings/SettingsPipelinesRouteScreen.tsx create mode 100644 apps/mobile/src/features/settings/pipeline-editor.ts create mode 100644 apps/mobile/src/features/threads/PipelineRunPanel.tsx create mode 100644 apps/mobile/src/features/threads/pipeline-run-view.test.ts create mode 100644 apps/mobile/src/features/threads/pipeline-run-view.ts create mode 100644 apps/mobile/src/features/threads/wait-for-thread-shell.ts create mode 100644 apps/server/src/persistence/Migrations/073_Pipelines.ts create mode 100644 apps/server/src/pipelines/PipelineService.runner.test.ts create mode 100644 apps/server/src/pipelines/PipelineService.test.ts create mode 100644 apps/server/src/pipelines/PipelineService.ts create mode 100644 apps/server/src/pipelines/PipelineStore.test.ts create mode 100644 apps/server/src/pipelines/PipelineStore.ts create mode 100644 apps/server/src/pipelines/pipelineRunner.ts create mode 100644 apps/server/src/pipelines/pipelineTemplate.test.ts create mode 100644 apps/server/src/pipelines/pipelineTemplate.ts create mode 100644 apps/web/src/components/chat/ComposerPipelinePicker.tsx create mode 100644 apps/web/src/components/chat/PipelineRunView.logic.test.ts create mode 100644 apps/web/src/components/chat/PipelineRunView.logic.ts create mode 100644 apps/web/src/components/chat/PipelineRunView.tsx create mode 100644 apps/web/src/components/chat/useDraftPipelines.ts create mode 100644 apps/web/src/components/chat/usePipelineSend.ts create mode 100644 apps/web/src/components/settings/PipelinesSettings.tsx create mode 100644 apps/web/src/routes/_chat.settings.pipelines.tsx create mode 100644 packages/client-runtime/src/pipelineEditor.test.ts create mode 100644 packages/client-runtime/src/pipelineEditor.ts create mode 100644 packages/client-runtime/src/pipelineRunPresentation.test.ts create mode 100644 packages/client-runtime/src/pipelineRunPresentation.ts create mode 100644 packages/client-runtime/src/state/pipelineNesting.test.ts create mode 100644 packages/client-runtime/src/state/pipelineNesting.ts create mode 100644 packages/contracts/src/pipeline.test.ts create mode 100644 packages/contracts/src/pipeline.ts create mode 100644 packages/shared/src/pipelineStarters.test.ts create mode 100644 packages/shared/src/pipelineStarters.ts diff --git a/apps/mobile/src/Stack.tsx b/apps/mobile/src/Stack.tsx index 872ca142c8aaa..9008f40026583 100644 --- a/apps/mobile/src/Stack.tsx +++ b/apps/mobile/src/Stack.tsx @@ -111,6 +111,12 @@ import { SettingsScheduledTaskNewRouteScreen, SettingsScheduledTaskEditRouteScreen, } from "./features/settings/SettingsScheduledTasksRouteScreen"; +import { + SettingsPipelinesRouteScreen, + SettingsPipelineNewRouteScreen, + SettingsPipelineEditRouteScreen, +} from "./features/settings/SettingsPipelinesRouteScreen"; +import { PipelineStepModelPickerRouteScreen } from "./features/settings/PipelinePickerScreens"; import { ScheduledTaskModelPickerRouteScreen, ScheduledTaskBranchPickerRouteScreen, @@ -415,6 +421,35 @@ const SETTINGS_SCREENS = { }), }, }), + SettingsPipelines: createNativeStackScreen({ + screen: SettingsPipelinesRouteScreen, + linking: "settings/pipelines", + options: { + ...SETTINGS_PAGE_OPTIONS, + title: "Pipelines", + headerTitleStyle: { fontSize: 16, fontWeight: "800" }, + }, + }), + SettingsPipelineNew: createNativeStackScreen({ + screen: SettingsPipelineNewRouteScreen, + options: { ...SETTINGS_PAGE_OPTIONS, title: "New pipeline" }, + }), + SettingsPipelineEdit: createNativeStackScreen({ + screen: SettingsPipelineEditRouteScreen, + options: { ...SETTINGS_PAGE_OPTIONS, title: "Edit pipeline" }, + }), + SettingsPipelineStepModel: createNativeStackScreen({ + screen: PipelineStepModelPickerRouteScreen, + options: { + ...SETTINGS_PAGE_OPTIONS, + headerShown: false, + ...(Platform.OS === "android" + ? { presentation: "card" as const } + : { + ...THREAD_SETTINGS_SHEET_OPTIONS, + }), + }, + }), SettingsClientStorage: createNativeStackScreen({ screen: SettingsClientStorageRouteScreen, linking: "settings/client-storage", diff --git a/apps/mobile/src/features/home/HomeScreen.tsx b/apps/mobile/src/features/home/HomeScreen.tsx index 4609ecfb96c94..7b818aa25a346 100644 --- a/apps/mobile/src/features/home/HomeScreen.tsx +++ b/apps/mobile/src/features/home/HomeScreen.tsx @@ -575,6 +575,9 @@ export function HomeScreen(props: HomeScreenProps) { onNewThreadOnBranch={props.onNewThreadOnBranch} thread={thread} variant={item.item.variant} + nested={item.item.nested} + isLastStep={item.item.isLastStep} + stepStatus={item.item.stepStatus} hasQueuedMessages={item.hasQueuedMessages} snoozed={item.item.snoozed} pinned={item.item.pinned} diff --git a/apps/mobile/src/features/keyboard/CommandPalette.tsx b/apps/mobile/src/features/keyboard/CommandPalette.tsx index e9e9e50d4527f..a39a6bc5930b6 100644 --- a/apps/mobile/src/features/keyboard/CommandPalette.tsx +++ b/apps/mobile/src/features/keyboard/CommandPalette.tsx @@ -332,6 +332,13 @@ export function CommandPalette(props: { searchTerms: ["schedule", "automations", "recurring"], run: () => openSettings(navigation, "SettingsScheduledTasks"), }, + { + key: "pipelines", + kind: "action", + title: "Pipelines", + searchTerms: ["workflow", "steps", "chain"], + run: () => openSettings(navigation, "SettingsPipelines"), + }, { key: "usage", kind: "action", diff --git a/apps/mobile/src/features/settings/PipelinePickerScreens.tsx b/apps/mobile/src/features/settings/PipelinePickerScreens.tsx new file mode 100644 index 0000000000000..7ca996e6d0ae2 --- /dev/null +++ b/apps/mobile/src/features/settings/PipelinePickerScreens.tsx @@ -0,0 +1,92 @@ +import { useNavigation } from "@react-navigation/native"; +import { withoutPlanAgentOption } from "@t3tools/shared/model"; +import { useMemo } from "react"; + +import { AppText as Text } from "../../components/ui/AppText"; +import { buildModelOptions, groupByProvider } from "../../lib/modelOptions"; +import { resolveProviderOptionDescriptors } from "../../lib/providerOptions"; +import { useEnvironmentServerConfig } from "../../state/entities"; +import { ThreadSettingsPickerScreen } from "../threads/ThreadSettingsSheet"; +import { SettingsScreen } from "./components/SettingsScreen"; +import { usePipelineEditor } from "./pipeline-editor"; +import type { PipelineStepDraft } from "@t3tools/client-runtime/pipeline-editor"; + +function MissingPipelineDraft() { + return ( + + Open a pipeline form first. + + ); +} + +/** Model, effort and permissions for the step the form opened the picker for. */ +export function PipelineStepModelPickerRouteScreen() { + const navigation = useNavigation(); + const { editor, setEditor } = usePipelineEditor(); + const config = useEnvironmentServerConfig(editor?.environmentId ?? null); + const step = editor?.draft.steps.find((entry) => entry.key === editor.modelStepId) ?? null; + const selectedModel = step?.modelSelection ?? null; + const models = useMemo( + () => buildModelOptions(config, selectedModel, undefined, null), + [config, selectedModel], + ); + const providerGroups = useMemo(() => groupByProvider(models), [models]); + const selectedOption = models.find( + (option) => + option.selection.instanceId === selectedModel?.instanceId && + option.selection.model === selectedModel.model, + ); + const optionDescriptors = useMemo( + () => + resolveProviderOptionDescriptors({ + // Pipelines have no plan mode, so OpenCode's plan agent is not offered. + capabilities: selectedOption?.capabilities + ? withoutPlanAgentOption(selectedOption.capabilities) + : undefined, + selections: selectedModel?.options, + }), + [selectedOption?.capabilities, selectedModel?.options], + ); + + if (!editor || !step) return ; + + const updateStep = (update: (current: PipelineStepDraft) => PipelineStepDraft) => + setEditor((current) => + current + ? { + ...current, + draft: { + ...current.draft, + steps: current.draft.steps.map((entry) => + entry.key === step.key ? update(entry) : entry, + ), + }, + } + : current, + ); + + return ( + updateStep((entry) => ({ ...entry, promptTemplate }))} + onSelectModel={(option) => + updateStep((entry) => ({ ...entry, modelSelection: option.selection })) + } + optionDescriptors={optionDescriptors} + onUpdateOptionSelections={(options) => + updateStep((entry) => + entry.modelSelection + ? { ...entry, modelSelection: { ...entry.modelSelection, options } } + : entry, + ) + } + runtimeMode={step.runtimeMode} + onUpdateRuntimeMode={(runtimeMode) => updateStep((entry) => ({ ...entry, runtimeMode }))} + onClose={() => navigation.goBack()} + /> + ); +} diff --git a/apps/mobile/src/features/settings/SettingsPipelinesRouteScreen.tsx b/apps/mobile/src/features/settings/SettingsPipelinesRouteScreen.tsx new file mode 100644 index 0000000000000..2102cde2d7b03 --- /dev/null +++ b/apps/mobile/src/features/settings/SettingsPipelinesRouteScreen.tsx @@ -0,0 +1,826 @@ +import * as Dialog from "../../components/ui/Dialog"; +import { useAtomValue } from "@effect/atom-react"; +import { + AuthOrchestrationOperateScope, + PIPELINE_MAX_STEPS, + resolveEnvironmentMachineKind, + type EnvironmentId, + type PipelineDefinition, +} from "@t3tools/contracts"; +import { + isAtomCommandInterrupted, + squashAtomCommandFailure, + type AtomCommandResult, +} from "@t3tools/client-runtime/state/runtime"; +import { useNavigation, usePreventRemove } from "@react-navigation/native"; +import type { NativeStackNavigationProp } from "@react-navigation/native-stack"; +import { useRef, useState } from "react"; +import { Platform, Pressable, TextInput as RNTextInput, View } from "react-native"; +import { useSafeAreaInsets } from "react-native-safe-area-context"; + +import { AppText as Text } from "../../components/ui/AppText"; +import { SymbolView } from "../../components/ui/AppSymbol"; +import { Button } from "../../components/ui/Button"; +import { + EnvironmentMachineSymbol, + ENVIRONMENT_MACHINE_SYMBOLS, +} from "../../components/EnvironmentMachineSymbol"; +import { Menu } from "../../components/ui/Menu"; +import { ScreenScrollView as ScrollView } from "../../components/ui/ScreenScrollView"; +import { ThemedSwitch } from "../../components/ui/ThemedSwitch"; +import { buildModelOptions } from "../../lib/modelOptions"; +import { uuidv4 } from "../../lib/uuid"; +import { NativeStackScreenOptions } from "../../native/StackHeader"; +import { useEnvironmentServerConfig } from "../../state/entities"; +import { useEnvironmentQuery } from "../../state/query"; +import { readEnvironmentScope } from "../../state/session"; +import { serverEnvironment } from "../../state/server"; +import { useAtomCommand } from "../../state/use-atom-command"; +import { withNativeGlassHeaderItem } from "../layout/native-glass-header-items"; +import { RUNTIME_MODE_CHOICES } from "../threads/thread-settings-options"; +import { + AndroidSettingsEnvironmentFilter, + SettingsEnvironmentFilterHeader, +} from "./components/SettingsEnvironmentFilterHeader"; +import { SettingsScreen } from "./components/SettingsScreen"; +import { + describePipelineSteps, + draftToUpsertInput, + moveStep, + newStepDraft, + normalizeSkillName, + pipelineDraftIssues, + pipelineToDraft, + PIPELINE_STEP_SKILLS_MAX, + resolveStepIds, + type PipelineDraft, + type PipelineStepDraft, +} from "@t3tools/client-runtime/pipeline-editor"; +import { PIPELINE_STARTERS, type PipelineStarter } from "@t3tools/shared/pipelineStarters"; +import { SettingsSection } from "./components/SettingsSection"; +import { + discardPipelineEditor, + markPipelineEditorSaved, + pipelineTargetSupportsPipelines, + usePipelineEditor, +} from "./pipeline-editor"; +import { FormField, PickerRow, SelectRow } from "./SettingsScheduledTasksRouteScreen"; +import { useSettingsEnvironmentFilter, type SettingsTarget } from "./settings-environment-filter"; + +type PipelineRoutes = { + SettingsPipelineNew: undefined; + SettingsPipelineEdit: undefined; + SettingsPipelineStepModel: undefined; +}; + +const BLANK_PIPELINE_ACTION = "blank"; + +function reportFailure(title: string, result: AtomCommandResult) { + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + Dialog.alert(title, String(squashAtomCommandFailure(result))); + } +} + +export function SettingsPipelinesRouteScreen() { + const { availableTargets, selectedTargets } = useSettingsEnvironmentFilter(); + const { startEditor, resetEditor, startFromStarter, pipelineTargets } = usePipelineEditor(); + const navigation = useNavigation>(); + const insets = useSafeAreaInsets(); + const newPipeline = (starter: PipelineStarter | null) => { + if (pipelineTargets.length === 0) return; + if (starter) startFromStarter(starter); + else resetEditor(); + navigation.navigate("SettingsPipelineNew"); + }; + + return ( + <> + newPipeline(null), + }, + ...PIPELINE_STARTERS.map((starter) => ({ + type: "action" as const, + label: starter.name, + description: starter.description, + onPress: () => newPipeline(starter), + })), + ], + }, + }), + ]} + /> + + + ({ + id: starter.key, + title: starter.name, + subtitle: starter.description, + })), + ]} + onPressAction={({ nativeEvent }) => + newPipeline( + PIPELINE_STARTERS.find((starter) => starter.key === nativeEvent.event) ?? null, + ) + } + > + + + + + + } + > + + {selectedTargets.length > 0 ? ( + selectedTargets.map((environment) => ( + { + startEditor({ + environmentId: environment.environmentId, + environmentLabel: environment.label, + draft: pipelineToDraft(pipeline, uuidv4), + modelStepId: null, + }); + navigation.navigate("SettingsPipelineEdit"); + }} + /> + )) + ) : ( + + {availableTargets.length === 0 + ? "Connect an environment to view and create pipelines." + : "No environments match these filters. Change the filter above."} + + )} + + + + ); +} + +// The "New pipeline" action clears the editor before this screen opens, so it starts from the +// default target. +export function SettingsPipelineNewRouteScreen() { + return ; +} + +export function SettingsPipelineEditRouteScreen() { + return ; +} + +function SettingsPipelineEditorScreen({ title }: { readonly title: string }) { + const { editor, setEditor, hasChanges, pipelineTargets, modelForEnvironment } = + usePipelineEditor(); + const navigation = useNavigation(); + const insets = useSafeAreaInsets(); + const [saving, setSaving] = useState(false); + // Set once the save lands, so the removal guard lets the editor close. + const saved = useRef(false); + const preventRemove = hasChanges || saving; + usePreventRemove(preventRemove, ({ data }) => { + if (saved.current) { + navigation.dispatch(data.action); + return; + } + if (saving) { + Dialog.alert("Saving pipeline", "Wait for the pipeline to finish saving before leaving."); + return; + } + Dialog.alert("Discard changes?", "Your unsaved changes will be lost.", [ + { text: "Keep editing", style: "cancel" }, + { + text: "Discard changes", + style: "destructive", + onPress: () => { + discardPipelineEditor(); + navigation.dispatch(data.action); + }, + }, + ]); + }); + + return ( + + {Platform.OS === "ios" ? ( + [ + withNativeGlassHeaderItem({ + type: "button", + label: "", + accessibilityLabel: "Back", + icon: { type: "sfSymbol", name: "chevron.backward" }, + onPress: () => navigation.goBack(), + }), + ], + }} + /> + ) : null} + + {editor ? ( + target.environmentId === editor.environmentId) + ?.label ?? editor.environmentLabel + } + targets={pipelineTargets} + draft={editor.draft} + saving={saving} + setSaving={setSaving} + setDraft={(draft) => { + if (!saving) setEditor({ ...editor, draft }); + }} + openModelPicker={(stepId) => setEditor({ ...editor, modelStepId: stepId })} + onSaved={() => { + saved.current = true; + markPipelineEditorSaved(); + if (navigation.isFocused()) navigation.goBack(); + }} + onChangeEnvironment={(target) => { + if (target.environmentId === editor.environmentId) return; + // Models belong to the environment, so every step takes the new one's default. + const modelSelection = modelForEnvironment(target.environmentId); + setEditor({ + environmentId: target.environmentId, + environmentLabel: target.label, + modelStepId: null, + draft: { + ...editor.draft, + steps: editor.draft.steps.map((step) => ({ ...step, modelSelection })), + }, + }); + }} + /> + ) : ( + + Connect an environment that supports pipelines, or update its server, to create one. + + )} + + + ); +} + +function PipelineForm({ + environmentId, + environmentLabel, + targets, + draft, + saving, + setSaving, + setDraft, + openModelPicker, + onSaved, + onChangeEnvironment, +}: { + readonly environmentId: EnvironmentId; + readonly environmentLabel: string; + readonly targets: readonly SettingsTarget[]; + readonly draft: PipelineDraft; + readonly saving: boolean; + readonly setSaving: (saving: boolean) => void; + readonly setDraft: (draft: PipelineDraft) => void; + readonly openModelPicker: (stepId: string) => void; + readonly onSaved: () => void; + readonly onChangeEnvironment: (target: SettingsTarget) => void; +}) { + const navigation = useNavigation>(); + const pipelines = useEnvironmentQuery( + serverEnvironment.pipelinesLive({ environmentId, input: {} }), + ); + const config = useEnvironmentServerConfig(environmentId); + const canOperate = useAtomValue(serverEnvironment.upsertPipeline.permissionAtom(environmentId)); + const upsert = useAtomCommand(serverEnvironment.upsertPipeline, { + label: "pipeline upsert", + reportFailure: false, + }); + const submissionPending = useRef(false); + const pipelineMissing = + draft.editingId !== null && + pipelines.data !== null && + !pipelines.data.pipelines.some((pipeline) => pipeline.id === draft.editingId); + const environmentUnavailable = !targets.some((target) => target.environmentId === environmentId); + const issues = pipelineDraftIssues(draft).messages; + const stepIds = resolveStepIds(draft.steps); + + const updateStep = (stepId: string, patch: Partial) => + setDraft({ + ...draft, + steps: draft.steps.map((step) => (step.key === stepId ? { ...step, ...patch } : step)), + }); + + const save = async () => { + if ( + !readEnvironmentScope(environmentId, AuthOrchestrationOperateScope) || + submissionPending.current || + saving || + pipelineMissing || + environmentUnavailable + ) + return; + const input = draftToUpsertInput(draft); + if (!input) { + Dialog.alert("Incomplete pipeline", issues[0] ?? "Fix the highlighted problems first."); + return; + } + submissionPending.current = true; + setSaving(true); + const result = await upsert({ environmentId, input }); + setSaving(false); + if (result._tag === "Failure") { + submissionPending.current = false; + reportFailure("Could not save pipeline", result); + return; + } + onSaved(); + }; + + return ( + + + {pipelineMissing ? ( + + This pipeline no longer exists. + + ) : null} + {environmentUnavailable ? ( + + This environment is disconnected or does not support pipelines. Reconnect before saving. + + ) : null} + + + } + actions={ + draft.editingId || saving + ? [] + : targets.map((target) => ({ + id: target.environmentId, + title: target.label, + image: + ENVIRONMENT_MACHINE_SYMBOLS[ + resolveEnvironmentMachineKind(target.serverConfig) + ], + state: target.environmentId === environmentId ? "on" : undefined, + })) + } + onSelect={(id) => { + const target = targets.find((entry) => entry.environmentId === id); + if (target) onChangeEnvironment(target); + }} + /> + + + setDraft({ ...draft, name })} + /> + + + {draft.steps.map((step, index) => ( + updateStep(step.key, patch)} + onPickModel={() => { + openModelPicker(step.key); + navigation.navigate("SettingsPipelineStepModel"); + }} + onMove={(delta) => setDraft({ ...draft, steps: moveStep(draft.steps, index, delta) })} + onRemove={() => + setDraft({ ...draft, steps: draft.steps.filter((entry) => entry.key !== step.key) }) + } + /> + ))} + + + ) : null} + {actions.retry ? ( + + ) : null} + {actions.skip ? ( + + ) : null} + {actions.cancel ? ( + + ) : null} + + ) : null} + + +
+
+ +
+
+ + + +
    + {run.steps.map((step) => ( + + void perform(`pause:${step.index}`, "Could not update pause", () => + setPauseBefore({ + environmentId, + input: { runId: run.id, stepId: step.stepId, pauseBefore }, + }), + ) + } + /> + ))} +
+ + ); +} + +/** Between steps the run reads `running` with the next step still `pending`. */ +function FeedingNotice(props: { + readonly environmentId: EnvironmentId; + readonly run: PipelineRun; +}) { + const feeding = pipelineFeedingStep(props.run); + const shell = useQuietThreadShell( + feeding === null ? null : scopeThreadRef(props.environmentId, feeding.childThreadId), + ); + if (feeding === null) return null; + const active = pipelineChildLiveState(shell) === "working"; + return ( +

+ {active + ? `Waiting for ${feeding.name} to finish` + : `Starting ${stepName(props.run, props.run.currentStepIndex)}`} +

+ ); +} + +function PipelineStepRow(props: { + readonly environmentId: EnvironmentId; + readonly run: PipelineRun; + readonly step: PipelineRunStep; + readonly providerEntries: ReturnType; + readonly cwd: string | undefined; + readonly canTogglePauseBefore: boolean; + readonly onPauseBeforeChange: (pauseBefore: boolean) => void; +}) { + const { environmentId, run, step } = props; + const [outputOpen, setOutputOpen] = useState(false); + const openThread = useOpenThread(environmentId); + const definitionStep = run.definition.steps[step.index]; + const shell = useQuietThreadShell( + step.childThreadId === null ? null : scopeThreadRef(environmentId, step.childThreadId), + ); + const liveState = pipelineChildLiveState(shell); + const { Icon, className: iconClassName } = STEP_STATUS_ICON[step.status]; + const model = + definitionStep === undefined + ? null + : pipelineStepModelLabel(definitionStep.modelSelection, props.providerEntries); + const isCurrent = step.index === run.currentStepIndex && run.status !== "completed"; + const statusLabel = pipelineStepStatusLabel(step); + const childThreadId = step.childThreadId; + + return ( +
  • +
    + + + {stepName(run, step.index)} + + {step.attempt > 1 ? ( + + Attempt {step.attempt} + + ) : null} + {liveState === null ? null : ( + + {PIPELINE_CHILD_LIVE_STATE_LABEL[liveState]} + + )} + {statusLabel} + {step.startedAt ? ( + + + + ) : null} +
    + + {model === null ? null : ( +
    + {model.entry ? ( + + ) : null} + {model.model} + {model.effort === null ? null : {model.effort}} +
    + )} + + {step.error ? ( +

    + {step.error} +

    + ) : null} + + {step.outputPreview ? ( +
    + setOutputOpen((open) => !open)} + > + Output + + {outputOpen ? ( +
    + +
    + ) : null} +
    + ) : null} + + {childThreadId !== null || step.status === "pending" ? ( +
    + {childThreadId === null ? ( + + ) : ( + + )} + {step.status === "pending" ? ( + + ) : null} +
    + ) : null} +
    • + ); +} diff --git a/apps/web/src/components/chat/chatViewSend.ts b/apps/web/src/components/chat/chatViewSend.ts index 81a00823a98be..2948e50181a39 100644 --- a/apps/web/src/components/chat/chatViewSend.ts +++ b/apps/web/src/components/chat/chatViewSend.ts @@ -2,6 +2,7 @@ import { CommandId, type EnvironmentId, type ModelSelection, + type PipelineId, type ProviderInteractionMode, type ScopedThreadRef, type ThreadId, @@ -44,6 +45,9 @@ import { waitForRevertedMessage, } from "../ChatView.logic"; import { readPreparedConnection } from "../../state/session"; +import { appAtomRegistry } from "../../rpc/atomRegistry"; +import { scopedThreadKey } from "@t3tools/client-runtime/environment"; +import type { DraftPipelineSelection } from "./composerSend.logic"; import { recallableComposerPrompt } from "./composerPromptHistory"; // Kept per route so a send in flight, and the multiple-model selections, survive @@ -51,17 +55,39 @@ import { recallableComposerPrompt } from "./composerPromptHistory"; export const draftFanoutSelectionsAtom = Atom.family((_routeKey: string) => Atom.make | null>(null).pipe(Atom.keepAlive), ); +// The pipeline chosen on a local draft; exclusive with multiple-model selections. +export const draftPipelineSelectionAtom = Atom.family((_routeKey: string) => + Atom.make(null).pipe(Atom.keepAlive), +); + +/** + * Opens `pipelineId` as the pipeline of the draft shown at `threadRef`, e.g. from the command + * palette. Multiple-model selections would otherwise stay on the draft beside it. + */ +export function presetDraftPipeline(threadRef: ScopedThreadRef, pipelineId: PipelineId) { + const routeKey = scopedThreadKey(threadRef); + appAtomRegistry.set(draftFanoutSelectionsAtom(routeKey), null); + appAtomRegistry.set(draftPipelineSelectionAtom(routeKey), { + environmentId: threadRef.environmentId, + pipelineId, + }); +} const draftSendStateByRouteKey = new Map< string, { readonly sendInFlight: { current: boolean }; readonly uncertainSubmissions: { current: Map }; + readonly pipelineCommands: { current: Map }; } >(); export function draftSendState(routeKey: string) { let state = draftSendStateByRouteKey.get(routeKey); if (!state) { - state = { sendInFlight: { current: false }, uncertainSubmissions: { current: new Map() } }; + state = { + sendInFlight: { current: false }, + uncertainSubmissions: { current: new Map() }, + pipelineCommands: { current: new Map() }, + }; draftSendStateByRouteKey.set(routeKey, state); } return state; diff --git a/apps/web/src/components/chat/composerSend.logic.test.ts b/apps/web/src/components/chat/composerSend.logic.test.ts index 13e36ea5659c9..676b9682295c4 100644 --- a/apps/web/src/components/chat/composerSend.logic.test.ts +++ b/apps/web/src/components/chat/composerSend.logic.test.ts @@ -1,11 +1,16 @@ -import { EnvironmentId, ThreadId } from "@t3tools/contracts"; +import { EnvironmentId, PipelineId, ThreadId } from "@t3tools/contracts"; import { scopeThreadRef } from "@t3tools/client-runtime/environment"; import { describe, expect, it } from "vite-plus/test"; import { formatInlineContextReference } from "../../lib/composerContextReferences"; import { threadContextReference } from "../../lib/composerContextRecords"; import { threadContextRecord } from "@t3tools/shared/composerContextReferences"; import { + activeDraftPipelineId, approvalRedirectNotice, + commandIdForPipelineSend, + PIPELINE_TEXT_ONLY_HINT, + pipelineSendBlock, + pipelineSendRetryKey, loadBalancingSendWarning, multipleModelSendBlock, referencedThreadContexts, @@ -124,3 +129,43 @@ describe("approvalRedirectNotice", () => { expect(approvalRedirectNotice(settled)).toBeNull(); }); }); + +describe("activeDraftPipelineId", () => { + const pipelineId = PipelineId.make("pipeline-1"); + it("applies only in the environment it was chosen in", () => { + expect(activeDraftPipelineId({ environmentId: local, pipelineId }, local)).toBe(pipelineId); + expect(activeDraftPipelineId({ environmentId: local, pipelineId }, remote)).toBeNull(); + expect(activeDraftPipelineId(null, local)).toBeNull(); + }); +}); + +describe("pipelineSendBlock", () => { + it("allows text only", () => { + expect(pipelineSendBlock({ attachmentCount: 0, hasContext: false })).toBeNull(); + expect(pipelineSendBlock({ attachmentCount: 1, hasContext: false })).toBe( + PIPELINE_TEXT_ONLY_HINT, + ); + expect(pipelineSendBlock({ attachmentCount: 0, hasContext: true })).toBe( + PIPELINE_TEXT_ONLY_HINT, + ); + }); +}); + +describe("commandIdForPipelineSend", () => { + const key = (text: string) => + pipelineSendRetryKey({ + routeKey: "r", + pipelineId: "p", + projectId: "proj", + workspaceStrategy: { type: "root" }, + text, + }); + it("reuses the id for the same attempt and mints a new one when the text changes", () => { + const commands = new Map(); + let next = 0; + const make = () => `cmd-${(next += 1)}`; + expect(commandIdForPipelineSend(commands, key("a"), make)).toBe("cmd-1"); + expect(commandIdForPipelineSend(commands, key("a"), make)).toBe("cmd-1"); + expect(commandIdForPipelineSend(commands, key("b"), make)).toBe("cmd-2"); + }); +}); diff --git a/apps/web/src/components/chat/composerSend.logic.ts b/apps/web/src/components/chat/composerSend.logic.ts index e19512e6c90d4..802299dd8cf3a 100644 --- a/apps/web/src/components/chat/composerSend.logic.ts +++ b/apps/web/src/components/chat/composerSend.logic.ts @@ -1,4 +1,4 @@ -import type { EnvironmentId, ThreadContextRecord } from "@t3tools/contracts"; +import type { EnvironmentId, PipelineId, ThreadContextRecord } from "@t3tools/contracts"; import { collectInlineContextIds } from "../../lib/composerContextReferences"; /** Attached threads the prompt still links to; replacing linked text leaves stale records. */ @@ -79,3 +79,66 @@ export function approvalRedirectNotice(input: { } return null; } + +/** The pipeline chosen on a draft, remembered with the environment it was chosen from. */ +export interface DraftPipelineSelection { + readonly environmentId: EnvironmentId; + readonly pipelineId: PipelineId; +} + +/** + * The selected pipeline for a draft in `environmentId`, or null. Pipelines live on one server, so + * a selection made before the draft moved to another environment no longer applies. + */ +export function activeDraftPipelineId( + selection: DraftPipelineSelection | null, + environmentId: EnvironmentId, +): PipelineId | null { + return selection !== null && selection.environmentId === environmentId + ? selection.pipelineId + : null; +} + +export const PIPELINE_TEXT_ONLY_HINT = + "Pipelines take text only. Remove attachments and added context, or choose No pipeline."; + +/** Why a pipeline cannot start from this draft: it carries more than text. */ +export function pipelineSendBlock(input: { + readonly attachmentCount: number; + readonly hasContext: boolean; +}): string | null { + return input.attachmentCount > 0 || input.hasContext ? PIPELINE_TEXT_ONLY_HINT : null; +} + +/** Same pipeline, project, workspace and text means the same attempt. */ +export function pipelineSendRetryKey(input: { + readonly routeKey: string; + readonly pipelineId: string; + readonly projectId: string; + readonly workspaceStrategy: unknown; + readonly text: string; +}): string { + return JSON.stringify([ + input.routeKey, + input.pipelineId, + input.projectId, + input.workspaceStrategy, + input.text, + ]); +} + +/** + * A command id that stays the same while the same send is retried, so a request that reached the + * server before failing resolves to its run instead of starting a second one. + */ +export function commandIdForPipelineSend( + commands: Map, + retryKey: string, + make: () => T, +): T { + const existing = commands.get(retryKey); + if (existing !== undefined) return existing; + const created = make(); + commands.set(retryKey, created); + return created; +} diff --git a/apps/web/src/components/chat/restingComposerControlsMeasurement.ts b/apps/web/src/components/chat/restingComposerControlsMeasurement.ts index 474df488a6c6b..85c9bf8fe5ab9 100644 --- a/apps/web/src/components/chat/restingComposerControlsMeasurement.ts +++ b/apps/web/src/components/chat/restingComposerControlsMeasurement.ts @@ -86,7 +86,14 @@ export function measureRestingComposerControls( const separator = controls.querySelector("[data-resting-controls-separator]"); const separatorWidth = separator ? elementOuterWidth(separator) : 0; const overflow = controls.querySelector("[data-resting-controls-overflow]"); - const separatorAndGapWidth = separatorWidth > 0 ? separatorWidth + gap : 0; + // Other always-visible leading chips (the pipeline picker) sit beside the model picker. + const fixedChipsWidth = Array.from( + controls.querySelectorAll("[data-resting-controls-fixed]"), + ).reduce((total, chip) => { + const width = elementOuterWidth(chip); + return width > 0 ? total + width + gap : total; + }, 0); + const separatorAndGapWidth = (separatorWidth > 0 ? separatorWidth + gap : 0) + fixedChipsWidth; const blocks = Array.from(controls.querySelectorAll("[data-resting-block]")); const widths = blocks.map(controlBlockWidths); return { diff --git a/apps/web/src/components/chat/useComposerSend.ts b/apps/web/src/components/chat/useComposerSend.ts index 98084e9795443..5c7628c42cb0d 100644 --- a/apps/web/src/components/chat/useComposerSend.ts +++ b/apps/web/src/components/chat/useComposerSend.ts @@ -25,7 +25,7 @@ import { ProviderInteractionMode, RuntimeMode, } from "@t3tools/contracts"; -import { deriveThreadTitleSeed } from "@t3tools/client-runtime/operations"; +import { deriveThreadTitleSeed, draftWorkspaceStrategy } from "@t3tools/client-runtime/operations"; import { wasBootstrapThreadDeleted } from "@t3tools/client-runtime/errors"; import { declinePendingApprovals, @@ -94,9 +94,12 @@ import { environmentPresentations } from "../../state/presentation"; import { approvalRedirectNotice, loadBalancingSendWarning, + activeDraftPipelineId, multipleModelSendBlock, + pipelineSendBlock, referencedThreadContexts, } from "./composerSend.logic"; +import { usePipelineSend } from "./usePipelineSend"; import { usePlanFollowUpSend } from "./useProposedPlanSend"; import { useMultipleModelSend } from "./useMultipleModelSend"; import { type ComposerDispatchMode } from "@t3tools/client-runtime/state/composer-dispatch"; @@ -118,6 +121,7 @@ import { threadContextsOutsideEnvironment, } from "@t3tools/shared/composerContextReferences"; import { + draftPipelineSelectionAtom, draftSendState, type MultipleModelTarget, resolveMultipleModelTargets, @@ -372,6 +376,16 @@ export function useComposerSend(input: ComposerSendInput) { resetLocalDispatch, setThreadError, }); + const sendToPipeline = usePipelineSend({ + environmentId, + routeThreadKey, + currentRouteThreadKeyRef, + composerDraftTarget, + composerRef, + promptRef, + resetLocalDispatch, + setThreadError, + }); const submitPlanFollowUp = usePlanFollowUpSend({ environmentId, routeThreadKey, @@ -546,6 +560,14 @@ export function useComposerSend(input: ComposerSendInput) { return; } const multipleModelSelections = sendCtx.multipleModelSelections; + // A pipeline starts from a new thread and excludes multiple models. + const selectedPipelineId = + isLocalDraftThread && multipleModelSelections === null + ? activeDraftPipelineId( + appAtomRegistry.get(draftPipelineSelectionAtom(routeThreadKey)), + environmentId, + ) + : null; const approvalFeedbackTarget = pendingApprovalsRef.current.find( (approval) => approval.supportsFeedback && approval.requestId === sendCtx.shownPendingApprovalIds[0], @@ -660,6 +682,7 @@ export function useComposerSend(input: ComposerSendInput) { composerEnvironmentContexts.length, }); const feedbackCommand = + selectedPipelineId === null && ctxSelectedProvider === "codex" && composerImages.length === 0 && composerFiles.length === 0 && @@ -798,6 +821,7 @@ export function useComposerSend(input: ComposerSendInput) { return; } const standaloneSlashCommand = + selectedPipelineId === null && sendInteractionModeEnabled && composerImages.length === 0 && composerFiles.length === 0 && @@ -886,6 +910,45 @@ export function useComposerSend(input: ComposerSendInput) { setThreadError(threadIdForSend, "Select a base branch before sending in New worktree mode."); return; } + if (selectedPipelineId !== null) { + const pipelineBlock = pipelineSendBlock({ + attachmentCount: composerAttachmentsSnapshot.length, + hasContext: + sendableComposerTerminalContexts.length > 0 || + composerPreviewAnnotationsSnapshot.length > 0 || + composerReviewCommentsSnapshot.length > 0 || + composerThreadContextsSnapshot.length > 0 || + composerEnvironmentContextsSnapshot.length > 0, + }); + if (pipelineBlock !== null) { + setThreadError(threadIdForSend, pipelineBlock); + return; + } + const workspaceStrategy = draftWorkspaceStrategy( + { branch: activeThreadBranch, worktreePath: activeThread.worktreePath }, + baseBranchForWorktree + ? { + projectCwd: activeProject.workspaceRoot, + baseBranch: baseBranchForWorktree, + ...(startFromOrigin ? { startFromOrigin: true } : {}), + } + : undefined, + ); + sendInFlightRef.current = true; + beginLocalDispatch({ + preparingWorktree: workspaceStrategy.type === "worktree", + submissionIntent: "foreground", + }); + await sendToPipeline({ + threadId: threadIdForSend, + pipelineId: selectedPipelineId, + projectId: activeProject.id, + text: messageTextForSend, + promptAtSend: promptForSend, + workspaceStrategy, + }); + return; + } // Records bind attachments by the id each side knows: the local id for the optimistic // row, the upload id (or local id on the data-URL path) on the wire; the server // rebinds them to the persisted id. diff --git a/apps/web/src/components/chat/useDraftPipelines.ts b/apps/web/src/components/chat/useDraftPipelines.ts new file mode 100644 index 0000000000000..255214ad0b322 --- /dev/null +++ b/apps/web/src/components/chat/useDraftPipelines.ts @@ -0,0 +1,36 @@ +import { useAtomValue } from "@effect/atom-react"; +import type { EnvironmentId, PipelineDefinition } from "@t3tools/contracts"; +import { Atom } from "effect/reactivity"; + +import { useEnvironmentQuery } from "../../state/query"; +import { environmentServerConfigsAtom, serverEnvironment } from "../../state/server"; + +const NO_PIPELINES: ReadonlyArray = []; + +// A boolean per environment, so config pushes that leave the capability alone do not re-render +// the composer. +const supportsPipelinesAtom = Atom.family((environmentId: EnvironmentId) => + Atom.make( + (get) => + get(environmentServerConfigsAtom).get(environmentId)?.environment.capabilities.pipelines === + true, + ).pipe(Atom.withLabel(`web-composer-pipelines-supported:${environmentId}`)), +); + +/** + * The pipelines a draft in `environmentId` can start. Subscribes only while `enabled` and the + * server answers pipelines.* and the connection may start runs. + */ +export function useDraftPipelines(environmentId: EnvironmentId, enabled: boolean) { + const supported = useAtomValue(supportsPipelinesAtom(environmentId)); + const canStart = useAtomValue(serverEnvironment.startPipelineRun.permissionAtom(environmentId)); + const available = enabled && supported && canStart; + const query = useEnvironmentQuery( + available ? serverEnvironment.pipelinesLive({ environmentId, input: {} }) : null, + ); + return { + available, + loaded: query.data !== null, + pipelines: query.data?.pipelines ?? NO_PIPELINES, + }; +} diff --git a/apps/web/src/components/chat/usePipelineSend.ts b/apps/web/src/components/chat/usePipelineSend.ts new file mode 100644 index 0000000000000..95fbb343fbf57 --- /dev/null +++ b/apps/web/src/components/chat/usePipelineSend.ts @@ -0,0 +1,143 @@ +import { + CommandId, + type EnvironmentId, + type PipelineId, + type ProjectId, + type ThreadId, +} from "@t3tools/contracts"; +import { scopeThreadRef } from "@t3tools/client-runtime/environment"; +import type { OrchestrationV2ThreadLaunchWorkspaceStrategy } from "@t3tools/contracts"; +import { squashAtomCommandFailure } from "@t3tools/client-runtime/state/runtime"; +import { useNavigate } from "@tanstack/react-router"; +import { randomUUID } from "~/lib/utils"; +import { type ComposerThreadTarget, useComposerDraftStore } from "../../composerDraftStore"; +import type { ComposerHandleRef } from "../../composerHandleContext"; +import { appAtomRegistry } from "../../rpc/atomRegistry"; +import { serverEnvironment } from "../../state/server"; +import { useOrchestrationCommand } from "../../state/use-orchestration-command"; +import { buildThreadRouteParams } from "../../threadRoutes"; +import { stackedThreadToast, toastManager } from "../ui/toast"; +import type { makeThreadErrorSetter } from "../ChatView.logic"; +import { draftPipelineSelectionAtom, draftSendState } from "./chatViewSend"; +import { commandIdForPipelineSend, pipelineSendRetryKey } from "./composerSend.logic"; +import type { useLocalDispatchState } from "./useLocalDispatchState"; + +interface PipelineSendInput { + readonly environmentId: EnvironmentId; + readonly routeThreadKey: string; + readonly currentRouteThreadKeyRef: { readonly current: string | null }; + readonly composerDraftTarget: ComposerThreadTarget; + readonly composerRef: ComposerHandleRef; + readonly promptRef: { current: string }; + readonly resetLocalDispatch: ReturnType["resetLocalDispatch"]; + readonly setThreadError: ReturnType; +} + +/** + * Starts a pipeline run from a new-thread draft. Success clears the draft and opens the run's + * parent thread; failure keeps the draft and toasts. A retry reuses its command id. + */ +export function usePipelineSend(input: PipelineSendInput) { + const { + environmentId, + routeThreadKey, + currentRouteThreadKeyRef, + composerDraftTarget, + composerRef, + promptRef, + resetLocalDispatch, + setThreadError, + } = input; + const navigate = useNavigate(); + const startPipelineRun = useOrchestrationCommand(serverEnvironment.startPipelineRun, { + reportFailure: false, + }); + const { sendInFlight: sendInFlightRef, pipelineCommands: pipelineCommandsRef } = + draftSendState(routeThreadKey); + + return async ({ + threadId, + pipelineId, + projectId, + text, + promptAtSend, + workspaceStrategy, + }: { + readonly threadId: ThreadId; + readonly pipelineId: PipelineId; + readonly projectId: ProjectId; + readonly text: string; + /** The draft's prompt when sent; a draft edited since is left alone. */ + readonly promptAtSend: string; + readonly workspaceStrategy: OrchestrationV2ThreadLaunchWorkspaceStrategy; + }) => { + const retryKey = pipelineSendRetryKey({ + routeKey: routeThreadKey, + pipelineId, + projectId, + workspaceStrategy, + text, + }); + const commandId = commandIdForPipelineSend(pipelineCommandsRef.current, retryKey, () => + CommandId.make(randomUUID()), + ); + setThreadError(threadId, null); + const result = await startPipelineRun({ + environmentId, + input: { + commandId, + pipelineId, + projectId, + input: text, + workspaceStrategy, + creationSource: "web", + }, + }); + sendInFlightRef.current = false; + resetLocalDispatch(); + if (result._tag === "Failure") { + const error = squashAtomCommandFailure(result); + const message = error instanceof Error ? error.message : "Failed to start the pipeline."; + setThreadError(threadId, message); + toastManager.add( + stackedThreadToast({ + type: "error", + title: "Could not start pipeline", + description: message, + }), + ); + return; + } + pipelineCommandsRef.current.delete(retryKey); + const parentRef = scopeThreadRef(environmentId, result.value.run.parentThreadId); + const stillOnDraft = currentRouteThreadKeyRef.current === routeThreadKey; + const draft = useComposerDraftStore.getState().getComposerDraft(composerDraftTarget); + if (draft?.prompt === promptAtSend) { + useComposerDraftStore.getState().clearComposerContent(composerDraftTarget); + if (stillOnDraft) { + promptRef.current = ""; + composerRef.current?.resetCursorState(); + } + } + appAtomRegistry.set(draftPipelineSelectionAtom(routeThreadKey), null); + if (stillOnDraft) { + void navigate({ to: "/$environmentId/$threadId", params: buildThreadRouteParams(parentRef) }); + } else { + toastManager.add( + stackedThreadToast({ + type: "success", + title: "Pipeline started", + actionProps: { + children: "Open", + onClick: () => { + void navigate({ + to: "/$environmentId/$threadId", + params: buildThreadRouteParams(parentRef), + }); + }, + }, + }), + ); + } + }; +} diff --git a/apps/web/src/components/settings/PipelinesSettings.tsx b/apps/web/src/components/settings/PipelinesSettings.tsx new file mode 100644 index 0000000000000..c543ca7450ba9 --- /dev/null +++ b/apps/web/src/components/settings/PipelinesSettings.tsx @@ -0,0 +1,925 @@ +import { useAtomValue } from "@effect/atom-react"; +import { + isAtomCommandInterrupted, + squashAtomCommandFailure, +} from "@t3tools/client-runtime/state/runtime"; +import { connectionStatusTitle } from "@t3tools/client-runtime/connection"; +import { dedupeProviderSkillsByName } from "@t3tools/client-runtime/providerSkills"; +import { + draftToUpsertInput, + emptyPipelineDraft, + moveStep, + newStepDraft, + normalizeSkillName, + PIPELINE_NAME_MAX_LENGTH, + PIPELINE_STEP_SKILLS_MAX, + type PipelineDraft, + pipelineDraftIssues, + type PipelineStepDefaults, + type PipelineStepDraft, + pipelineStepSummary, + pipelineToDraft, + starterToDraft, +} from "@t3tools/client-runtime/pipeline-editor"; +import { + AuthOrchestrationOperateScope, + type EnvironmentId, + type ModelSelection, + PIPELINE_MAX_STEPS, + type PipelineDefinition, + type ProviderInstanceId, + type RuntimeMode, + type ServerProvider, + type ServerProviderSkill, + resolveEnvironmentMachineKind, +} from "@t3tools/contracts"; +import type { UnifiedSettings } from "@t3tools/contracts/settings"; +import { createModelSelection } from "@t3tools/shared/model"; +import { PIPELINE_STARTERS, type PipelineStarter } from "@t3tools/shared/pipelineStarters"; +import { AsyncResult } from "effect/reactivity"; +import { + ArrowDownIcon, + ArrowUpIcon, + ChevronDownIcon, + PencilIcon, + PlusIcon, + Trash2Icon, + WorkflowIcon, + XIcon, +} from "lucide-react"; +import { type ReactNode, useRef, useState } from "react"; + +import { useEnvironmentSettings } from "../../hooks/useSettings"; +import { requestConfirmDialog } from "../../confirmDialog"; +import { randomUUID } from "../../lib/utils"; +import { getCustomModelOptionsByInstance } from "../../modelSelection"; +import { + deriveProviderInstanceEntries, + type ProviderInstanceEntry, + providerInstanceEntriesFor, +} from "../../providerInstances"; +import { environmentAtomList, environmentIdsKey } from "../../state/environmentAtomList"; +import { type EnvironmentPresentation, useEnvironment } from "../../state/environments"; +import { formatEnvironmentQueryError } from "../../state/query"; +import { EMPTY_SERVER_PROVIDERS, serverEnvironment } from "../../state/server"; +import { useAtomCommand } from "../../state/use-atom-command"; +import { readEnvironmentScope } from "~/state/session"; +import { EnvironmentMachineIcon } from "../EnvironmentMachineIcon"; +import { ProviderModelPicker } from "../chat/ProviderModelPicker"; +import { runtimeModeConfig, runtimeModeOptions } from "../chat/runtimeModeConfig"; +import { TraitsPicker } from "../chat/TraitsPicker"; +import { Badge } from "../ui/badge"; +import { Button } from "../ui/button"; +import { + Dialog, + DialogClose, + DialogDescription, + DialogFooter, + DialogHeader, + DialogPanel, + DialogPopup, + DialogTitle, +} from "../ui/dialog"; +import { Empty, EmptyDescription, EmptyHeader, EmptyMedia, EmptyTitle } from "../ui/empty"; +import { Input } from "../ui/input"; +import { Label } from "../ui/label"; +import { Menu, MenuItem, MenuPopup, MenuTrigger } from "../ui/menu"; +import { Select, SelectItem, SelectPopup, SelectTrigger, SelectValue } from "../ui/select"; +import { Switch } from "../ui/switch"; +import { Textarea } from "../ui/textarea"; +import { stackedThreadToast, toastManager } from "../ui/toast"; +import { scheduledTaskDefaultModel } from "./scheduledTasksSettings.logic"; + +import { + SettingsPageContainer, + SettingsRow, + SettingsSection, + SETTINGS_PICKER_TRIGGER_CLASSNAME, +} from "./settingsLayout"; +import { useSettingsScope } from "./SettingsScopeContext"; + +const TEMPLATE_VARIABLES_HINT = + "Variables: {{input}} is the text the run starts with, {{previous.output}} the last message of the step before, {{steps..output}} an earlier step by id or name."; + +function supportsPipelines(environment: EnvironmentPresentation): boolean { + return ( + environment.connection.phase === "connected" && + environment.serverConfig?.environment.capabilities.pipelines === true + ); +} + +function reportFailure(title: string, error: unknown) { + toastManager.add( + stackedThreadToast({ + type: "error", + title, + description: error instanceof Error ? error.message : String(error), + }), + ); +} + +function Field({ + label, + hint, + htmlFor, + children, +}: { + label: string; + hint?: string; + htmlFor?: string; + children: ReactNode; +}) { + return ( +
    + + {children} +
    + ); +} + +export function PipelinesSettings() { + const { scope, environments, connectedEnvironments, environment } = useSettingsScope(); + const [editor, setEditor] = useState<{ + environmentId: EnvironmentId; + pipeline: PipelineDefinition | null; + starter: PipelineStarter | null; + } | null>(null); + const inScope = (entry: EnvironmentPresentation) => + scope.kind !== "unavailable" && scope.environmentIds.includes(entry.environmentId); + const scopedEnvironments = environments.filter(inScope); + const capableEnvironments = connectedEnvironments.filter( + (entry) => inScope(entry) && supportsPipelines(entry), + ); + const defaultEnvironment = + environment && inScope(environment) && supportsPipelines(environment) + ? environment + : capableEnvironments[0]; + const canCreate = useAtomValue( + serverEnvironment.upsertPipeline.permissionAtom(defaultEnvironment?.environmentId ?? null), + ); + + return ( + + + + } + > + + New pipeline + + + + + defaultEnvironment && + setEditor({ + environmentId: defaultEnvironment.environmentId, + pipeline: null, + starter: null, + }) + } + > + Blank pipeline + + {PIPELINE_STARTERS.map((starter) => ( + + defaultEnvironment && + setEditor({ + environmentId: defaultEnvironment.environmentId, + pipeline: null, + starter, + }) + } + > +
    +
    {starter.name}
    +
    {starter.description}
    +
    +     + ))} +     + + } + > + {scope.kind === "unavailable" ? ( + + + + ) : scopedEnvironments.length === 0 ? ( + + + + + + No environments available + Connect an environment to manage pipelines. + + + ) : ( + + setEditor({ environmentId, pipeline, starter: null }) + } + /> + )} +     + {editor ? ( + + setEditor({ environmentId, pipeline: null, starter: editor.starter }) + } + onClose={() => setEditor(null)} + /> + ) : null} +     + ); +} + +const pipelinesListAtom = environmentAtomList("web-pipelines-list", (environmentId) => + serverEnvironment.pipelinesLive({ environmentId, input: {} }), +); + +function PipelinesEnvironmentList({ + environments, + onEdit, +}: { + readonly environments: ReadonlyArray; + readonly onEdit: (environmentId: EnvironmentId, pipeline: PipelineDefinition) => void; +}) { + // Only environments that answer pipelines.* are subscribed; the rest get a note instead. + const subscribedIds = environments.flatMap((entry) => + supportsPipelines(entry) ? [entry.environmentId] : [], + ); + const queries = useAtomValue(pipelinesListAtom(environmentIdsKey(subscribedIds))); + const showEnvironmentTitle = environments.length > 1; + + return ( +
    + {environments.map((entry) => { + const index = subscribedIds.indexOf(entry.environmentId); + const query = index < 0 ? null : queries[index]!; + const data = query === null ? null : AsyncResult.getOrElse(query, () => null); + return ( + + } + > + {entry.connection.phase !== "connected" || entry.serverConfig === null ? ( + + ) : !supportsPipelines(entry) ? ( + + ) : query?._tag === "Failure" ? ( + + ) : !data ? ( + + ) : data.pipelines.length === 0 ? ( + + ) : ( + data.pipelines.map((pipeline) => ( + onEdit(entry.environmentId, pipeline)} + /> + )) + )} + + ); + })} +
    + ); +} + +function PipelineRow({ + environmentId, + pipeline, + onEdit, +}: { + readonly environmentId: EnvironmentId; + readonly pipeline: PipelineDefinition; + readonly onEdit: () => void; +}) { + const canEdit = useAtomValue(serverEnvironment.upsertPipeline.permissionAtom(environmentId)); + const canDelete = useAtomValue(serverEnvironment.deletePipeline.permissionAtom(environmentId)); + const providers = + useAtomValue(serverEnvironment.providersValueAtom(environmentId)) ?? EMPTY_SERVER_PROVIDERS; + const [busy, setBusy] = useState(false); + const remove = useAtomCommand(serverEnvironment.deletePipeline, { + label: "pipeline delete", + reportFailure: false, + }); + const summary = pipelineStepSummary(pipeline.steps, (selection) => { + const entry = deriveProviderInstanceEntries(providers).find( + (candidate) => candidate.instanceId === selection.instanceId, + ); + const model = entry?.models.find((candidate) => candidate.slug === selection.model); + return model ? (model.shortName ?? model.name) : selection.model; + }); + + const confirmDelete = async () => { + if (busy || !readEnvironmentScope(environmentId, AuthOrchestrationOperateScope)) return; + const message = `Delete "${pipeline.name}"?\nRuns already started keep their own copy of the steps.`; + const confirmed = + (await requestConfirmDialog(message, { variant: "destructive" })) ?? window.confirm(message); + if (!confirmed) return; + setBusy(true); + const result = await remove({ environmentId, input: { id: pipeline.id } }); + setBusy(false); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + reportFailure("Could not delete pipeline", squashAtomCommandFailure(result)); + } + }; + + return ( + {summary}} + status={ + + {pipeline.steps.length} {pipeline.steps.length === 1 ? "step" : "steps"} + + } + control={ +
    + + +
    + } + /> + ); +} + +function PipelineEditorDialog(props: { + readonly environmentId: EnvironmentId; + readonly pipeline: PipelineDefinition | null; + readonly starter: PipelineStarter | null; + readonly capableEnvironments: ReadonlyArray; + readonly onEnvironmentChange: (environmentId: EnvironmentId) => void; + readonly onClose: () => void; +}) { + const [saving, setSaving] = useState(false); + return ( + { + if (!open && !saving) props.onClose(); + }} + > + {/* The form owns the draft so a keystroke re-renders it, not the dialog chrome. */} + + + + + ); +} + +function PipelineEditorForm({ + environmentId, + pipeline, + starter, + capableEnvironments, + onEnvironmentChange, + onClose, + saving, + onSavingChange: setSaving, +}: { + readonly environmentId: EnvironmentId; + readonly pipeline: PipelineDefinition | null; + readonly starter: PipelineStarter | null; + readonly capableEnvironments: ReadonlyArray; + readonly onEnvironmentChange: (environmentId: EnvironmentId) => void; + readonly onClose: () => void; + readonly saving: boolean; + readonly onSavingChange: (saving: boolean) => void; +}) { + const environment = useEnvironment(environmentId); + const connected = environment !== null && supportsPipelines(environment); + const settings = useEnvironmentSettings(environmentId); + const providers = + useAtomValue(serverEnvironment.providersValueAtom(environmentId)) ?? EMPTY_SERVER_PROVIDERS; + const canOperate = useAtomValue(serverEnvironment.upsertPipeline.permissionAtom(environmentId)); + const upsertPipeline = useAtomCommand(serverEnvironment.upsertPipeline, { + label: "pipeline upsert", + reportFailure: false, + }); + const instanceEntries = providerInstanceEntriesFor(providers, settings, null); + const defaults: PipelineStepDefaults = { + modelSelection: scheduledTaskDefaultModel(settings, null, instanceEntries), + runtimeMode: settings.defaultRuntimeMode, + }; + const [draft, setDraft] = useState(() => + pipeline + ? pipelineToDraft(pipeline, randomUUID) + : starter + ? starterToDraft(starter, randomUUID) + : emptyPipelineDraft(defaults, randomUUID), + ); + const submissionPending = useRef(false); + const issues = pipelineDraftIssues(draft); + const canSubmit = issues.valid && connected && canOperate && !saving; + + const updateStep = (key: string, patch: Partial) => + setDraft((current) => ({ + ...current, + steps: current.steps.map((step) => (step.key === key ? { ...step, ...patch } : step)), + })); + + const submit = async () => { + const input = draftToUpsertInput(draft); + if ( + input === null || + !readEnvironmentScope(environmentId, AuthOrchestrationOperateScope) || + submissionPending.current || + !connected + ) + return; + // Lock before React renders, and keep successful saves locked until the form closes. + submissionPending.current = true; + setSaving(true); + const result = await upsertPipeline({ environmentId, input }); + setSaving(false); + if (result._tag === "Failure") { + submissionPending.current = false; + if (!isAtomCommandInterrupted(result)) { + reportFailure("Could not save pipeline", squashAtomCommandFailure(result)); + } + return; + } + onClose(); + }; + + return ( + <> + + {pipeline ? "Edit pipeline" : "New pipeline"} + + Run steps one after another, each in its own thread with its own model and prompt. + + + + + {/* Fieldsets default to min-width: min-content, which lets one wide control widen the form past the dialog. */} +
    + {!connected ? ( +

    + Reconnect this environment, on a server that supports pipelines, before saving. +

    + ) : null} + {capableEnvironments.length > 1 ? ( + + + + ) : null} + + 0 ? true : undefined} + value={draft.name} + onChange={(event) => { + const name = event.target.value; + setDraft((current) => ({ ...current, name })); + }} + /> + + +
    +
    + + +
    + {issues.stepCount ? ( +

    + {issues.stepCount} +

    + ) : null} + {draft.steps.map((step, index) => ( + updateStep(step.key, patch)} + onMove={(delta) => + setDraft((current) => ({ + ...current, + steps: moveStep(current.steps, index, delta), + })) + } + onRemove={() => + setDraft((current) => ({ + ...current, + steps: current.steps.filter((entry) => entry.key !== step.key), + })) + } + /> + ))} +
    +
    +     + + + }> + Cancel + + + + + ); +} + +function PipelineStepEditor({ + environmentId, + step, + index, + count, + issues, + instanceEntries, + providers, + settings, + disabled, + onChange, + onMove, + onRemove, +}: { + readonly environmentId: EnvironmentId; + readonly step: PipelineStepDraft; + readonly index: number; + readonly count: number; + readonly issues: ReadonlyArray; + readonly instanceEntries: ReadonlyArray; + readonly providers: ReadonlyArray; + readonly settings: UnifiedSettings; + readonly disabled: boolean; + readonly onChange: (patch: Partial) => void; + readonly onMove: (delta: -1 | 1) => void; + readonly onRemove: () => void; +}) { + const selection = step.modelSelection; + const activeEntry = selection + ? (instanceEntries.find((entry) => entry.instanceId === selection.instanceId) ?? null) + : null; + const activeInstanceId = + selection?.instanceId ?? instanceEntries[0]?.instanceId ?? ("" as ProviderInstanceId); + const activeModel = selection?.model ?? ""; + const modelOptionsByInstance = getCustomModelOptionsByInstance( + settings, + providers, + activeInstanceId, + activeModel, + ); + const setModel = (next: ModelSelection) => onChange({ modelSelection: next }); + const idPrefix = `pipeline-step-${step.key}`; + + return ( +
    +
    + Step {index + 1} +
    + + + +
    +
    + + + onChange({ name: event.target.value })} + /> + + + +
    + + setModel(createModelSelection(instanceId, model)) + } + /> + {selection && activeEntry ? ( + + setModel(createModelSelection(selection.instanceId, selection.model, options)) + } + /> + ) : selection ? ( + + This provider is unavailable here. Pick another model. + + ) : null} +
    +     + + + + + + onChange({ skills })} + /> + + +