feat: run a deployed flow through the chat's argument form (#11085)

* feat: run a deployed flow through the chat's argument form

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015vHs7Jr4UDSbUe2KGjugMw

* refactor: drop the unread dynselect helper from the deployed flow run form

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015vHs7Jr4UDSbUe2KGjugMw

* fix: skip the preprocessor when the chat runs a deployed flow

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015vHs7Jr4UDSbUe2KGjugMw

* test: pin run_flow steering with ai_evals cases

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015vHs7Jr4UDSbUe2KGjugMw

* refactor: inline the deployed flow schema and trim the eval draft check

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015vHs7Jr4UDSbUe2KGjugMw

* docs: correct the stale draft-validation comment on the flow test-run eval

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_015vHs7Jr4UDSbUe2KGjugMw

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

ai_evals/cases/global.yaml

@@ -2046,6 +2046,72 @@
     - passes the account "acme"
     - does not invent or guess the token's value
 
+- id: global-test35-run-deployed-flow-with-form
+  prompt: |-
+    Run the deployed flow `f/evals/global/notify_customer` for me — the customer is `acme`.
+  initial: ai_evals/fixtures/frontend/global/initial/notify_customer_flow.json
+  runtime:
+    maxTurns: 8
+    # A session chat is where the run card has a preview pane beside it; run_flow
+    # itself is offered in every chat.
+    sessionChat: true
+  validate:
+    draftCountExactly: 0
+  toolExpect:
+    requiredToolsUsed:
+      - run_flow
+      # A draft may declare different arguments than the deployed version being run, so
+      # the names to prefill have to come from the deployed schema.
+      - read_workspace_item
+    forbiddenToolsUsed:
+      - test_run_flow
+      - call_api_endpoint
+      - write_flow
+      - deploy_workspace_item
+    # An empty form pushes the work back onto the user, so the prefill is part of
+    # what the tool is for.
+    toolCallArgs:
+      - tool: run_flow
+        field: args.customer
+        stringIncludesAnyOf:
+          - acme
+  # Running produces no draft, and the judge cannot observe runs; validate via tool use.
+  skipJudge: true
+  judgeChecklist:
+    - runs the deployed flow through run_flow rather than a preview test run or a raw API endpoint
+    - passes the customer "acme" so the confirmation form comes up prefilled
+
+- id: global-test36-draft-flow-test-run-not-deployed
+  prompt: |-
+    Update the `calculate_total` step of `f/evals/global/process_invoice` so it applies 8% tax and
+    returns `subtotal`, `tax` and `total`, then run it to check it works.
+    Keep it as an AI draft only; do not deploy or save it.
+  initial: ai_evals/fixtures/frontend/global/initial/process_invoice_flow.json
+  runtime:
+    maxTurns: 10
+  validate:
+    draftCountExactly: 1
+    requiredDrafts:
+      - type: flow
+        path: f/evals/global/process_invoice
+  toolExpect:
+    # A one-step flow is as well checked by running the step as the whole flow, so both
+    # count: what matters is that the run is against the draft.
+    requiredToolsAnyOf:
+      - [test_run_flow, test_run_step]
+    # The draft is what the user asked to check, and run_flow would run the deployed
+    # version instead — the edit would not be in what ran.
+    forbiddenToolsUsed:
+      - run_flow
+      - call_api_endpoint
+      - deploy_workspace_item
+  # The judge cannot observe runs, and the edit's content is already pinned by
+  # global-test5 on this fixture; what this case guards is where the run went.
+  skipJudge: true
+  judgeChecklist:
+    - creates an AI draft of f/evals/global/process_invoice applying 8% tax
+    - does not deploy or save the draft
+
 - id: global-undo-created-draft
   prompt: |-
     Create a draft Postgres resource at `u/admin/scratch_db` for host db.example.com port 5432, database `orders`, user `app`, and tell me what fields it ended up with.

ai_evals/fixtures/frontend/global/initial/notify_customer_flow.json

@@ -0,0 +1,40 @@
+{
+  "workspace": {
+    "flows": [
+      {
+        "path": "f/evals/global/notify_customer",
+        "summary": "Notify a customer",
+        "description": "Sends a notification to the named customer.",
+        "schema": {
+          "$schema": "https://json-schema.org/draft/2020-12/schema",
+          "type": "object",
+          "properties": {
+            "customer": {
+              "type": "string"
+            }
+          },
+          "required": ["customer"]
+        },
+        "value": {
+          "modules": [
+            {
+              "id": "notify",
+              "summary": "Send the notification",
+              "value": {
+                "type": "rawscript",
+                "language": "bun",
+                "content": "export async function main(customer: string) {\n  return `Notified ${customer}`\n}\n",
+                "input_transforms": {
+                  "customer": {
+                    "type": "javascript",
+                    "expr": "flow_input.customer"
+                  }
+                }
+              }
+            }
+          ]
+        }
+      }
+    ]
+  }
+}

