diff --git a/backend/windmill-api-settings/src/lib.rs b/backend/windmill-api-settings/src/lib.rs index 3d474ff4b1..bd34ca35ef 100644 --- a/backend/windmill-api-settings/src/lib.rs +++ b/backend/windmill-api-settings/src/lib.rs @@ -1506,6 +1506,10 @@ struct CustomInstanceDbLogs { db_connect: String, #[serde(skip_serializing_if = "String::is_empty")] grant_permissions: String, + #[serde(skip_serializing_if = "String::is_empty")] + replication_user: String, + #[serde(skip_serializing_if = "Option::is_none")] + replication_user_error: Option, } async fn list_custom_instance_pg_databases( @@ -1705,30 +1709,24 @@ async fn setup_custom_instance_pg_database_inner( )) })?; - // The replication attribute lives on a dedicated role used by postgres trigger - // connections. The getter creates the role (with its stored password) when the - // migration couldn't. - if let Err(e) = windmill_common::utils::get_custom_pg_instance_replication_password(db).await { - tracing::error!("Failed to ensure custom_instance_replication_user exists: {e:#}"); - } - if let Err(e) = client - .batch_execute( - "ALTER ROLE custom_instance_replication_user REPLICATION; - GRANT custom_instance_user TO custom_instance_replication_user; - ALTER ROLE custom_instance_user NOREPLICATION;", - ) - .await - { - tracing::error!( - "Failed to grant replication permission to custom_instance_replication_user: {e:#}" - ); - } - logs.grant_permissions = "OK".to_string(); drop(client); // /!\ Drop before joining to avoid deadlock windmill_common::shutdown_pg_connection(join_handle).await?; + // Roles are cluster-wide, so the dedicated role used by postgres trigger connections is + // provisioned on the main pool rather than on the new database. Reported as its own step + // rather than failing the setup: without the role the database still serves datatables, only + // postgres triggers on them break. + match windmill_common::utils::ensure_custom_instance_replication_user(db).await { + Ok(()) => logs.replication_user = "OK".to_string(), + Err(e) => { + tracing::error!("Failed to provision custom_instance_replication_user: {e:#}"); + logs.replication_user = "FAIL".to_string(); + logs.replication_user_error = Some(e.to_string()); + } + } + Ok(()) } diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index 10bc2fa97e..0f80695b66 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -29200,6 +29200,10 @@ components: $ref: "#/components/schemas/LoggedWizardStatus" grant_permissions: $ref: "#/components/schemas/LoggedWizardStatus" + replication_user: + $ref: "#/components/schemas/LoggedWizardStatus" + replication_user_error: + type: string CustomInstanceDbTag: type: string diff --git a/backend/windmill-api/src/live_migrations.rs b/backend/windmill-api/src/live_migrations.rs index 7c359300c3..4c6f5e245c 100644 --- a/backend/windmill-api/src/live_migrations.rs +++ b/backend/windmill-api/src/live_migrations.rs @@ -23,9 +23,39 @@ pub async fn custom_migrations(migrator: &mut CustomMigrator) -> Result<(), Erro tracing::error!("Could not normalize custom_instance_user attributes: {err:#}"); } + if let Err(err) = ensure_custom_instance_replication_user(migrator).await { + tracing::error!( + "Could not provision custom_instance_replication_user: {err:#}. Postgres triggers on \ + custom-instance datatables will not work until the role can open replication \ + connections: grant the role owning DATABASE_URL either SUPERUSER or (PG 16+) the \ + REPLICATION attribute. On AWS RDS this is handled by granting rds_replication, which \ + requires no change; other managed providers are not covered" + ); + } + Ok(()) } +// Converged on every boot, not once: the one-shot migration creates the role with the +// REPLICATION attribute, which managed postgres rejects outright, so instances set up before +// the provider-role fallback existed have no role at all. Scoped to instances that actually +// have a custom-instance database, so the rest never see the error. +async fn ensure_custom_instance_replication_user( + migrator: &mut CustomMigrator, +) -> Result<(), Error> { + let has_custom_instance_db = sqlx::query_scalar::<_, bool>( + "SELECT COALESCE(value->'databases', '{}'::jsonb) <> '{}'::jsonb + FROM global_settings WHERE name = 'custom_instance_pg_databases'", + ) + .fetch_optional(migrator.connection()) + .await? + .unwrap_or(false); + if !has_custom_instance_db { + return Ok(()); + } + windmill_common::utils::ensure_custom_instance_replication_user(migrator.connection()).await +} + // Converged on every boot, not once: the one-shot migration swallows errors (it must not // abort startup without superuser), and an older instance sharing the cluster can re-add // the attribute. REPLICATION belongs only on custom_instance_replication_user. diff --git a/backend/windmill-api/src/static_assets.rs b/backend/windmill-api/src/static_assets.rs index 35728bc693..06a79d022e 100644 --- a/backend/windmill-api/src/static_assets.rs +++ b/backend/windmill-api/src/static_assets.rs @@ -50,8 +50,15 @@ const TWO_HUNDRED: &str = "200.html"; /// Check if the original path requires cross-origin isolation headers. /// -/// These headers are needed for SharedArrayBuffer and TypeScript workers -/// (raw app editor at `/apps_raw/`, in-browser bundler at `/ui_builder/`). +/// CANONICAL COEP RATIONALE (the dev-server mirror in `frontend/vite.config.js` +/// and the navigation guards in `frontend/src/routes/(root)/(logged)/+layout.svelte` +/// point here): the headers are needed for SharedArrayBuffer and TypeScript +/// workers (raw app editor at `/apps_raw/edit|add`, in-browser bundler at +/// `/ui_builder/`). The raw app *viewer* (`/apps_raw/get/`) must NOT get them: +/// COEP `require-corp` blocks the viewed app's cross-origin subresources +/// (external images, embeds) that lack CORP — and since headers stick to the +/// document, apps would break on a page reload while working when reached via +/// client-side navigation. /// /// Public apps (`/public/` and custom paths `/a/`) opt in via the `wm_coep` /// query param: a public (raw) app must set COEP to be embeddable as an iframe @@ -62,7 +69,9 @@ const TWO_HUNDRED: &str = "200.html"; /// explicitly requests it. #[cfg(feature = "static_frontend")] fn needs_cross_origin_isolation(original_path: &str, query: Option<&str>) -> bool { - original_path.starts_with("/apps_raw/") + // no trailing slash on edit/add: matches the +layout.svelte guards + original_path.starts_with("/apps_raw/edit") + || original_path.starts_with("/apps_raw/add") || original_path.starts_with("/ui_builder/") || ((original_path.starts_with("/public/") || original_path.starts_with("/a/")) && query_has_flag(query, "wm_coep")) @@ -152,8 +161,15 @@ mod tests { fn test_needs_cross_origin_isolation() { // editor + bundler are always isolated, regardless of query assert!(needs_cross_origin_isolation("/apps_raw/edit/foo", None)); + assert!(needs_cross_origin_isolation("/apps_raw/add", None)); assert!(needs_cross_origin_isolation("/ui_builder/index.html", None)); + // the raw app viewer must NOT be isolated + assert!(!needs_cross_origin_isolation( + "/apps_raw/get/u/foo/bar", + None + )); + // public apps (and custom paths) are isolated only when they opt in via wm_coep assert!(needs_cross_origin_isolation( "/public/ws/secret", diff --git a/backend/windmill-common/src/instance_config.rs b/backend/windmill-common/src/instance_config.rs index 7e32fe6e56..abf1c1b662 100644 --- a/backend/windmill-common/src/instance_config.rs +++ b/backend/windmill-common/src/instance_config.rs @@ -819,6 +819,10 @@ pub struct CustomInstanceDbLogs { pub db_connect: String, #[serde(skip_serializing_if = "String::is_empty")] pub grant_permissions: String, + #[serde(skip_serializing_if = "String::is_empty")] + pub replication_user: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub replication_user_error: Option, } // --------------------------------------------------------------------------- diff --git a/backend/windmill-common/src/utils.rs b/backend/windmill-common/src/utils.rs index e2597fdfce..7b1f2a13f9 100644 --- a/backend/windmill-common/src/utils.rs +++ b/backend/windmill-common/src/utils.rs @@ -1058,30 +1058,90 @@ pub async fn get_custom_pg_instance_password(db: &DB) -> Result { ) } -const REFRESH_CUSTOM_INSTANCE_REPLICATION_USER_SQL: &str = r#" +/// PL/pgSQL granting `custom_instance_replication_user` the ability to open replication +/// connections, inlined into the `DO` blocks below. +/// +/// The REPLICATION attribute requires a real superuser on PG <= 15 (PG 16+ accepts +/// CREATEROLE + REPLICATION), and managed postgres never hands one out: on RDS the +/// capability is carried by the `rds_replication` role instead. Both privilege failures +/// raise `insufficient_privilege`, so the attribute is attempted first and the provider +/// role is the fallback. +const GRANT_REPLICATION_CAPABILITY_PLPGSQL: &str = r#" + BEGIN + EXECUTE 'ALTER ROLE custom_instance_replication_user REPLICATION'; + EXCEPTION WHEN insufficient_privilege THEN + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'rds_replication') THEN + EXECUTE 'GRANT rds_replication TO custom_instance_replication_user'; + ELSE + RAISE; + END IF; + END; + + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_user') THEN + EXECUTE 'GRANT custom_instance_user TO custom_instance_replication_user'; + -- Stripping REPLICATION off custom_instance_user is a cleanup, so it is both + -- guarded and best-effort: the clause is superuser-only on PG <= 15 even when the + -- attribute is already unset, and failing it must not roll back the role above. + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_user' AND rolreplication) THEN + BEGIN + EXECUTE 'ALTER ROLE custom_instance_user NOREPLICATION'; + EXCEPTION WHEN insufficient_privilege THEN + NULL; + END; + END IF; + END IF; +"#; + +/// `rotate_condition` decides when a new password is generated: always for the refresh +/// endpoint, only when the role or its stored password is missing for the boot converge. +/// Both variants are a single statement so they can run on any executor, and both are +/// atomic: a role that cannot be granted replication is rolled back rather than left +/// half-provisioned. +fn provision_replication_user_sql(rotate_condition: &str) -> String { + format!( + r#" DO $$ DECLARE pwd text; BEGIN - SELECT gen_random_uuid()::text INTO pwd; + -- Same lock as get_custom_pg_instance_replication_password: without it, every API + -- replica booting onto a version that provisions the role races into CREATE USER, + -- and the losers abort on duplicate_object. + PERFORM pg_advisory_xact_lock(hashtext('custom_instance_replication_pwd')); - IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_replication_user') THEN - EXECUTE format('ALTER USER custom_instance_replication_user WITH PASSWORD %L REPLICATION', pwd); - ELSE - EXECUTE format('CREATE USER custom_instance_replication_user WITH PASSWORD %L REPLICATION', pwd); + IF {rotate_condition} THEN + SELECT gen_random_uuid()::text INTO pwd; + + IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_replication_user') THEN + EXECUTE format('ALTER USER custom_instance_replication_user WITH PASSWORD %L', pwd); + ELSE + EXECUTE format('CREATE USER custom_instance_replication_user WITH PASSWORD %L', pwd); + END IF; + + INSERT INTO global_settings (name, value) + VALUES ('custom_instance_replication_pwd', to_jsonb(pwd::text)) + ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value; END IF; - - IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_user') THEN - GRANT custom_instance_user TO custom_instance_replication_user; - ALTER ROLE custom_instance_user NOREPLICATION; - END IF; - - INSERT INTO global_settings (name, value) - VALUES ('custom_instance_replication_pwd', to_jsonb(pwd::text)) - ON CONFLICT (name) DO UPDATE SET value = EXCLUDED.value; +{GRANT_REPLICATION_CAPABILITY_PLPGSQL} END $$; -"#; +"# + ) +} + +lazy_static::lazy_static! { + static ref REFRESH_CUSTOM_INSTANCE_REPLICATION_USER_SQL: String = + provision_replication_user_sql("TRUE"); + + // The password test mirrors REPLICATION_PWD_READ_SQL, whose readers flatten a NULL away: a row + // holding a JSON null reads back as no password, so it must rotate rather than count as + // provisioned and strand the getter. + static ref ENSURE_CUSTOM_INSTANCE_REPLICATION_USER_SQL: String = provision_replication_user_sql( + "NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'custom_instance_replication_user') + OR NOT EXISTS (SELECT 1 FROM global_settings + WHERE name = 'custom_instance_replication_pwd' AND value #>> '{}' IS NOT NULL)" + ); +} const REPLICATION_PWD_READ_SQL: &str = "SELECT value #>> '{}' FROM global_settings WHERE name = 'custom_instance_replication_pwd'"; @@ -1093,12 +1153,26 @@ const REPLICATION_PWD_READ_SQL: &str = /// Authorization: rotates a stored database credential and performs no authorization /// itself — callers MUST restrict this to superadmin or internal server paths. pub async fn refresh_custom_instance_replication_user_pwd(db: &DB) -> Result<()> { - sqlx::query(REFRESH_CUSTOM_INSTANCE_REPLICATION_USER_SQL) + sqlx::query(REFRESH_CUSTOM_INSTANCE_REPLICATION_USER_SQL.as_str()) .execute(db) .await?; Ok(()) } +/// Create `custom_instance_replication_user` if it is missing and (re)grant it replication, +/// leaving an existing password in place. Idempotent, so it can be converged on every boot. +/// +/// Authorization: same contract as [`refresh_custom_instance_replication_user_pwd`] — +/// callers MUST restrict this to superadmin or internal server paths. +pub async fn ensure_custom_instance_replication_user<'c>( + executor: impl sqlx::PgExecutor<'c>, +) -> Result<()> { + sqlx::query(ENSURE_CUSTOM_INSTANCE_REPLICATION_USER_SQL.as_str()) + .execute(executor) + .await?; + Ok(()) +} + /// Authorization: returns a stored database credential and performs no authorization /// itself — callers MUST restrict this to superadmin or internal server paths (mirrors /// [`get_custom_pg_instance_password`]). @@ -1127,7 +1201,7 @@ pub async fn get_custom_pg_instance_replication_password(db: &DB) -> Result>(REPLICATION_PWD_READ_SQL) diff --git a/backend/windmill-trigger-postgres/src/handler.rs b/backend/windmill-trigger-postgres/src/handler.rs index 0592c48bf7..92dd2e1e19 100644 --- a/backend/windmill-trigger-postgres/src/handler.rs +++ b/backend/windmill-trigger-postgres/src/handler.rs @@ -208,8 +208,7 @@ impl TriggerCrud for PostgresTrigger { publication, publication_data.map(|publication| publication.0), ) - .await - .map_err(to_anyhow)?; + .await?; } remote_db_tx.commit().await.map_err(to_anyhow)?; @@ -714,15 +713,13 @@ pub async fn update_pg_publication( table_to_track.as_deref(), &transaction_to_track, ) - .await - .map_err(to_anyhow)?; + .await?; } else { let pg_14 = check_if_valid_publication_for_postgres_version( pg_connection, table_to_track.as_deref(), ) - .await - .map_err(to_anyhow)?; + .await?; let mut query = format!("ALTER PUBLICATION {} SET ", quoted_publication_name); let mut first = true; @@ -832,8 +829,7 @@ pub async fn alter_publication( publication_data, publication.map(|publication| publication.0), ) - .await - .map_err(to_anyhow)?; + .await?; tx.commit().await.map_err(to_anyhow)?; diff --git a/backend/windmill-trigger-postgres/src/lib.rs b/backend/windmill-trigger-postgres/src/lib.rs index a295183dfc..571aac0a66 100644 --- a/backend/windmill-trigger-postgres/src/lib.rs +++ b/backend/windmill-trigger-postgres/src/lib.rs @@ -508,6 +508,10 @@ pub async fn create_pg_publication( query.push_str(")"); } + // A row filter has no bind parameter, so it goes in raw. Publication + // DDL must therefore keep running through `Client::execute`, whose + // `Parse` refuses more than one command: under `simple_query` or + // `batch_execute` a filter could stack statements. if let Some(where_clause) = &table.where_clause { query.push_str(" WHERE ("); query.push_str(where_clause); diff --git a/frontend/src/lib/components/copilot/chat/flow/FlowAIChat.svelte b/frontend/src/lib/components/copilot/chat/flow/FlowAIChat.svelte index 1c6376bc62..e8e3207e05 100644 --- a/frontend/src/lib/components/copilot/chat/flow/FlowAIChat.svelte +++ b/frontend/src/lib/components/copilot/chat/flow/FlowAIChat.svelte @@ -197,7 +197,15 @@ return { errorCount: 0, warningCount: 0, errors: [], warnings: [] } }, - setFlowJson: async ({ modules, schema, preprocessorModule, failureModule, groups, notes }) => { + setFlowJson: async ({ + modules, + schema, + preprocessorModule, + failureModule, + groups, + notes, + settings + }) => { try { if ( modules !== undefined || @@ -205,7 +213,8 @@ preprocessorModule !== undefined || failureModule !== undefined || groups !== undefined || - notes !== undefined + notes !== undefined || + settings !== undefined ) { // Take snapshot of current flowStore and set as beforeFlow if (!diffManager?.hasPendingChanges) { @@ -221,7 +230,8 @@ preprocessorModule, failureModule, groups, - notes + notes, + settings }) // Refresh the state store to update UI diff --git a/frontend/src/lib/components/copilot/chat/flow/core.ts b/frontend/src/lib/components/copilot/chat/flow/core.ts index 2e6fea35a7..98f35cd3a9 100644 --- a/frontend/src/lib/components/copilot/chat/flow/core.ts +++ b/frontend/src/lib/components/copilot/chat/flow/core.ts @@ -43,10 +43,13 @@ import { flowModuleSchema } from './openFlowZod.gen' import { collectAllFlowModuleIdsFromModules } from '$lib/components/flows/flowTree' import { buildEditableFlowJson as buildEditableFlowJsonBase, + FLOW_VALUE_SETTINGS_KEYS, + pickFlowValueSettings, validateEditableFlowJson, validateFlowModules, validateFlowSchema, - type EditableFlowJson + type EditableFlowJson, + type FlowValueSettings } from './editableFlowJson' import { FLOW_CHAT_SPECIAL_MODULES, getFlowPrompt } from '$system_prompts' @@ -57,6 +60,9 @@ type FlowJsonUpdate = { failureModule?: FlowModule | null groups?: FlowGroup[] | null notes?: FlowNote[] | null + /** Full state of the top-level FlowValue settings: when provided, keys + * absent from it are removed from the flow value. */ + settings?: FlowValueSettings } function formatEmptyInlineScriptWarning({ @@ -563,7 +569,8 @@ export const flowTools: Tool[] = [ preprocessorModule: parsedFlow.preprocessor_module, failureModule: parsedFlow.failure_module, groups: parsedFlow.groups, - notes: parsedFlow.notes + notes: parsedFlow.notes, + settings: pickFlowValueSettings(parsedFlow) }) const warning = formatEmptyInlineScriptWarning(updateResult) @@ -854,7 +861,7 @@ export function prepareFlowSystemMessage(customPrompt?: string): ChatCompletionS Use \`patch_flow_json\` for small, localized changes when you can target an exact snippet from the \`CURRENT FLOW JSON COMPACT\` block below. Always copy the exact search text from the \`CURRENT FLOW JSON COMPACT\` block below. -The compact JSON is a single object with \`modules\`, \`schema\`, \`preprocessor_module\`, \`failure_module\`, \`groups\`, and \`notes\` keys. +The compact JSON is a single object with \`modules\`, \`schema\`, \`preprocessor_module\`, \`failure_module\`, \`groups\`, and \`notes\` keys, plus any top-level flow settings that are set (${FLOW_VALUE_SETTINGS_KEYS.join(', ')}). Settings can be added, edited, or removed with \`patch_flow_json\` as top-level keys — e.g. \`chat_input_enabled: true\` marks the flow as chat-style (flow-as-chat); keep it intact when restructuring such a flow. \`set_flow_json\` never changes these settings. **Parameters:** - \`old_string\`: Exact JSON text to find diff --git a/frontend/src/lib/components/copilot/chat/flow/editableFlowJson.test.ts b/frontend/src/lib/components/copilot/chat/flow/editableFlowJson.test.ts new file mode 100644 index 0000000000..9a106f5131 --- /dev/null +++ b/frontend/src/lib/components/copilot/chat/flow/editableFlowJson.test.ts @@ -0,0 +1,102 @@ +import { describe, expect, it, vi } from 'vitest' +import type { FlowModule, FlowValue } from '$lib/gen' +import { + applyEditableFlowJsonToFlow, + buildEditableFlowJson, + validateEditableFlowJson +} from './editableFlowJson' +import { createInlineScriptSession } from './inlineScriptsUtils' + +vi.mock('../shared', () => ({ + SPECIAL_MODULE_IDS: { + PREPROCESSOR: 'preprocessor', + FAILURE: 'failure' + } +})) + +function makeRawScriptModule(id: string, content: string): FlowModule { + return { + id, + summary: id, + value: { + type: 'rawscript', + language: 'bun', + content, + input_transforms: {} + } + } as FlowModule +} + +function makeFlowValue(extra: Record = {}): FlowValue { + return { + modules: [makeRawScriptModule('step_a', 'code a')], + ...extra + } as FlowValue +} + +describe('flow settings in the compact editable view', () => { + it('round-trips top-level flow settings through build → validate → apply', () => { + const value = makeFlowValue({ chat_input_enabled: true, same_worker: true }) + const session = createInlineScriptSession() + const editable = buildEditableFlowJson({ value }, session) + + expect(editable.chat_input_enabled).toBe(true) + expect(editable.same_worker).toBe(true) + + const revalidated = validateEditableFlowJson(JSON.parse(JSON.stringify(editable))) + const result = applyEditableFlowJsonToFlow(value, revalidated, session) + + expect(result.chat_input_enabled).toBe(true) + expect(result.same_worker).toBe(true) + expect(result.modules[0]?.value).toMatchObject({ type: 'rawscript', content: 'code a' }) + }) + + it('applies a patched-in setting and deletes a removed one', () => { + const value = makeFlowValue({ cache_ttl: 60 }) + const session = createInlineScriptSession() + const editable = buildEditableFlowJson({ value }, session) + + const patched = JSON.parse(JSON.stringify(editable)) + patched.chat_input_enabled = true + delete patched.cache_ttl + + const result = applyEditableFlowJsonToFlow(value, validateEditableFlowJson(patched), session) + + expect(result.chat_input_enabled).toBe(true) + expect('cache_ttl' in result).toBe(false) + }) + + it('preserves original FlowValue fields outside the compact view', () => { + const value = makeFlowValue({ some_future_field: 'kept' }) + const session = createInlineScriptSession() + const editable = buildEditableFlowJson({ value }, session) + + expect('some_future_field' in editable).toBe(false) + + const result = applyEditableFlowJsonToFlow( + value, + validateEditableFlowJson(JSON.parse(JSON.stringify(editable))), + session + ) + + expect((result as Record).some_future_field).toBe('kept') + }) + + it('rejects unknown top-level keys instead of silently dropping them', () => { + expect(() => + validateEditableFlowJson({ + modules: [makeRawScriptModule('step_a', 'code a')], + chat_enabled: true + }) + ).toThrow(/Unknown top-level flow key\(s\): chat_enabled/) + }) + + it('rejects settings with the wrong type', () => { + expect(() => + validateEditableFlowJson({ + modules: [makeRawScriptModule('step_a', 'code a')], + chat_input_enabled: 'yes' + }) + ).toThrow(/chat_input_enabled/) + }) +}) diff --git a/frontend/src/lib/components/copilot/chat/flow/editableFlowJson.ts b/frontend/src/lib/components/copilot/chat/flow/editableFlowJson.ts index e920d4bdfa..1258f94a93 100644 --- a/frontend/src/lib/components/copilot/chat/flow/editableFlowJson.ts +++ b/frontend/src/lib/components/copilot/chat/flow/editableFlowJson.ts @@ -12,6 +12,18 @@ import { type FlowNote } from './helperUtils' import { flowModuleSchema, flowModulesSchema } from './openFlowZod.gen' +import { + FLOW_VALUE_SETTINGS_KEYS, + flowValueSettingsSchema, + pickFlowValueSettings, + type FlowValueSettings +} from './flowValueSettings' + +export { + FLOW_VALUE_SETTINGS_KEYS, + pickFlowValueSettings, + type FlowValueSettings +} from './flowValueSettings' /** * Compact, agent-friendly representation of a flow. @@ -27,7 +39,7 @@ export type EditableFlowJson = { failure_module: FlowModule | null groups: FlowGroup[] | null notes: FlowNote[] | null -} +} & FlowValueSettings /** Optional input to the rich-error path of `validateEditableFlowJson`. */ type SchemaErrorContext = { @@ -279,10 +291,19 @@ function validateOptionalFlowModule(rawModule: unknown, fieldName: string): Flow return result.data } +export const EDITABLE_FLOW_STRUCTURAL_KEYS = [ + 'modules', + 'schema', + 'preprocessor_module', + 'failure_module', + 'groups', + 'notes' +] as const + /** * Parse and validate a raw object as an `EditableFlowJson`. Validates module * shape, schema shape, optional special modules (with their reserved ids), - * groups, and that no module ids collide. + * groups, top-level flow settings, and that no module ids collide. */ export function validateEditableFlowJson( rawFlow: unknown, @@ -293,6 +314,28 @@ export function validateEditableFlowJson( } const flow = rawFlow as Record + + // Reject unknown top-level keys: silently dropping them would make patch + // tools report success for edits that never land on the flow. + const allowedKeys = new Set([ + ...EDITABLE_FLOW_STRUCTURAL_KEYS, + ...FLOW_VALUE_SETTINGS_KEYS + ]) + const unknownKeys = Object.keys(flow).filter((key) => !allowedKeys.has(key)) + if (unknownKeys.length > 0) { + throw new Error( + `Unknown top-level flow key(s): ${unknownKeys.join(', ')}. Allowed keys: ${[...allowedKeys].join(', ')}` + ) + } + + const settingsResult = flowValueSettingsSchema.safeParse(flow) + if (!settingsResult.success) { + const issue = settingsResult.error.issues[0] + const path = issue?.path?.join('.') ?? 'settings' + throw new Error(`Invalid flow setting ${path}: ${issue?.message ?? 'unknown error'}`) + } + const settings = pickFlowValueSettings(settingsResult.data) + const modules = validateFlowModules(flow.modules, ctx) const schema = validateFlowSchema(flow.schema) const preprocessorModule = validateOptionalFlowModule( @@ -345,7 +388,8 @@ export function validateEditableFlowJson( preprocessor_module: preprocessorModule, failure_module: failureModule, groups, - notes + notes, + ...settings } } @@ -401,7 +445,8 @@ export function buildEditableFlowJson( preprocessor_module: preprocessorModule ?? null, failure_module: failureModule ?? null, groups: flow.value.groups ?? null, - notes: flow.value.notes ?? null + notes: flow.value.notes ?? null, + ...pickFlowValueSettings(flow.value) } } @@ -420,8 +465,12 @@ export function restoreSpecialRawscriptModule( /** * Inverse of `buildEditableFlowJson`. Replaces `inline_script.` * placeholders in `editable.modules` and the special modules with the content - * stored in `session`. Other fields on the original FlowValue (`same_worker`, - * `concurrent_limit`, etc.) are preserved. + * stored in `session`. + * + * The compact view is the full state for the settings in + * `FLOW_VALUE_SETTINGS_KEYS`: a settings key absent from `editable` is removed + * from the result, so patches can unset them. Fields of the original FlowValue + * outside that list are preserved untouched. * * Pair with `buildEditableFlowJson` for round-trip patches: extract → patch * the compact view → restore. @@ -431,7 +480,7 @@ export function applyEditableFlowJsonToFlow( editable: EditableFlowJson, session: InlineScriptSession ): FlowValue { - return { + const result: FlowValue = { ...originalValue, modules: session.restoreInlineScriptReferences(editable.modules), preprocessor_module: @@ -440,6 +489,14 @@ export function applyEditableFlowJsonToFlow( groups: editable.groups ?? undefined, notes: editable.notes ?? undefined } + for (const key of FLOW_VALUE_SETTINGS_KEYS) { + if (editable[key] !== undefined) { + ;(result as Record)[key] = editable[key] + } else { + delete (result as Record)[key] + } + } + return result } /** diff --git a/frontend/src/lib/components/copilot/chat/flow/flowValueSettings.ts b/frontend/src/lib/components/copilot/chat/flow/flowValueSettings.ts new file mode 100644 index 0000000000..a660bf1ae3 --- /dev/null +++ b/frontend/src/lib/components/copilot/chat/flow/flowValueSettings.ts @@ -0,0 +1,52 @@ +import { z } from 'zod' +import type { FlowValue } from '$lib/gen' + +/** + * Non-structural top-level FlowValue settings exposed in the compact view so + * they round-trip through patch/write tools instead of being silently dropped. + * Must stay in sync with the top-level scalar/config fields of `FlowValue`. + */ +export const flowValueSettingsSchema = z + .object({ + same_worker: z.boolean(), + preserve_step_tags: z.boolean(), + concurrent_limit: z.number(), + concurrency_key: z.string(), + concurrency_time_window_s: z.number(), + debounce_delay_s: z.number(), + debounce_key: z.string(), + debounce_args_to_accumulate: z.array(z.string()), + max_total_debouncing_time: z.number(), + max_total_debounces_amount: z.number(), + skip_expr: z.string(), + cache_ttl: z.number(), + cache_ignore_s3_path: z.boolean(), + delete_after_secs: z.number(), + flow_env: z.record(z.string(), z.any()), + priority: z.number(), + early_return: z.string(), + chat_input_enabled: z.boolean() + }) + .partial() + +type FlowValueSettingsKey = keyof z.infer & keyof FlowValue + +export const FLOW_VALUE_SETTINGS_KEYS = Object.keys( + flowValueSettingsSchema.shape +) as FlowValueSettingsKey[] + +export type FlowValueSettings = Pick + +/** + * Extract the defined non-structural settings from a FlowValue (or an + * EditableFlowJson, which carries the same keys). + */ +export function pickFlowValueSettings(source: Record): FlowValueSettings { + const settings: Record = {} + for (const key of FLOW_VALUE_SETTINGS_KEYS) { + if (source[key] !== undefined) { + settings[key] = source[key] + } + } + return settings as FlowValueSettings +} diff --git a/frontend/src/lib/components/copilot/chat/flow/helperUtils.test.ts b/frontend/src/lib/components/copilot/chat/flow/helperUtils.test.ts index 0d46e36f6a..ae763155f7 100644 --- a/frontend/src/lib/components/copilot/chat/flow/helperUtils.test.ts +++ b/frontend/src/lib/components/copilot/chat/flow/helperUtils.test.ts @@ -53,6 +53,39 @@ function makeAiAgentWithTool(agentId: string, toolId: string, toolContent: strin } describe('applyFlowJsonUpdate', () => { + it('leaves top-level flow settings untouched when settings is not provided', () => { + const flow = { + value: { + modules: [makeRawScriptModule('step_a', 'code a')], + chat_input_enabled: true, + same_worker: true + } + } + + applyFlowJsonUpdate(flow as any, createInlineScriptSession(), { + modules: [makeRawScriptModule('step_b', 'code b')] + }) + + expect((flow.value as any).chat_input_enabled).toBe(true) + expect((flow.value as any).same_worker).toBe(true) + }) + + it('applies full-state settings: sets provided keys and removes absent ones', () => { + const flow = { + value: { + modules: [makeRawScriptModule('step_a', 'code a')], + chat_input_enabled: true + } + } + + applyFlowJsonUpdate(flow as any, createInlineScriptSession(), { + settings: { same_worker: true } + }) + + expect((flow.value as any).same_worker).toBe(true) + expect('chat_input_enabled' in flow.value).toBe(false) + }) + it('accepts new self-referenced inline scripts and initializes them as empty', () => { const flow = { value: { diff --git a/frontend/src/lib/components/copilot/chat/flow/helperUtils.ts b/frontend/src/lib/components/copilot/chat/flow/helperUtils.ts index ba5d98fdfb..c5a7f7be22 100644 --- a/frontend/src/lib/components/copilot/chat/flow/helperUtils.ts +++ b/frontend/src/lib/components/copilot/chat/flow/helperUtils.ts @@ -8,6 +8,7 @@ import { NoteColor } from '$lib/components/graph/noteColors' import type { InlineScriptSession } from './inlineScriptsUtils' +import { FLOW_VALUE_SETTINGS_KEYS, type FlowValueSettings } from './flowValueSettings' /** Allowed note/group color names — matches the NoteColor palette the note and * group editors use. The note renderer keys `NOTE_COLORS` by these exact names, @@ -56,6 +57,9 @@ export interface FlowJsonUpdate { failureModule?: FlowModule | null groups?: FlowGroup[] | null notes?: FlowNote[] | null + /** Full state of the top-level FlowValue settings: when provided, keys + * absent from it are removed from the flow value. */ + settings?: FlowValueSettings } export interface FlowJsonUpdateResult { @@ -269,7 +273,7 @@ export function validateFlowNotes(rawNotes: unknown, moduleIds?: Set): F export function applyFlowJsonUpdate( flow: FlowLike, inlineScriptSession: InlineScriptSession, - { modules, schema, preprocessorModule, failureModule, groups, notes }: FlowJsonUpdate + { modules, schema, preprocessorModule, failureModule, groups, notes, settings }: FlowJsonUpdate ): FlowJsonUpdateResult { const emptyInlineScriptModuleIds = new Set() @@ -307,6 +311,16 @@ export function applyFlowJsonUpdate( flow.value.notes = notes == null || notes.length === 0 ? undefined : notes } + if (settings !== undefined) { + for (const key of FLOW_VALUE_SETTINGS_KEYS) { + if (settings[key] !== undefined) { + ;(flow.value as Record)[key] = settings[key] + } else { + delete (flow.value as Record)[key] + } + } + } + return { emptyInlineScriptModuleIds: Array.from(emptyInlineScriptModuleIds) } diff --git a/frontend/src/lib/components/copilot/chat/global/core.test.ts b/frontend/src/lib/components/copilot/chat/global/core.test.ts index 19c05e00ef..2ce668d789 100644 --- a/frontend/src/lib/components/copilot/chat/global/core.test.ts +++ b/frontend/src/lib/components/copilot/chat/global/core.test.ts @@ -1941,7 +1941,7 @@ describe('global AI tools', () => { path: 'f/flows/existing', summary: 'deployed summary', description: 'deployed description', - value: { modules: [] }, + value: { modules: [], chat_input_enabled: true, same_worker: true }, schema: { properties: { deployed: { type: 'boolean' } } }, edited_by: 'admin', edited_at: '2026-05-22T09:00:00Z', @@ -1961,7 +1961,11 @@ describe('global AI tools', () => { path: 'f/flows/existing', summary: 'new summary', description: 'deployed description', - value: { modules: [{ id: 'step', value: { type: 'identity' } }] } + value: { + modules: [{ id: 'step', value: { type: 'identity' } }], + chat_input_enabled: true, + same_worker: true + } }) }) diff --git a/frontend/src/lib/components/copilot/chat/global/core.ts b/frontend/src/lib/components/copilot/chat/global/core.ts index e512c60d90..80204ba02b 100644 --- a/frontend/src/lib/components/copilot/chat/global/core.ts +++ b/frontend/src/lib/components/copilot/chat/global/core.ts @@ -58,7 +58,10 @@ import { isChromiumBrowser } from '$lib/utils' import { applyEditableFlowJsonToFlow, buildEditableFlowJson, + EDITABLE_FLOW_STRUCTURAL_KEYS, type EditableFlowJson, + FLOW_VALUE_SETTINGS_KEYS, + pickFlowValueSettings, finalizeUnresolvedInlineScripts, restoreSpecialRawscriptModule, validateEditableFlowJson @@ -521,6 +524,7 @@ function appendEmptyInlineScriptWarning(result: string, editable: EditableFlowJs function editableFlowToDraftValue(editable: EditableFlowJson): FlowDraftValue { const value: FlowValue = { + ...pickFlowValueSettings(editable), modules: editable.modules, preprocessor_module: editable.preprocessor_module ?? undefined, failure_module: editable.failure_module ?? undefined, @@ -1947,8 +1951,9 @@ function getFlowInstructions(): string { - Global mode writes complete draft payloads only; it does not save, deploy, run, scaffold local files, or generate metadata. - Paths follow the conventions in the system prompt: default to \`u//\` when the user gave a bare name; only use \`f//\` when the folder is known to exist. Never invent a folder. -- \`write_flow\` mirrors flow mode's \`set_flow_json\`: pass \`path\`, optional \`summary\`, optional \`description\`, required \`modules\`, and optional \`schema\`, \`preprocessor_module\`, \`failure_module\`, \`groups\`, and \`notes\`. \`summary\` and \`description\` are top-level flow metadata (not part of the compact value \`patch_flow_json\` edits); the flow-structure arguments are JSON strings, matching the tool schema descriptions. -- \`read_workspace_item\` returns a compact flow \`value\` object with \`modules\`, \`schema\`, \`preprocessor_module\`, \`failure_module\`, \`groups\`, and \`notes\`. +- \`write_flow\` mirrors flow mode's \`set_flow_json\`: pass \`path\`, optional \`summary\`, optional \`description\`, required \`modules\`, and optional \`schema\`, \`preprocessor_module\`, \`failure_module\`, \`groups\`, and \`notes\`. \`summary\` and \`description\` are top-level flow metadata (not part of the compact value \`patch_flow_json\` edits); the flow-structure arguments are JSON strings, matching the tool schema descriptions. When overwriting an existing flow, top-level flow settings (see below) are preserved from the current flow — use \`patch_flow_json\` to change them. +- \`read_workspace_item\` returns a compact flow \`value\` object with \`modules\`, \`schema\`, \`preprocessor_module\`, \`failure_module\`, \`groups\`, \`notes\`, and any top-level flow settings that are set. +- Top-level flow settings appear as top-level keys of the compact flow value and can be added, edited, or removed with \`patch_flow_json\`: ${FLOW_VALUE_SETTINGS_KEYS.join(', ')}. For example, \`chat_input_enabled: true\` marks a flow as chat-style (flow-as-chat); keep it intact when restructuring such a flow. - \`modules\` contains normal sequential modules. Use top-level \`preprocessor_module\` and \`failure_module\` for special modules; do not put \`preprocessor\` or \`failure\` in \`modules\`. - Every module needs a stable unique \`id\` and a useful \`summary\` when the schema supports it. - Prefer path/script/flow modules when composing existing workspace logic. Use rawscript modules only when new inline code is needed. @@ -2878,7 +2883,8 @@ export const globalTools: Tool<{}>[] = [ path: parsed.path, summary: parsed.summary, description: parsed.description, - flow: editableFlowToDraftValue(resolved) + flow: editableFlowToDraftValue(resolved), + preserveBaseValueSettings: true }, ctx ) @@ -4169,8 +4175,14 @@ type FlowDraftArgs = { description?: string flow: FlowDraftValue override?: boolean + /** Carry over the base value's non-structural fields (chat_input_enabled, + * same_worker, ...) into the new value. Set by write_flow, whose arguments + * cannot express them; patch_flow_json passes the full value state instead. */ + preserveBaseValueSettings?: boolean } +const FLOW_STRUCTURAL_VALUE_KEYS = new Set(EDITABLE_FLOW_STRUCTURAL_KEYS) + const FLOW_SPEC: WriteSpec = { probe: (workspace, path) => FlowService.existsFlowByPath({ workspace, path }), fetchDeployed: (workspace, path) => FlowService.getFlowByPath({ workspace, path }), @@ -4179,6 +4191,16 @@ const FLOW_SPEC: WriteSpec = { if (args.flow.groups !== undefined && args.flow.groups !== null) { value.groups = structuredClone(args.flow.groups) } + if (args.preserveBaseValueSettings && base?.value) { + for (const [key, fieldValue] of Object.entries(base.value)) { + if ( + !FLOW_STRUCTURAL_VALUE_KEYS.has(key) && + (value as Record)[key] === undefined + ) { + ;(value as Record)[key] = structuredClone(fieldValue) + } + } + } return base ? { ...structuredClone(base), diff --git a/frontend/src/lib/components/flows/content/FlowInput.svelte b/frontend/src/lib/components/flows/content/FlowInput.svelte index 03125168e0..21bbdaa1ec 100644 --- a/frontend/src/lib/components/flows/content/FlowInput.svelte +++ b/frontend/src/lib/components/flows/content/FlowInput.svelte @@ -44,7 +44,7 @@ import CaptureTable from '$lib/components/triggers/CaptureTable.svelte' import { isObjectTooBig, readFieldsRecursively } from '$lib/utils' import { refreshFlowStateStore } from '$lib/components/flows/flowStoreRefresh.svelte' - import type { AiAgent, ScriptLang } from '$lib/gen' + import type { AiAgent, InputTransform, ScriptLang } from '$lib/gen' import { deepEqual } from 'fast-equals' import Toggle from '$lib/components/Toggle.svelte' import { AI_AGENT_SCHEMA } from '../flowInfers' @@ -599,24 +599,45 @@ false ) } else if (aiAgentModules.length === 1) { - // Exactly one AI agent exists, configure it + // Exactly one AI agent exists: fill in defaults only for inputs the + // user hasn't configured, so re-enabling chat mode on an already + // configured agent doesn't clobber a custom user_message expression + // or a deliberate memory choice (e.g. off). const aiAgent = aiAgentModules[0] const value = aiAgent.value as AiAgent - // Set user_message to flow_input.user_message - value.input_transforms['user_message'] = { - type: 'javascript', - expr: 'flow_input.user_message' + // Degenerate shapes the input form can produce without deliberate + // configuration count as unconfigured: empty static value (undefined + // persists as null through JSON round-trips), blank JS expression + // (the JS toggle seeds a bare backtick pair), or an AI transform + // (meaningless for the chat input). + const isUnconfigured = (transform: InputTransform | undefined) => + transform === undefined || + (transform.type === 'static' && (transform.value == null || transform.value === '')) || + (transform.type === 'javascript' && transform.expr.replaceAll('`', '').trim() === '') || + transform.type === 'ai' + + const applied: string[] = [] + if (isUnconfigured(value.input_transforms['user_message'])) { + value.input_transforms['user_message'] = { + type: 'javascript', + expr: 'flow_input.user_message' + } + applied.push('user message input') } - // Set messages_context_length to 10 - value.input_transforms['memory'] = { - type: 'static', - value: { kind: 'auto', context_length: 10 } + if (isUnconfigured(value.input_transforms['memory'])) { + value.input_transforms['memory'] = { + type: 'static', + value: { kind: 'auto', context_length: 10 } + } + applied.push('context memory set to 10') } sendUserToast( - 'Chat mode enabled. AI agent configured with user message input and context memory set to 10.', + applied.length > 0 + ? `Chat mode enabled. AI agent configured with ${applied.join(' and ')}.` + : 'Chat mode enabled. Existing AI agent configuration kept unchanged.', false ) } diff --git a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte index c7ed838b6f..75c0d03d65 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleComponent.svelte @@ -1380,7 +1380,11 @@ maximum number of attempts as defined below. {/snippet} - + {:else if advancedSelected === 'runtime' && advancedRuntimeSelected === 'concurrency'}
@@ -1545,7 +1549,11 @@ {:else if advancedSelected === 'sleep'}
- +
{:else if advancedSelected === 'debounce'}
diff --git a/frontend/src/lib/components/flows/content/FlowModuleSleep.svelte b/frontend/src/lib/components/flows/content/FlowModuleSleep.svelte index dc45738338..92086ccd37 100644 --- a/frontend/src/lib/components/flows/content/FlowModuleSleep.svelte +++ b/frontend/src/lib/components/flows/content/FlowModuleSleep.svelte @@ -12,13 +12,16 @@ import Section from '$lib/components/Section.svelte' import Label from '$lib/components/Label.svelte' import { getStepPropPicker } from '../previousResults' + import { SAME_WORKER_INCOMPATIBLE_MSG } from '../utils.svelte' + import { Alert } from '$lib/components/common' interface Props { flowModule: FlowModule previousModuleId: string | undefined + isAgentTool?: boolean } - let { flowModule = $bindable(), previousModuleId }: Props = $props() + let { flowModule = $bindable(), previousModuleId, isAgentTool = false }: Props = $props() const { selectionManager, flowStore, flowStateStore, previewArgs } = getContext('FlowEditorContext') @@ -44,6 +47,8 @@ const result = flowStateStore.val[selectionManager.getSelectedId()]?.previewResult ?? {} let isSleepEnabled = $derived(Boolean(flowModule.sleep)) + // Agent tools never go through the flow scheduler, so `same_worker` doesn't apply to them. + let sameWorker = $derived(Boolean(!isAgentTool && flowStore.val.value.same_worker))
@@ -54,8 +59,15 @@ {/snippet} + {#if sameWorker} + + {SAME_WORKER_INCOMPATIBLE_MSG} Disable `Same Worker` in the flow settings to use a sleep. + + {/if} + { if (isSleepEnabled && flowModule.sleep != undefined) { @@ -72,7 +84,7 @@ }} />