diff --git a/backend/tests/v2_job_delete_orphans.rs b/backend/tests/v2_job_delete_orphans.rs index adae746d09..8c8976c420 100644 --- a/backend/tests/v2_job_delete_orphans.rs +++ b/backend/tests/v2_job_delete_orphans.rs @@ -276,13 +276,8 @@ async fn test_workspace_delete_removes_side_rows(db: Pool) -> anyhow:: Ok(()) } -/// The `/jobs/delete` purge endpoint must scope every side-table delete to the path -/// workspace. A `test-workspace` admin passing a job id from another workspace must not be -/// able to delete that workspace's job or side rows (the side tables no longer cascade, so -/// the scoping has to live in each explicit delete). /// The purge endpoint carries its own copy of the emptied-conversation rule, so it gets the -/// same guard: the conversation and its memory go only with the last message, and only in -/// the caller's workspace. +/// same guard: the conversation and its memory go with the last message, and not before. #[sqlx::test(fixtures("base"))] async fn test_jobs_export_delete_removes_a_conversation_once_its_last_message_goes( db: Pool, @@ -349,6 +344,10 @@ async fn test_jobs_export_delete_removes_a_conversation_once_its_last_message_go Ok(()) } +/// The `/jobs/delete` purge endpoint must scope every side-table delete to the path +/// workspace. A `test-workspace` admin passing a job id from another workspace must not be +/// able to delete that workspace's job or side rows (the side tables no longer cascade, so +/// the scoping has to live in each explicit delete). #[sqlx::test(fixtures("base"))] async fn test_jobs_export_delete_is_workspace_scoped(db: Pool) -> anyhow::Result<()> { initialize_tracing().await; diff --git a/backend/windmill-api-jobs/src/jobs_export.rs b/backend/windmill-api-jobs/src/jobs_export.rs index aacd99f8aa..8b80edc349 100644 --- a/backend/windmill-api-jobs/src/jobs_export.rs +++ b/backend/windmill-api-jobs/src/jobs_export.rs @@ -692,7 +692,9 @@ pub async fn delete_jobs( .await? .rows_affected(); - let emptied_conversations: Vec = sqlx::query_scalar!( + // One row per message deleted, so the conversation of a chat losing several appears + // several times: the count is taken before the dedup below. + let mut conversation_ids: Vec = sqlx::query_scalar!( "DELETE FROM flow_conversation_message m USING flow_conversation c WHERE m.conversation_id = c.id AND c.workspace_id = $1 AND m.job_id = ANY($2) @@ -702,11 +704,10 @@ pub async fn delete_jobs( ) .fetch_all(&mut *tx) .await?; - let conversation_message_deleted = emptied_conversations.len() as u64; + let conversation_message_deleted = conversation_ids.len() as u64; // Same rule as retention (windmill_common::jobs::delete_jobs): a conversation with no // messages left goes, and the agent's memory for it with it. - let mut conversation_ids = emptied_conversations; conversation_ids.sort_unstable(); conversation_ids.dedup(); if !conversation_ids.is_empty() { diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 236c9a4c1d..74a4071b21 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -27603,7 +27603,7 @@ components: FlowConversation: type: object required: - [id, workspace_id, flow_path, created_at, updated_at, created_by] + [id, workspace_id, flow_path, created_at, updated_at, created_by, is_test] properties: id: type: string diff --git a/backend/windmill-worker/src/ai/utils.rs b/backend/windmill-worker/src/ai/utils.rs index 2f7767fdf8..198e91750c 100644 --- a/backend/windmill-worker/src/ai/utils.rs +++ b/backend/windmill-worker/src/ai/utils.rs @@ -13,7 +13,7 @@ use windmill_common::flows::FlowModuleValue; use windmill_common::{ db::DB, error::Error, - flow_conversations::{add_message_to_conversation_tx, MessageType, MessageExtras}, + flow_conversations::{add_message_to_conversation_tx, MessageExtras, MessageType}, flow_status::AgentAction, flows::{InputTransform, Step}, jobs::JobKind, diff --git a/backend/windmill-worker/src/ai_executor.rs b/backend/windmill-worker/src/ai_executor.rs index 86fd4a2c64..0d3524beee 100644 --- a/backend/windmill-worker/src/ai_executor.rs +++ b/backend/windmill-worker/src/ai_executor.rs @@ -1457,12 +1457,9 @@ pub async fn run_agent( let step_name = step_name.clone(); // The thinking is streamed and never returned in a response // body, so the answer's row is the only place it can be kept. - let extras = response_reasoning - .clone() - .map(|reasoning| MessageExtras { - reasoning: Some(reasoning), - ..Default::default() - }); + let extras = response_reasoning.clone().map(|reasoning| { + MessageExtras { reasoning: Some(reasoning), ..Default::default() } + }); // Spawn task because we do not need to wait for the result tokio::spawn(async move { diff --git a/frontend/src/lib/components/copilot/ChatModelSettings.svelte b/frontend/src/lib/components/copilot/ChatModelSettings.svelte index aa1b7ce8e4..e60e4f1181 100644 --- a/frontend/src/lib/components/copilot/ChatModelSettings.svelte +++ b/frontend/src/lib/components/copilot/ChatModelSettings.svelte @@ -12,12 +12,12 @@ import MenuItemWrapper from '$lib/components/meltComponents/MenuItemWrapper.svelte' import Button from '$lib/components/common/button/Button.svelte' import ReasoningEffortSlider from './ReasoningEffortSlider.svelte' + import { getReasoningCapability, resolveEffectiveReasoning } from './reasoningRegistry' import { - getReasoningCapability, - resolveEffectiveReasoning, - REASONING_OFF - } from './reasoningRegistry' - import type { ChatModelSettingsConfig, ChoiceSection } from './chatModelSettings' + reasoningDisplay, + type ChatModelSettingsConfig, + type ChoiceSection + } from './chatModelSettings' import type { Item } from '$lib/utils' import type { MenubarMenuElements, createDropdownMenu } from '@melt-ui/svelte' import { twMerge } from 'tailwind-merge' @@ -33,12 +33,6 @@ ? getReasoningCapability(reasoning.provider, reasoning.model) : { supported: false, levels: [] as string[], canDisable: false } ) - // An off position only where the model can truly disable, else the provider would - // coerce it to the lowest level; then the provider-native levels. - const stops = $derived([ - ...(capability.canDisable && reasoning?.offToken !== undefined ? [reasoning.offToken] : []), - ...capability.levels - ]) // Effective effort accounts for the default-on level on capable models. const effective = $derived( reasoning @@ -49,15 +43,12 @@ }) : undefined ) - const currentStop = $derived( - reasoning && reasoning.value === reasoning.offToken - ? (reasoning.offToken ?? '') - : (effective ?? stops[stops.length - 1] ?? '') - ) - // Trigger suffix: the effort token, or 'off' when disabled. Omitted entirely for - // models with no reasoning support. The off token is provider-native and can read as - // anything ('none', 'disabled'); on the button it always reads as off. - const effortLabel = $derived(capability.supported ? (effective ?? REASONING_OFF) : undefined) + // The stops, the one in use and the trigger's suffix are decided together, in one + // tested place: they have to agree, and three rounds of review found them disagreeing. + const display = $derived(reasoningDisplay(reasoning, capability, effective)) + const stops = $derived(display.stops) + const currentStop = $derived(display.currentStop) + const effortLabel = $derived(display.label) let effortSlider: ReasoningEffortSlider | undefined = $state(undefined) @@ -197,7 +188,7 @@ {/each} {#if reasoning}
- {#if capability.supported && stops.length > 1} + {#if capability.supported} @@ -212,9 +203,12 @@ current={currentStop} onSelect={reasoning.onSelect} format={(stop) => (stop === reasoning?.offToken ? 'off' : stop)} + overrideLabel={stops.includes(currentStop) ? undefined : effortLabel} /> {:else} + & { provider: any; model: string } +) { + const full = { + value: undefined, + offToken: undefined, + sendsDefaultWhenUnset: false, + onSelect: () => {}, + ...reasoning + } as ChatModelSettingsReasoning + const capability = getReasoningCapability(full.provider, full.model) + // Composed exactly as the component composes it, so the test exercises the real pair. + const effective = resolveEffectiveReasoning({ + provider: full.provider, + model: full.model, + reasoning: full.value + }) + return reasoningDisplay(full, capability, effective) +} + +describe('reasoningDisplay', () => { + it('says nothing for a model that cannot reason', () => { + const shown = display({ provider: 'openai', model: 'gpt-4o' }) + expect(shown.label).toBeUndefined() + expect(shown.stops).toEqual([]) + }) + + // The session chat's own sentinel: what it stores is already the word the reader sees. + it('reads the session chat off sentinel as off', () => { + const shown = display({ + provider: 'openai', + model: 'gpt-5.1', + offToken: REASONING_OFF, + value: REASONING_OFF, + sendsDefaultWhenUnset: true + }) + expect(shown.label).toBe(REASONING_OFF) + expect(shown.currentStop).toBe(REASONING_OFF) + }) + + // An agent writes the provider's own token, which can read as anything. + it('reads a provider-native off token as off too', () => { + const shown = display({ + provider: 'openai', + model: 'gpt-5.1', + offToken: 'none', + value: 'none' + }) + expect(shown.label).toBe(REASONING_OFF) + expect(shown.currentStop).toBe('none') + expect(shown.stops[0]).toBe('none') + }) + + it('names the level a chat that fills one in will send', () => { + const shown = display({ + provider: 'openai', + model: 'gpt-5.1', + offToken: REASONING_OFF, + value: undefined, + sendsDefaultWhenUnset: true + }) + expect(shown.label).toBe('high') + }) + + // An agent step omits the field, so naming a level would claim something untrue. + it('names no level where an unset effort is simply not sent', () => { + const shown = display({ + provider: 'anthropic', + model: 'claude-sonnet-5', + offToken: 'none', + value: undefined + }) + expect(shown.label).toBe(REASONING_PROVIDER_DEFAULT) + expect(shown.currentStop).toBe('') + }) + + // Claude 4.x only thinks when asked, so an absent effort is already off — and the flow + // chat must be able to get back to it after a level has been picked. + it('offers omission as the off stop where that is how the model disables', () => { + const unset = display({ provider: 'anthropic', model: 'claude-opus-4-6', offToken: '' }) + expect(unset.label).toBe(REASONING_OFF) + expect(unset.stops[0]).toBe('') + const picked = display({ + provider: 'anthropic', + model: 'claude-opus-4-6', + offToken: '', + value: 'high' + }) + expect(picked.currentStop).toBe('high') + expect(picked.stops).toContain('') + }) + + // gpt-5 reasons at medium with no effort sent, so an empty off token buys no off stop. + it('offers no off where the model cannot stop thinking', () => { + const shown = display({ provider: 'openai', model: 'gpt-5', offToken: '' }) + expect(shown.stops).not.toContain('') + expect(shown.label).toBe(REASONING_PROVIDER_DEFAULT) + }) +}) diff --git a/frontend/src/lib/components/copilot/chatModelSettings.ts b/frontend/src/lib/components/copilot/chatModelSettings.ts index db14cb6616..4e8c6bda58 100644 --- a/frontend/src/lib/components/copilot/chatModelSettings.ts +++ b/frontend/src/lib/components/copilot/chatModelSettings.ts @@ -1,5 +1,6 @@ import type { AIProvider } from '$lib/gen' import type { Item } from '$lib/utils' +import { REASONING_OFF } from './reasoningRegistry' /** * The contract between a chat and its model button. @@ -57,11 +58,67 @@ export type ChatModelSettingsConfig = { * sentinel and translates when it calls the provider, an agent writes the * provider-native token straight into its step. */ - reasoning?: { - provider: AIProvider - model: string - value: string | undefined - offToken: string | undefined - onSelect: (token: string) => void + reasoning?: ChatModelSettingsReasoning +} + +export type ChatModelSettingsReasoning = { + provider: AIProvider + model: string + value: string | undefined + offToken: string | undefined + /** + * What an unset value means on the wire. The copilot fills one in before calling the + * provider, so unset really runs at the default effort and the button says so. An agent + * step omits the field entirely, so unset means whatever the provider does by itself — + * naming a level there would state something the run does not do. + */ + sendsDefaultWhenUnset: boolean + onSelect: (token: string) => void +} + +/** What an unset agent effort reads as: the provider decides, and we do not know what. */ +export const REASONING_PROVIDER_DEFAULT = 'default' + +/** + * What the menu shows for the reasoning ladder: the stops the slider offers, the one it + * sits on, and the suffix on the trigger. + * + * Pure and here rather than in the component because these three have to agree — a stop + * the slider renders as `off` must not read as the provider's own `none` on the button — + * and because the rules are provider-shaped enough to be worth testing directly. + */ +export function reasoningDisplay( + reasoning: ChatModelSettingsReasoning | undefined, + capability: { supported: boolean; levels: string[]; canDisable: boolean }, + effective: string | undefined +): { + stops: string[] + currentStop: string + /** Trigger suffix, or undefined when there is nothing truthful to say. */ + label: string | undefined +} { + if (!reasoning || !capability.supported) { + return { stops: [], currentStop: '', label: undefined } } + // An off position only where the model can truly disable, else the provider would + // coerce it to the lowest level; then the provider-native levels. + const offToken = capability.canDisable ? reasoning.offToken : undefined + const stops = [...(offToken !== undefined ? [offToken] : []), ...capability.levels] + // An agent whose model disables by omission stores the empty string, which the run + // treats as no effort at all — so an unset value already sits on that stop. + const isOff = offToken !== undefined && (reasoning.value ?? '') === offToken + if (isOff) { + // The off token is provider-native and can read as anything ('none', 'disabled'); + // on the button and on the slider it always reads as off. + return { stops, currentStop: offToken as string, label: REASONING_OFF } + } + if (reasoning.value === undefined || reasoning.value === '') { + // Where the provider takes an explicit disable, unset is a third state — the + // provider's own level, above off — that the ladder has no position for. The + // button still names it, and every stop the ladder does offer stays reachable. + return reasoning.sendsDefaultWhenUnset + ? { stops, currentStop: effective ?? '', label: effective ?? REASONING_OFF } + : { stops, currentStop: '', label: REASONING_PROVIDER_DEFAULT } + } + return { stops, currentStop: reasoning.value, label: reasoning.value } } diff --git a/frontend/src/lib/components/flows/conversations/FlowChat.svelte b/frontend/src/lib/components/flows/conversations/FlowChat.svelte index 52fe2af124..495df5d9ab 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChat.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowChat.svelte @@ -45,6 +45,8 @@ const manager = createFlowChatManager() manager.operatingWorkspace = () => flowEditorContext?.opWorkspace?.() manager.conversationKind = conversationKind + // The filter moves; what this surface runs does not. + manager.surfaceKind = conversationKind === 'test' ? 'test' : 'deployed' // The editor is the only surface with both kinds in play, and it is the one that opens // on test chats. A deployed flow lists what its users started, with no way to ask for // anything else. @@ -54,7 +56,10 @@ $effect(() => { if ($workspaceStore) { manager.initialize(onRunFlow, path, useStreaming) - void manager.selectLatestConversation() + // Reads the open conversation, and this effect tears down with `cleanup()`: tracked, + // the first send of a fresh chat would select the conversation it just created and + // so abort its own turn. + untrack(() => manager.selectLatestConversation()) } return () => { diff --git a/frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte b/frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte index ea8f46587f..9a0cf9131f 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowChatInterface.svelte @@ -8,13 +8,15 @@ import Modal from '$lib/components/common/modal/Modal.svelte' import SchemaForm from '$lib/components/SchemaForm.svelte' import { type DynamicInput } from '$lib/utils' - import { type FlowModule } from '$lib/gen' + import { type AIProvider, type FlowModule } from '$lib/gen' + import { getReasoningCapability } from '$lib/components/copilot/reasoningRegistry' import { useWorkspaceStorageConfigured } from '$lib/components/inputTransformEnv.svelte' import { workspaceStore } from '$lib/stores' import FlowChatModelSettings from './FlowChatModelSettings.svelte' import { agentModelGap, agentModelWiringInputs, + attachmentsTargetFor, isEmptyAgentChatInputValue, PER_TURN_AGENT_CHAT_INPUT_KEY, resolveAgentChatInputs, @@ -50,19 +52,14 @@ return undefined }) - // Inputs an AI agent step reads straight out of the flow input get a composer chip - // instead of a modal field; the rest stay in the modal. + // The flow inputs an AI agent step reads straight out of `flow_input`, which the + // composer may then edit itself instead of asking for them in the modal. const agentChatInputs = $derived(resolveAgentChatInputs(flowModules, additionalInputsSchema)) - // The composer's attachments feed this input; it never appears as a chip or a - // modal field, because the paperclip is its editor. + // The composer's attachments feed this input, and the paperclip is its whole editor. const attachmentsInput = $derived( agentChatInputs.find((input) => input.key === PER_TURN_AGENT_CHAT_INPUT_KEY) ) - const attachmentsTarget = $derived( - attachmentsInput - ? { name: attachmentsInput.name, multiple: attachmentsInput.property?.type === 'array' } - : undefined - ) + const attachmentsTarget = $derived(attachmentsTargetFor(attachmentsInput)) const chatWorkspace = $derived(manager.operatingWorkspace?.() ?? $workspaceStore) @@ -77,26 +74,6 @@ // chat says what to go and do instead of offering controls that write nowhere. const modelGap = $derived(agentModelGap(modelWiring)) - const modalSchema = $derived.by(() => { - if (!additionalInputsSchema) return undefined - const promoted = new Set([ - ...agentChatInputs.map((input) => input.name), - ...agentModelWiringInputs(modelWiring) - ]) - const properties = Object.fromEntries( - Object.entries(additionalInputsSchema.properties ?? {}).filter(([key]) => !promoted.has(key)) - ) - if (Object.keys(properties).length === 0) return undefined - const required: string[] = Array.isArray(additionalInputsSchema.required) - ? additionalInputsSchema.required - : [] - return { - ...additionalInputsSchema, - properties, - required: required.filter((key) => !promoted.has(key)) - } - }) - // LocalStorage helpers const STORAGE_KEY_PREFIX = 'windmill_flow_chat_inputs_' @@ -124,6 +101,20 @@ ...inputValues }) + // Whether the model button will offer the thinking slider, which it does only for a + // model that reasons. A provider kind or model we cannot read leaves it unknown, and + // the field then stays in the modal rather than behind a control that never appears. + const effortEditable = $derived.by(() => { + if (!modelWiring || modelWiring.whole) return false + const pick = (field: 'model' | 'kind') => { + const name = modelWiring.fields[field] + return name ? effectiveInputs[name] : modelWiring.fixed[field] + } + const [kind, model] = [pick('kind'), pick('model')] + if (typeof kind !== 'string' || !kind || typeof model !== 'string' || !model) return false + return getReasoningCapability(kind as AIProvider, model).supported + }) + function getStorageKey(): string { return `${STORAGE_KEY_PREFIX}${path}` } @@ -167,15 +158,39 @@ attachmentsTarget: () => attachmentsTarget, workspace: () => chatWorkspace, canAttach: () => workspaceStorage.current, - inputsShownInComposer: () => agentModelWiringInputs(modelWiring) + inputsShownInComposer: () => agentModelWiringInputs(modelWiring, effortEditable), + inputsSchema: () => additionalInputsSchema }) setChatViewHost(chatHost) - // A message held mid-run goes out once the run settles. Attachments count as a - // message of their own, so a queue with files and no text still has to flush. - const hasQueuedTurn = $derived( - !!chatHost.queuedMessage || chatHost.queuedImages.length > 0 || chatHost.queuedBlobs.length > 0 - ) + // A message held mid-run goes out once the run settles. Keyed on the text alone, as + // `flushQueuedMessage` is: a turn here cannot run without one, so the composer refuses + // an attachment-only send rather than queueing files nothing would ever drain. + const hasQueuedTurn = $derived(!!chatHost.queuedMessage.trim()) + + // What the Configure-inputs modal asks for: every flow input the composer does not + // edit itself. Below the host, because whether the paperclip is offered is its answer. + const modalSchema = $derived.by(() => { + if (!additionalInputsSchema) return undefined + const promoted = new Set([ + // The paperclip's own condition, not half of it: an attachments input the composer + // has no editor for — no object storage in the workspace, say — stays askable here. + ...(chatHost.supportsMessageAttachments && attachmentsTarget ? [attachmentsTarget.name] : []), + ...agentModelWiringInputs(modelWiring, effortEditable) + ]) + const properties = Object.fromEntries( + Object.entries(additionalInputsSchema.properties ?? {}).filter(([key]) => !promoted.has(key)) + ) + if (Object.keys(properties).length === 0) return undefined + const required: string[] = Array.isArray(additionalInputsSchema.required) + ? additionalInputsSchema.required + : [] + return { + ...additionalInputsSchema, + properties, + required: required.filter((key) => !promoted.has(key)) + } + }) $effect(() => { if (!chatHost.loading && hasQueuedTurn) { chatHost.flushQueuedMessage() @@ -268,8 +283,10 @@ {emptyHint} footerSettings={modalSchema || modelWiring ? footerSettings : undefined} placeholder="Send a message to run the flow" - disabled={deploymentInProgress || !!modelGap} - disabledMessage={deploymentInProgress ? 'Deployment in progress' : (modelGap ?? '')} + disabled={deploymentInProgress || !!modelGap || !!manager.wrongKindReason} + disabledMessage={deploymentInProgress + ? 'Deployment in progress' + : (modelGap ?? manager.wrongKindReason ?? '')} loadPastChat={() => {}} deletePastChat={() => {}} saveAndClear={() => {}} diff --git a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts index bd6854ed1a..d4ec2a8a5e 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/FlowChatManager.svelte.ts @@ -115,6 +115,13 @@ export class FlowChatManager { * would put editor scratch in front of the flow's users. */ canFilterConversationKind = $state(false) + /** + * What this surface's own runs are, which the filter does not change: the editor runs + * previews, the flow page runs the deployed flow. A conversation is fixed to one kind + * at creation, so a turn sent from here into a conversation of the other kind would be + * stored as part of it and the mixing would be invisible afterwards. + */ + surfaceKind = $state>('deployed') selectedConversationId = $state(undefined) conversationListComponent = $state(undefined) @@ -211,6 +218,8 @@ export class FlowChatManager { created_at: new Date().toISOString(), updated_at: new Date().toISOString(), created_by: get(userStore)!.username!, + // The kind the first turn will give it: a draft started here runs on this surface. + is_test: this.surfaceKind === 'test', isDraft: true } @@ -338,6 +347,19 @@ export class FlowChatManager { this.isDispatchingTurn = false } + /** + * Why the composer must stay shut, when the open conversation belongs to the other + * surface. Reading such a chat is fine; adding to it from here is not. + */ + get wrongKindReason(): string | undefined { + const open = this.conversations.find((c) => c.id === this.selectedConversationId) + if (!open || open.isDraft || (open.is_test === true) === (this.surfaceKind === 'test')) + return undefined + return this.surfaceKind === 'test' + ? 'This chat belongs to the deployed flow. Start a new chat to test.' + : 'This chat was run from the editor. Start a new chat to continue here.' + } + /** A turn is being dispatched or is running: nothing may move the conversation under it. */ get isTurnInFlight(): boolean { return this.isLoading || this.isWaitingForResponse || this.isDispatchingTurn @@ -604,8 +626,8 @@ export class FlowChatManager { * conversation while it runs, and the turn belongs to the one they sent it from. */ pinnedConversationId?: string - ) { - if (this.isLoading) return + ): Promise { + if (this.isLoading) return false const isNewConversation = this.messages.length === 0 @@ -621,7 +643,7 @@ export class FlowChatManager { if (!currentConversationId) { console.error('No conversation ID found') - return + return false } // Invalidate the conversation cache @@ -643,19 +665,22 @@ export class FlowChatManager { this.isLoading = true this.isWaitingForResponse = true + // This turn's own answer, not shared state: a queued follow-up can flush while this + // one is still finishing, and re-enter sendMessage before it reads the result. + let started = false try { await tick() this.scrollToUserMessage(userMessage.id) if (this.#useStreaming && this.#path) { - await this.handleStreamingMessage( + started = await this.handleStreamingMessage( messageContent, currentConversationId, isNewConversation, additionalInputs ) } else { - await this.handlePollingMessage( + started = await this.handlePollingMessage( messageContent, currentConversationId, isNewConversation, @@ -669,6 +694,7 @@ export class FlowChatManager { // the finally because the streaming path keeps `isLoading` for its own stream, // and without this the composer and the sidebar stay locked until a reload. this.#turnFailedToStart() + started = false } finally { if (!this.#useStreaming) { this.isLoading = false @@ -680,7 +706,9 @@ export class FlowChatManager { // renameConversation: on the streaming path this runs before any refresh, so the // local row still carries the typed name and an equality check would skip the write. // Cleared only once it lands, so a failed run keeps the name for the next attempt. - if (this.#draftTitle?.id === currentConversationId) { + // Only when a turn actually ran: nothing created the row otherwise, so the write + // would 404 and stack a rename failure on top of the real one. + if (started && this.#draftTitle?.id === currentConversationId) { const { title } = this.#draftTitle if (await this.#writeConversationTitle(currentConversationId, title)) { this.#draftTitle = undefined @@ -689,14 +717,22 @@ export class FlowChatManager { await tick() this.focusInput() + if (!started) { + // Nothing ran, so the row claiming a turn has to go with it — the caller puts + // the message back in the composer. + this.messages = this.messages.filter((m) => m.id !== userMessage.id) + return false + } + return true } + /** Answers whether a job was actually started. */ private async handleStreamingMessage( messageContent: string, currentConversationId: string, isNewConversation: boolean, additionalInputs?: Record - ) { + ): Promise { // Close any existing EventSource if (this.currentEventSource) { this.currentEventSource.close() @@ -713,7 +749,7 @@ export class FlowChatManager { if (!jobId) { console.error('No jobId returned from onRunFlow') this.#turnFailedToStart() - return + return false } // Build the EventSource URL @@ -835,23 +871,30 @@ export class FlowChatManager { this.cleanup() } } catch (error) { + // Everything that can throw here happens before the stream is live — the run + // request itself (which the deployed page's launcher throws from), or building + // the EventSource. Either way no turn ran. console.error('Stream connection error:', error) sendUserToast('Failed to connect to stream', true) this.cleanup() + this.#turnFailedToStart() + return false } + return true } + /** Answers whether a job was actually started. */ private async handlePollingMessage( messageContent: string, currentConversationId: string, isNewConversation: boolean, additionalInputs?: Record - ) { + ): Promise { const jobId = await this.#onRunFlow?.(messageContent, currentConversationId, additionalInputs) if (!jobId) { console.error('No jobId returned from onRunFlow') this.#turnFailedToStart() - return + return false } // Store the current job ID so it can be cancelled @@ -864,6 +907,7 @@ export class FlowChatManager { // Start polling for intermediate messages in non-streaming mode too this.startPolling(currentConversationId) this.pollJobResult(jobId) + return true } } diff --git a/frontend/src/lib/components/flows/conversations/FlowChatModelSettings.svelte b/frontend/src/lib/components/flows/conversations/FlowChatModelSettings.svelte index d878ac703e..8ebc550104 100644 --- a/frontend/src/lib/components/flows/conversations/FlowChatModelSettings.svelte +++ b/frontend/src/lib/components/flows/conversations/FlowChatModelSettings.svelte @@ -213,8 +213,13 @@ model, value: typeof effort === 'string' ? effort : undefined, // An agent writes the provider-native token straight into its step, so - // there is no sentinel to translate later. - offToken: explicitOffToken(provider, model), + // there is no sentinel to translate later. Where a model disables by + // omission instead, the empty string is that off: the run reads an + // empty `reasoning_effort` as absent (types.rs `get_reasoning_effort`). + offToken: explicitOffToken(provider, model) ?? '', + // An agent step omits `reasoning_effort` when it is unset, so the provider + // picks — naming a level would claim something the run does not do. + sendsDefaultWhenUnset: false, onSelect: (token) => setFields({ reasoning_effort: token }) } : undefined diff --git a/frontend/src/lib/components/flows/conversations/agentChatInputs.test.ts b/frontend/src/lib/components/flows/conversations/agentChatInputs.test.ts index bb848c3bd2..c0b6b301f8 100644 --- a/frontend/src/lib/components/flows/conversations/agentChatInputs.test.ts +++ b/frontend/src/lib/components/flows/conversations/agentChatInputs.test.ts @@ -1,6 +1,8 @@ import { describe, expect, it } from 'vitest' import { agentModelGap, + agentModelWiringInputs, + attachmentsTargetFor, parseProviderTransform, resolveAgentChatInputs, resolveAgentModelWiring @@ -130,6 +132,36 @@ describe('resolveAgentChatInputs', () => { }) }) +describe('agentModelWiringInputs', () => { + // The button writes `kind` only alongside a resource, since a provider is picked as a + // pair. Hiding a kind input it cannot write would leave the run without one. + it('keeps a kind input the model button cannot write', () => { + const wiring = resolveAgentModelWiring([ + agent(`({ kind: flow_input.k, "resource": "$res:u/admin/claude", model: flow_input.m })`) + ]) + expect(agentModelWiringInputs(wiring)).toEqual(['m']) + }) + + it('hides a kind input it writes with the resource', () => { + const wiring = resolveAgentModelWiring([ + agent(`({ kind: flow_input.k, resource: flow_input.r, model: flow_input.m })`) + ]) + expect(agentModelWiringInputs(wiring, true)?.sort()).toEqual(['k', 'm', 'r']) + }) + + // The slider only appears for a model that reasons, so on one that does not the effort + // input has no editor on the button and has to stay in the modal. + it('keeps a reasoning_effort input where the model cannot think', () => { + const wiring = resolveAgentModelWiring([ + agent( + `({ "kind": "openai", "resource": "$res:u/admin/oai", "model": "gpt-4o", reasoning_effort: flow_input.thinking })` + ) + ]) + expect(agentModelWiringInputs(wiring, false)).toEqual([]) + expect(agentModelWiringInputs(wiring, true)).toEqual(['thinking']) + }) +}) + describe('resolveAgentModelWiring', () => { const fixedResource = `"kind": "anthropic", "resource": "$res:u/admin/claude"` @@ -240,3 +272,31 @@ describe('resolveAgentModelWiring', () => { ).toBeUndefined() }) }) + +describe('attachmentsTargetFor', () => { + const input = (property: Record) => + ({ name: 'files', key: 'user_attachments', property, required: false }) as any + + it('takes a list of s3 files, and says it holds several', () => { + expect( + attachmentsTargetFor(input({ type: 'array', items: { resourceType: 's3object' } })) + ).toEqual({ name: 'files', multiple: true }) + }) + + it('takes a single s3 file', () => { + expect(attachmentsTargetFor(input({ format: 'resource-s3_object' }))).toEqual({ + name: 'files', + multiple: false + }) + }) + + // The transform can build the s3 object itself, promoting an input that holds a key + // rather than a file. Uploading into it would write an object where a string is declared. + it('offers no paperclip where the input cannot hold a file', () => { + expect(attachmentsTargetFor(input({ type: 'string' }))).toBeUndefined() + expect( + attachmentsTargetFor(input({ type: 'array', items: { type: 'string' } })) + ).toBeUndefined() + expect(attachmentsTargetFor(undefined)).toBeUndefined() + }) +}) diff --git a/frontend/src/lib/components/flows/conversations/agentChatInputs.ts b/frontend/src/lib/components/flows/conversations/agentChatInputs.ts index 91f6231097..9a7f81bdc6 100644 --- a/frontend/src/lib/components/flows/conversations/agentChatInputs.ts +++ b/frontend/src/lib/components/flows/conversations/agentChatInputs.ts @@ -284,10 +284,55 @@ export function agentModelGap(wiring: AgentModelWiring | undefined): string | un : undefined } -/** Every flow input the wiring reads, so the modal does not ask for them a second time. */ -export function agentModelWiringInputs(wiring: AgentModelWiring | undefined): string[] { +/** + * The flow inputs the model button actually writes, so the modal does not ask for them a + * second time — and, just as much, so it still asks for the ones the button cannot reach. + * + * Two fields are conditional. The button writes `kind` only alongside a resource, since a + * provider is picked as a pair: a flow that wires `kind` while fixing the resource leaves + * the button nothing to write it with. And it offers the thinking slider only where the + * model reasons, so on a model that does not, a wired `reasoning_effort` has no editor + * there either. Hiding either one would leave the run short of an input with nowhere to + * supply it — and a required one would pass the modal's own completeness check. + */ +export function agentModelWiringInputs( + wiring: AgentModelWiring | undefined, + /** Whether the model in use reasons at all. False whenever it cannot be determined. */ + effortEditable: boolean = false +): string[] { if (!wiring) return [] - return [...(wiring.whole ? [wiring.whole] : []), ...Object.values(wiring.fields)] + if (wiring.whole) return [wiring.whole] + const driven: ProviderField[] = ['resource', 'model'] + if (wiring.fields.resource !== undefined) driven.push('kind') + if (effortEditable) driven.push('reasoning_effort') + return driven.map((field) => wiring.fields[field]).filter((name): name is string => !!name) +} + +/** Whether a schema entry holds an s3 file, as the flow input editor recognises one. */ +function holdsS3File(property: Record | undefined): boolean { + return ( + property?.format === 'resource-s3_object' || + property?.resourceType === 's3object' || + property?.resourceType === 's3_object' + ) +} + +/** + * Where the composer's attachments go, or nothing when there is nowhere they fit. + * + * The agent reads `user_attachments` through a transform that may reshape what it takes, so + * the flow input feeding it is not necessarily an s3 field: an expression building the s3 + * object itself promotes a plain string. Writing `{ s3, filename }` into that input fails at + * run time, so the paperclip appears only where the schema says the value belongs. + */ +export function attachmentsTargetFor( + input: AgentChatInput | undefined +): { name: string; multiple: boolean } | undefined { + if (!input) return undefined + if (holdsS3File(input.property)) return { name: input.name, multiple: false } + return input.property?.type === 'array' && holdsS3File(input.property.items) + ? { name: input.name, multiple: true } + : undefined } export function isEmptyAgentChatInputValue(value: any): boolean { diff --git a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts b/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts index 81a939f72e..adf3988e26 100644 --- a/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/flowChatViewHost.svelte.ts @@ -32,8 +32,6 @@ export type AttachmentsTarget = { name: string; multiple: boolean } export type FlowChatViewHostOptions = { additionalInputs?: () => Record | undefined - /** Called once the turn is dispatched, to clear anything that rides one message. */ - onSent?: () => void attachmentsTarget?: () => AttachmentsTarget | undefined workspace?: () => string | undefined /** Off when the workspace has no object storage — there is nowhere to upload to. */ @@ -41,6 +39,8 @@ export type FlowChatViewHostOptions = { /** Flow inputs the composer renders a control of its own for, so a message does not * repeat them as context chips. */ inputsShownInComposer?: () => string[] + /** The flow's input schema, which says which of a run's arguments are secret. */ + inputsSchema?: () => { properties?: Record } | undefined } /** @@ -177,6 +177,7 @@ export class FlowChatViewHost implements ChatViewHost { #messageInputs = new MessageInputsStore( () => this.#options.workspace?.(), + () => this.#options.inputsSchema?.(), () => new Set(this.#options.inputsShownInComposer?.() ?? []) ) #toolCalls = new ToolCallStore(() => this.#options.workspace?.()) @@ -263,6 +264,14 @@ export class FlowChatViewHost implements ChatViewHost { // The composer refuses an attachment-only send (requiresMessageText), so this is // the same rule at the other end: nothing runs without a message. if (!text) return false + // And nothing runs into a conversation belonging to the other surface: the composer + // is shut for it, but a queued turn could have been written before it was opened. + const wrongKind = this.#manager.wrongKindReason + if (wrongKind) { + sendUserToast(wrongKind, true) + this.#restoreToComposer(options) + return false + } // The per-turn cap again, at the place the truncation would happen: the composer // enforces it as files are attached, but a queue built over several turns arrives // here as one send, and a scalar input keeps `uploaded[0]` — uploading the rest @@ -306,8 +315,7 @@ export class FlowChatViewHost implements ChatViewHost { } this.#manager.inputMessage = text - this.#options.onSent?.() - await this.#manager.sendMessage( + const started = await this.#manager.sendMessage( Object.keys(args).length > 0 || this.#options.additionalInputs?.() ? args : undefined, (rowId) => { if (!sentInputs) return @@ -321,6 +329,13 @@ export class FlowChatViewHost implements ChatViewHost { }, conversationId ) + if (!started) { + // The upload succeeded and the run did not, so the composer's draft was spent on + // nothing. The uploaded objects stay where they are — a resend uploads its own, + // under its own prefix — but what the reader wrote comes back. + this.#restoreToComposer(options) + return false + } return true } diff --git a/frontend/src/lib/components/flows/conversations/jobBackedStore.svelte.ts b/frontend/src/lib/components/flows/conversations/jobBackedStore.svelte.ts index 2bdbdafefa..1ecd103464 100644 --- a/frontend/src/lib/components/flows/conversations/jobBackedStore.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/jobBackedStore.svelte.ts @@ -10,12 +10,20 @@ * The rules live here; what to fetch and how to read it is the caller's. */ +/** + * Fetches allowed out at once. A page of conversation rows asks for all of its jobs in the + * same render, and the answers only fill chips in below text that is already on screen. + */ +const MAX_CONCURRENT = 6 + export class JobBackedStore { #workspace: () => string | undefined #load: (workspace: string, jobId: string) => Promise #empty: T #byJob = $state>({}) #inFlight = new Set() + #waiting: string[] = [] + #running = 0 constructor( workspace: () => string | undefined, @@ -32,14 +40,35 @@ export class JobBackedStore { if (!jobId) return this.#empty const cached = this.#byJob[jobId] if (cached) return cached - void this.#fetch(jobId) + this.#enqueue(jobId) return this.#empty } + #enqueue(jobId: string) { + if (this.#inFlight.has(jobId)) return + this.#inFlight.add(jobId) + this.#waiting.push(jobId) + this.#pump() + } + + #pump() { + while (this.#running < MAX_CONCURRENT && this.#waiting.length > 0) { + const jobId = this.#waiting.shift()! + this.#running++ + void this.#fetch(jobId).finally(() => { + this.#running-- + this.#pump() + }) + } + } + async #fetch(jobId: string) { const workspace = this.#workspace() - if (!workspace || this.#inFlight.has(jobId)) return - this.#inFlight.add(jobId) + if (!workspace) { + // Neither cached nor in flight, so the row asks again once a workspace is known. + this.#inFlight.delete(jobId) + return + } try { this.#byJob = { ...this.#byJob, [jobId]: await this.#load(workspace, jobId) } } catch { diff --git a/frontend/src/lib/components/flows/conversations/messageInputContext.svelte.ts b/frontend/src/lib/components/flows/conversations/messageInputContext.svelte.ts index 73fdd69e9e..496bcac352 100644 --- a/frontend/src/lib/components/flows/conversations/messageInputContext.svelte.ts +++ b/frontend/src/lib/components/flows/conversations/messageInputContext.svelte.ts @@ -12,6 +12,7 @@ import { JobService } from '$lib/gen' import { JobBackedStore } from './jobBackedStore.svelte' import { base } from '$lib/base' +import { redactFileArgs, redactSecretArgs } from '$lib/components/job_args' import { createAttachedFileContextElement, type ContextElement @@ -67,12 +68,17 @@ const EMPTY: MessageInputs = { images: [], contextElements: [] } export function argsToMessageInputs( workspace: string, args: Record | undefined, + schema: { properties?: Record } | undefined, shownElsewhere: ReadonlySet = new Set() ): MessageInputs { if (!args) return EMPTY const images: AttachedImage[] = [] const contextElements: ContextElement[] = [] - for (const [name, value] of Object.entries(args)) { + // A chip is text on screen, so it goes through the same redaction the run page and the + // copilot apply to a job's arguments: a password input must not be readable here, and a + // base64 file is unreadable anyway. + const shown = redactFileArgs(redactSecretArgs(args, schema), schema) + for (const [name, value] of Object.entries(shown)) { if (name === 'user_message') continue // An input the composer has its own control for — the model button's provider fields — // is already on screen, and repeating it under every message is noise. @@ -119,11 +125,16 @@ export function attachmentsToMessageInputs( /** The run arguments behind the transcript's user rows. One fetch per turn while mounted. */ export class MessageInputsStore extends JobBackedStore { - constructor(workspace: () => string | undefined, shownElsewhere: () => ReadonlySet) { + constructor( + workspace: () => string | undefined, + schema: () => { properties?: Record } | undefined, + shownElsewhere: () => ReadonlySet + ) { super(workspace, EMPTY, async (ws, jobId) => argsToMessageInputs( ws, (await JobService.getJobArgs({ workspace: ws, id: jobId })) as any, + schema(), shownElsewhere() ) ) diff --git a/frontend/src/lib/components/flows/conversations/messageInputContext.test.ts b/frontend/src/lib/components/flows/conversations/messageInputContext.test.ts new file mode 100644 index 0000000000..d5f4535a00 --- /dev/null +++ b/frontend/src/lib/components/flows/conversations/messageInputContext.test.ts @@ -0,0 +1,47 @@ +import { describe, expect, it } from 'vitest' +import { argsToMessageInputs } from './messageInputContext.svelte' + +const SCHEMA = { + properties: { + token: { type: 'string', password: true }, + city: { type: 'string' }, + report: { type: 'object', format: 'resource-s3_object' } + } +} + +function summaries(elements: { title?: string; content?: string }[]) { + return elements.map((e) => e.content) +} + +describe('argsToMessageInputs', () => { + it('never puts a secret input on screen', () => { + const { contextElements } = argsToMessageInputs( + 'ws', + { token: 'hunter2', city: 'Paris' }, + SCHEMA + ) + expect(summaries(contextElements as any)).toEqual(['', 'Paris']) + }) + + // Only images get a thumbnail lane; every other s3 file is a chip naming its key. + it('splits attachments into thumbnails and file chips', () => { + const { images, contextElements } = argsToMessageInputs( + 'ws', + { report: [{ s3: 'a/shot.png' }, { s3: 'a/notes.pdf' }] }, + SCHEMA + ) + expect(images.map((i) => i.name)).toEqual(['shot.png']) + expect(images[0].dataUrl).toContain('file_key=a%2Fshot.png') + expect(contextElements).toHaveLength(1) + }) + + it('leaves out the message and anything the composer already shows', () => { + const { contextElements } = argsToMessageInputs( + 'ws', + { user_message: 'hi', city: 'Paris' }, + SCHEMA, + new Set(['city']) + ) + expect(contextElements).toEqual([]) + }) +})