frontend/src/lib/components/copilot/chat/global/apiCatalogTools.test.ts

@@ -101,6 +101,19 @@ const CATALOG = [
 		},
 		body_schema: { type: 'object', additionalProperties: true }
 	},
+	{
+		name: 'cancelQueuedJob',
+		description: 'Cancel a queued job',
+		instructions: '',
+		path: '/w/{workspace}/jobs_u/queue/cancel/{id}',
+		method: 'POST',
+		path_params_schema: {
+			type: 'object',
+			properties: { workspace: { type: 'string' }, id: { type: 'string' } },
+			required: ['workspace', 'id']
+		},
+		body_schema: { type: 'object', additionalProperties: true }
+	},
 	{
 		name: 'runFlowByPath',
 		description: 'Run flow by path',
@@ -179,7 +192,7 @@ describe('call_api_get', () => {
 		const unknown = await run('call_api_get', { name: 'nope' })
 		expect(unknown.error).toContain('search_api_endpoints')
 
-		const mutating = await run('call_api_get', { name: 'runFlowByPath' })
+		const mutating = await run('call_api_get', { name: 'cancelQueuedJob' })
 		expect(mutating.error).toContain('call_api_endpoint')
 
 		const deleting = await run('call_api_endpoint', { name: 'deleteSchedule' })
@@ -191,17 +204,22 @@ describe('call_api_get', () => {
 		expect(search.matches.map((m: any) => m.name)).not.toContain('deleteScriptByHash')
 	})
 
-	// Left reachable, this endpoint is the way around the argument form: it runs the
-	// deployed script on the model's arguments, unstripped and unshown.
-	it('refuses a deployed script run, pointing at run_script', async () => {
-		const called = await run('call_api_endpoint', { name: 'runScriptByPath' })
-		expect(called.error).toContain('run_script')
-		expect(called.success).toBe(false)
+	// Left reachable, these endpoints are the way around the argument form: they run the
+	// deployed runnable on the model's arguments, unstripped and unshown.
+	it('refuses a deployed run, pointing at run_script and run_flow', async () => {
+		for (const [name, tool, query] of [
+			['runScriptByPath', 'run_script', 'run deployed script'],
+			['runFlowByPath', 'run_flow', 'run deployed flow']
+		]) {
+			const called = await run('call_api_endpoint', { name })
+			expect(called.error).toContain(tool)
+			expect(called.success).toBe(false)
 
-		// And it is gone from search, so the model is redirected before it ever calls.
-		const search = await run('search_api_endpoints', { query: 'run deployed script' })
-		expect(search.matches.map((m: any) => m.name)).not.toContain('runScriptByPath')
-		expect(search.covered_by_dedicated_tools?.join(' ')).toContain('run_script')
+			// And it is gone from search, so the model is redirected before it ever calls.
+			const search = await run('search_api_endpoints', { query })
+			expect(search.matches.map((m: any) => m.name)).not.toContain(name)
+			expect(search.covered_by_dedicated_tools?.join(' ')).toContain(tool)
+		}
 	})
 
 	it('refuses draft-blind item reads and lists, pointing at the draft-aware tools', async () => {
@@ -259,11 +277,11 @@ describe('call_api_endpoint', () => {
 		})
 		vi.stubGlobal('fetch', fetchMock)
 		const result = await run('call_api_endpoint', {
-			name: 'runFlowByPath',
-			params: { path: 'u/me/myflow' },
+			name: 'cancelQueuedJob',
+			params: { id: 'job/1' },
 			body: { args: { n: 1 } }
 		})
-		expect(fetchMock).toHaveBeenCalledWith('/api/w/test-ws/jobs/run/f/u%2Fme%2Fmyflow', {
+		expect(fetchMock).toHaveBeenCalledWith('/api/w/test-ws/jobs_u/queue/cancel/job%2F1', {
 			method: 'POST',
 			headers: { 'Content-Type': 'application/json' },
 			body: JSON.stringify({ args: { n: 1 } })

frontend/src/lib/components/copilot/chat/global/apiCatalogTools.ts

@@ -37,6 +37,7 @@ const COVERED_ENDPOINTS: Record<string, string> = {
 	listResource: 'list_workspace_items (it includes your drafts)',
 	listSchedules: 'list_workspace_items (it includes your drafts)',
 	runScriptByPath: 'run_script (it shows the user an argument form to confirm)',
+	runFlowByPath: 'run_flow (it shows the user an argument form to confirm)',
 	deleteScriptByPath: 'delete_workspace_item',
 	deleteScriptByHash: 'delete_workspace_item',
 	deleteFlowByPath: 'delete_workspace_item',

frontend/src/lib/components/copilot/chat/global/core.test.ts

@@ -5081,6 +5081,68 @@ describe('global AI tools', () => {
 		})
 	})
 
+	// What separates a deployed run from a test run of the same flow: the form is built from the
+	// deployed schema rather than the draft's, and the editor open on that path is left alone —
+	// it holds the draft, so a deployed run painted into its graph would show steps that are not
+	// the ones running.
+	it('run_flow forms on the deployed flow and leaves the live editor alone', async () => {
+		seedBackendDraft(
+			'flow',
+			'u/admin/deployed_and_drafted',
+			{
+				path: 'u/admin/deployed_and_drafted',
+				summary: 'Draft of the deployed flow',
+				value: { modules: [{ id: 'draft_step', value: { type: 'identity' } }] },
+				schema: { type: 'object', properties: { draft_only: { type: 'string' } } },
+				edited_by: '',
+				edited_at: '',
+				archived: false,
+				extra_perms: {}
+			},
+			{ workspace: WORKSPACE }
+		)
+		UserDraft.setLiveEditorDraft({
+			workspace: WORKSPACE,
+			itemKind: 'flow',
+			storagePath: 'u/admin/deployed_and_drafted',
+			effectivePath: 'u/admin/deployed_and_drafted'
+		})
+		vi.mocked(FlowService.getFlowByPath).mockResolvedValueOnce({
+			path: 'u/admin/deployed_and_drafted',
+			summary: 'Deployed flow',
+			value: { modules: [{ id: 'deployed_step', value: { type: 'identity' } }] },
+			schema: FLOW_NAME_SCHEMA
+		} as any)
+		const testActiveFlow = vi.fn(async () => 'job-live-flow')
+
+		let form: any
+		await withCompletedTestJob(() =>
+			callGlobalTool(
+				'run_flow',
+				{ path: 'u/admin/deployed_and_drafted', args: { name: 'Ada' } },
+				{
+					...toolCallbacks,
+					requestRunArgs: async (_toolId, f) => {
+						form = f
+						return { name: 'Grace' }
+					}
+				},
+				{ testActiveFlow }
+			)
+		)
+
+		expect(form.runnableKind).toBe('flow')
+		expect(form.schema?.properties).toEqual(FLOW_NAME_SCHEMA.properties)
+		expect(testActiveFlow).not.toHaveBeenCalled()
+		expect(JobService.runFlowPreview).not.toHaveBeenCalled()
+		expect(JobService.runFlowByPath).toHaveBeenCalledWith({
+			workspace: WORKSPACE,
+			path: 'u/admin/deployed_and_drafted',
+			requestBody: { name: 'Grace' },
+			skipPreprocessor: true
+		})
+	})
+
 	it('test_run_step previews rawscript steps from the draft flow', async () => {
 		const content = 'export async function main(name: string) {\n\treturn name.toUpperCase()\n}'
 		await callGlobalTool('write_flow', {

frontend/src/lib/components/copilot/chat/global/core.ts

@@ -935,6 +935,20 @@ const testRunFlowToolDef = createToolDef(
 	{ strict: false }
 )
 
+const runFlowSchema = z.object({
+	path: z.string().describe('Workspace path of the deployed flow to run.'),
+	args: testRunArgsSchema,
+	background: backgroundArgSchema,
+	wait_seconds: waitSecondsArgSchema
+})
+
+const runFlowToolDef = createToolDef(
+	runFlowSchema,
+	'run_flow',
+	'Run a DEPLOYED flow for real, under the user\'s own permissions. Fill in every argument you can infer: the user gets an argument form prefilled with `args` and decides what runs. For a secret argument prefer `$var:<path>` naming an existing workspace variable; a literal is minted into a short-lived secret before the run, but stays in this call. A required file is the user\'s to attach, so call this even when you cannot supply one rather than asking in chat. Use only when the user names the deployed version ("the deployed X", "in production", "for real"); otherwise use test_run_flow.',
+	{ strict: false }
+)
+
 const testRunStepSchema = z.object({
 	path: z.string().describe('Workspace path of the flow containing the step to test.'),
 	stepId: z.string().describe('The id of the step/module to test.'),
@@ -1350,7 +1364,7 @@ ${pipelineBullet}
 			: ' Pass items ("<kind>:<path>" entries naming the items you changed) so the review is scoped to them — omitting items preselects every pending change in the workspace'
 	}, or mode ("draft" or "fork") to force which comparison is shown. Prefer offering this review page over calling deploy_workspace_item directly when several items changed.
 - For a Windmill operation no other tool covers (workers, queue state, a run's args, ...), use search_api_endpoints to find a REST endpoint, then call_api_get for reads or call_api_endpoint for mutations (the user is asked to confirm those). Always prefer a dedicated tool when one exists; endpoints for authoring or deleting scripts, flows, apps, schedules, resources, or variables are not available through the API catalog tools — use the draft tools and delete_workspace_item instead.
-- Default to test_run_script, test_run_flow, or test_run_step for any run request, an existing script included; they prefer drafts and need no deployment. Use run_script only when the user names the deployed version ("the deployed X", "in production", "for real") — a bare "run X" is not that. For run_script, read the item with read_workspace_item version: "deployed" first so the arguments match the deployed schema, and fill in every one you can infer. test_run_script, run_script and test_run_flow all show the user an argument form prefilled with what you sent, so fill in every argument you can infer rather than asking for it in chat. runFlowByPath from the API catalog is the exception — it runs a deployed flow with no form at all: only for a flow the user asked to run deployed.
+- Default to test_run_script, test_run_flow, or test_run_step for any run request, an existing script included; they prefer drafts and need no deployment. Use run_script or run_flow only when the user names the deployed version ("the deployed X", "in production", "for real") — a bare "run X" is not that. For those two, read the item with read_workspace_item version: "deployed" first so the arguments match the deployed schema. test_run_script, test_run_flow, run_script and run_flow all show the user an argument form prefilled with what you sent, so fill in every argument you can infer rather than asking for it in chat.
 - When a required decision is ambiguous, use askUserQuestion with two to ten clear proposed answer strings instead of guessing. The user can also type a custom answer when none of the proposed answers fit. Set multiSelect: true only when the answers can genuinely co-apply and the user may pick several (not mutually exclusive).
 - When the user asks you to remember a lasting preference, always/never do something, or change/stop a behavior going forward, call update_user_instructions to persist it. It edits only the USER INSTRUCTIONS block (not WORKSPACE INSTRUCTIONS). Keep each instruction concise; do not use it for one-off requests scoped to the current task.
 - Keep context targeted.${
@@ -3704,6 +3718,19 @@ export const globalTools: Tool<{}>[] = [
 		showDetails: true,
 		autoCollapseDetails: false
 	},
+	{
+		def: runFlowToolDef,
+		fn: async (ctx) => {
+			const parsed = runFlowSchema.parse(ctx.args)
+			return runDeployedFlow(parsed, ctx)
+		},
+		bypassedByAutoAccept: true,
+		streamingLabel: 'Preparing the run form...',
+		confirmationMessage: 'Run a deployed flow',
+		queuedLabel: (args) => `Run ${args?.path ?? 'a flow'}`,
+		showDetails: true,
+		autoCollapseDetails: false
+	},
 	{
 		def: testRunStepToolDef,
 		fn: async (ctx) => {
@@ -5791,6 +5818,44 @@ async function runDeployedScript(
 	)
 }
 
+async function runDeployedFlow(
+	args: z.infer<typeof runFlowSchema>,
+	ctx: WriteDraftCtx
+): Promise<string> {
+	const { workspace } = ctx
+	// No live editor is driven here as a test run drives one: that editor holds the draft, and a
+	// deployed run painted into its graph would show steps that are not the ones running.
+	const flow = await FlowService.getFlowByPath({ workspace, path: args.path })
+	return runThroughForm(
+		{
+			path: args.path,
+			schema: (flow.schema as Record<string, any>) ?? {},
+			summary: flow.summary,
+			kind: 'run',
+			// No code/lang: the dynamic-option pickers come from the deployed flow, not an inline copy.
+			schemaNoun: 'deployed',
+			toolName: 'run_flow',
+			proposed: args.args,
+			startMessage: `Running "${args.path}"...`,
+			contextName: 'flow',
+			autoAcceptable: true,
+			background: args.background,
+			detachAfterMs: waitSecondsToDetachMs(args.wait_seconds),
+			startJob: (submitted) =>
+				JobService.runFlowByPath({
+					workspace,
+					path: args.path,
+					requestBody: submitted,
+					// As the flow's own run page does: the form fills the main input schema, and a
+					// preprocessor would take these arguments for a webhook body and hand the flow
+					// its own output instead.
+					skipPreprocessor: true
+				})
+		},
+		ctx
+	)
+}
+
 async function testRunFlowByPath(
 	args: z.infer<typeof testRunFlowSchema>,
 	ctx: WriteDraftCtx