diff --git a/backend/Cargo.lock b/backend/Cargo.lock index 46b17fc08a..0b19814a2a 100644 --- a/backend/Cargo.lock +++ b/backend/Cargo.lock @@ -14147,7 +14147,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "ba3beec919fbdf99d719de8eda6adae3281f8a5b71ae40431f44dc7423053d34" dependencies = [ "loki-api", - "reqwest 0.12.24", + "reqwest 0.11.27", "serde", "serde_json", "snap", diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs index 212045607d..b2426db3cc 100644 --- a/backend/windmill-queue/src/jobs.rs +++ b/backend/windmill-queue/src/jobs.rs @@ -756,6 +756,7 @@ pub async fn add_completed_job_error( false, false, ) + .warn_after_seconds(10) .await?; Ok(result) } @@ -812,6 +813,7 @@ pub async fn add_completed_job( has_stream, from_cache, ) + .warn_after_seconds(10) }) .retry( ConstantBuilder::default() @@ -872,7 +874,7 @@ async fn commit_completed_job( ) -> windmill_common::error::Result<(Option, i64, bool)> { // let start = std::time::Instant::now(); - let mut tx = db.begin().await?; + let mut tx = db.begin().warn_after_seconds(10).await?; let job_id = queued_job.id; // tracing::error!("1 {:?}", start.elapsed()); @@ -927,6 +929,7 @@ async fn commit_completed_job( /* $10 */ result_columns as Option<&Vec>, ) .fetch_optional(&mut *tx) + .warn_after_seconds(10) .await .map_err(|e| Error::internal_err(format!("Could not add completed job {job_id}: {e:#}")))?; @@ -938,6 +941,7 @@ async fn commit_completed_job( job_id ) .fetch_one(&mut *tx) + .warn_after_seconds(10) .await .map_err(|e| Error::internal_err(format!("Could not add completed job {job_id}: {e:#}")))? .unwrap_or(false); @@ -963,6 +967,7 @@ async fn commit_completed_job( labels as Vec ) .execute(&mut *tx) + .warn_after_seconds(10) .await .map_err(|e| Error::InternalErr(format!("Could not update job labels: {e:#}")))?; } @@ -986,6 +991,7 @@ async fn commit_completed_job( parent_job ) .execute(&mut *tx) + .warn_after_seconds(10) .await .inspect_err(|e| { tracing::error!( @@ -998,7 +1004,7 @@ async fn commit_completed_job( // tracing::error!("Added completed job {:#?}", queued_job); let mut _skip_downstream_error_handlers = false; - tx = delete_job(tx, &job_id).await?; + tx = delete_job(tx, &job_id).warn_after_seconds(10).await?; // tracing::error!("3 {:?}", start.elapsed()); if queued_job.is_flow_step() { @@ -1019,13 +1025,14 @@ async fn commit_completed_job( &queued_job.workspace_id ) .execute(&mut *tx) + .warn_after_seconds(10) .await?; if flow_is_done { let r = sqlx::query_scalar!( "UPDATE parallel_monitor_lock SET last_ping = now() WHERE parent_flow_id = $1 and job_id = $2 RETURNING 1", parent_job, &queued_job.id - ).fetch_optional(&mut *tx).await?; + ).fetch_optional(&mut *tx).warn_after_seconds(10).await?; if r.is_some() { tracing::info!( "parallel flow iteration is done, setting parallel monitor last ping lock for job {}", @@ -1039,8 +1046,9 @@ async fn commit_completed_job( let schedule_path = queued_job.schedule_path().unwrap(); let script_path = queued_job.runnable_path.as_ref().unwrap(); - let schedule = - get_schedule_opt(&mut *tx, &queued_job.workspace_id, &schedule_path).await?; + let schedule = get_schedule_opt(&mut *tx, &queued_job.workspace_id, &schedule_path) + .warn_after_seconds(10) + .await?; if let Some(schedule) = schedule { #[cfg(feature = "enterprise")] @@ -1072,6 +1080,7 @@ async fn commit_completed_job( &queued_job.workspace_id ) .fetch_optional(&mut *tx) + .warn_after_seconds(10) .await? .flatten() .unwrap_or(false); @@ -1084,6 +1093,7 @@ async fn commit_completed_job( &script_path, &queued_job.workspace_id, )) + .warn_after_seconds(10) .await { match err { @@ -1106,6 +1116,7 @@ async fn commit_completed_job( queued_job.started_at.unwrap_or(chrono::Utc::now()), queued_job.priority, ) + .warn_after_seconds(10) .await { if !success { @@ -1122,6 +1133,7 @@ async fn commit_completed_job( err ), ) + .warn_after_seconds(10) .await; } } else { @@ -1148,6 +1160,7 @@ async fn commit_completed_job( queued_job.id.hyphenated().to_string(), ) .execute(&mut *tx) + .warn_after_seconds(10) .await .map_err(|e| { Error::internal_err(format!( @@ -1162,6 +1175,7 @@ async fn commit_completed_job( queued_job.id, ) .execute(&mut *tx) + .warn_after_seconds(10) .await { tracing::error!( @@ -1174,15 +1188,17 @@ async fn commit_completed_job( sqlx::query!("DELETE FROM job_perms WHERE job_id = $1", job_id) .execute(&mut *tx) + .warn_after_seconds(10) .await?; if !success || has_stream { sqlx::query!("DELETE FROM job_result_stream_v2 WHERE job_id = $1", job_id) .execute(&mut *tx) + .warn_after_seconds(10) .await?; } - tx.commit().await?; + tx.commit().warn_after_seconds(10).await?; tracing::info!( %job_id, diff --git a/backend/windmill-worker/src/result_processor.rs b/backend/windmill-worker/src/result_processor.rs index e5a7a0052c..524ae0f9ed 100644 --- a/backend/windmill-worker/src/result_processor.rs +++ b/backend/windmill-worker/src/result_processor.rs @@ -161,6 +161,7 @@ async fn process_jc( bench, ) .instrument(span) + .warn_after_seconds(10) .await; if let Some(root_job) = root_job { @@ -291,6 +292,7 @@ pub fn start_background_processor( #[cfg(feature = "benchmark")] &mut bench, ) + .warn_after_seconds(10) .await; if is_init_script_and_failure { @@ -544,6 +546,7 @@ pub async fn handle_receive_completed_job( #[cfg(feature = "benchmark")] bench, ) + .warn_after_seconds(10) .await; match processed_completed_job { @@ -807,6 +810,7 @@ pub async fn handle_job_error( err_json.clone(), worker_name, ) + .warn_after_seconds(10) .await }; @@ -850,7 +854,9 @@ pub async fn handle_job_error( if let Err(err) = updated_flow { if let Some(parent_job_id) = job.parent_job { if let Ok(Some(parent_job)) = - get_mini_completed_job(&parent_job_id, &job.workspace_id, db).await + get_mini_completed_job(&parent_job_id, &job.workspace_id, db) + .warn_after_seconds(10) + .await { let e = json!({"message": err.to_string(), "name": "InternalErr"}); append_logs( @@ -870,6 +876,7 @@ pub async fn handle_job_error( false, None, ) + .warn_after_seconds(10) .await; } } diff --git a/frontend/scripts/untar_ui_builder.js b/frontend/scripts/untar_ui_builder.js index 06704bc4f4..ae6c5a0cb2 100644 --- a/frontend/scripts/untar_ui_builder.js +++ b/frontend/scripts/untar_ui_builder.js @@ -20,7 +20,7 @@ console.log('Running postinstall for root project'); import { x } from 'tar' -const tarUrl = 'https://pub-06154ed168a24e73a86ab84db6bf15d8.r2.dev/ui_builder-b4fcf00.tar.gz' +const tarUrl = 'https://pub-06154ed168a24e73a86ab84db6bf15d8.r2.dev/ui_builder-6a45d08.tar.gz' const outputTarPath = path.join(process.cwd(), 'ui_builder.tar.gz') const extractTo = path.join(process.cwd(), 'static/ui_builder/') diff --git a/frontend/serve_ui_builder.sh b/frontend/serve_ui_builder.sh index 59eb6c3bd3..cce4dbde56 100755 --- a/frontend/serve_ui_builder.sh +++ b/frontend/serve_ui_builder.sh @@ -1,8 +1,8 @@ -node ./scripts/untar_ui_builder.js +# node ./scripts/untar_ui_builder.js -mkdir ui_builder_serve || true -cp -r static/ui_builder ui_builder_serve/ui_builder || true -rm -rf static/ui_builder || true +# mkdir ui_builder_serve || true +# cp -r static/ui_builder ui_builder_serve/ui_builder || true +# rm -rf static/ui_builder || true python3 -c " import os os.chdir('ui_builder_serve') diff --git a/frontend/src/lib/components/ResourcePicker.svelte b/frontend/src/lib/components/ResourcePicker.svelte index 94517edd29..c4548f872c 100644 --- a/frontend/src/lib/components/ResourcePicker.svelte +++ b/frontend/src/lib/components/ResourcePicker.svelte @@ -274,7 +274,7 @@ {/snippet} {#if value && hovering} -
+
diff --git a/frontend/src/lib/components/copilot/chat/AIButtonStyle.ts b/frontend/src/lib/components/copilot/chat/AIButtonStyle.ts new file mode 100644 index 0000000000..f7d6ed6fa7 --- /dev/null +++ b/frontend/src/lib/components/copilot/chat/AIButtonStyle.ts @@ -0,0 +1,13 @@ +export function AIBtnClasses(state: 'default' | 'selected' | 'green' = 'default') { + return twMerge( + ['selected', 'default'].includes(state) ? 'text-ai !border-ai/20 hover:bg-ai/15' : '', + { + default: '', + selected: 'bg-ai/10', + green: + 'bg-green-50 hover:bg-green-50 dark:bg-green-400/15 dark:hover:bg-green-400/15 text-green-800 border-green-200 dark:border-green-300/60 dark:text-green-400' + }[state] + ) +} + +import { twMerge } from 'tailwind-merge' diff --git a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts index 18fabfbc51..8eae180b5f 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts +++ b/frontend/src/lib/components/copilot/chat/AIChatManager.svelte.ts @@ -599,10 +599,15 @@ class AIChatManager { throw new Error('No flow helpers found') } - let snapshot: ExtendedOpenFlow | undefined = undefined + let snapshot: + | { type: 'flow'; value: ExtendedOpenFlow } + | { type: 'app'; value: number } + | undefined = undefined if (this.mode === AIMode.FLOW) { - snapshot = this.flowAiChatHelpers!.getFlowAndSelectedId().flow - this.flowAiChatHelpers!.setSnapshot(snapshot) + snapshot = { type: 'flow', value: this.flowAiChatHelpers!.getFlowAndSelectedId().flow } + this.flowAiChatHelpers!.setSnapshot(snapshot.value) + } else if (this.mode === AIMode.APP) { + snapshot = { type: 'app', value: this.appAiChatHelpers!.snapshot() } } this.displayMessages = [ diff --git a/frontend/src/lib/components/copilot/chat/AIChatMessage.svelte b/frontend/src/lib/components/copilot/chat/AIChatMessage.svelte index 805a7f17d6..2ce4728399 100644 --- a/frontend/src/lib/components/copilot/chat/AIChatMessage.svelte +++ b/frontend/src/lib/components/copilot/chat/AIChatMessage.svelte @@ -89,13 +89,17 @@ {/if} {#if message.role === 'user' && message.snapshot}
- Saved a flow snapshot + Saved {message.snapshot.type === 'flow' ? 'a flow' : 'an app'} snapshot + {/each} +
+ {/each} + + + + {/each} +
+{/if} diff --git a/frontend/src/lib/components/raw_apps/RawAppHistoryManager.svelte.ts b/frontend/src/lib/components/raw_apps/RawAppHistoryManager.svelte.ts new file mode 100644 index 0000000000..b81978ea9b --- /dev/null +++ b/frontend/src/lib/components/raw_apps/RawAppHistoryManager.svelte.ts @@ -0,0 +1,405 @@ +import type { Runnable } from './utils' +import { deepEqual } from 'fast-equals' + +/** + * Snapshot entry containing raw app state at a point in time + */ +export interface HistoryEntry { + id: number + timestamp: Date + files: Record + runnables: Record + summary: string +} + +/** + * A branch in the history tree + * Contains entries that diverged from a fork point + */ +export interface HistoryBranch { + id: number + forkPointId: number // ID of the entry this branch forked from + entries: HistoryEntry[] +} + +/** + * Configuration for history manager + */ +export interface HistoryConfig { + maxEntries: number + autoSnapshotInterval?: number // milliseconds +} + +/** + * History manager for raw apps with branching support + * + * Main timeline: The current working branch + * Branches: Preserved "futures" when navigating to historical points and making changes + * + * When selecting a historical entry and making changes: + * - Current "future" entries become a branch (forked from selected point) + * - The selected entry becomes the new "head" of main timeline + * + * When selecting an entry on a branch and making changes: + * - That branch becomes the main timeline + * - The old main timeline (from fork point onwards) becomes a branch + */ +export class RawAppHistoryManager { + // Main timeline entries + private entries = $state([]) + // Preserved branches (old "futures" that were branched off) + private branches = $state([]) + private autoSnapshotTimer: number | undefined = undefined + private getStateFn: + | (() => { + files: Record + runnables: Record + summary: string + }) + | undefined = undefined + private isCreatingSnapshot = $state(false) + // Currently selected entry index in main timeline (-1 = at latest/no selection) + private currentIndex = $state(-1) + // If viewing a branch, which branch and entry index + private currentBranchId = $state(undefined) + private currentBranchEntryIndex = $state(-1) + private entryIdCounter = $state(0) + private branchIdCounter = $state(0) + // Track if current state has pending changes + private hasPendingChanges = $state(false) + + // Derived state + public readonly hasEntries = $derived(this.entries.length > 0) + public readonly entryCount = $derived(this.entries.length) + public readonly allEntries = $derived(this.entries.slice()) + public readonly allBranches = $derived(this.branches.slice()) + public readonly canSnapshot = $derived(!this.isCreatingSnapshot) + + // The ID of the currently selected entry (main timeline or branch) + public readonly selectedEntryId = $derived.by(() => { + if (this.currentBranchId !== undefined) { + const branch = this.branches.find((b) => b.id === this.currentBranchId) + return branch?.entries[this.currentBranchEntryIndex]?.id + } + if (this.currentIndex === -1) return undefined + return this.entries[this.currentIndex]?.id + }) + + // Whether we need to save current state before navigating + public readonly needsSnapshotBeforeNav = $derived( + this.currentIndex === -1 && this.currentBranchId === undefined && this.hasPendingChanges + ) + + public readonly canUndo = $derived( + this.currentIndex > 0 || + (this.currentIndex === -1 && this.entries.length > 1) || + (this.currentIndex === -1 && this.entries.length === 1 && this.hasPendingChanges) + ) + + public readonly canRedo = $derived( + this.currentIndex !== -1 && this.currentIndex < this.entries.length - 1 + ) + + constructor(private config: HistoryConfig) {} + + /** + * Create a snapshot from provided state + */ + createSnapshot( + files: Record, + runnables: Record, + summary: string + ): HistoryEntry { + return { + id: this.entryIdCounter++, + timestamp: new Date(), + files: structuredClone($state.snapshot(files)), + runnables: structuredClone($state.snapshot(runnables)), + summary: $state.snapshot(summary) + } + } + + /** + * Check if state has changed since last snapshot + */ + private hasStateChanged( + files: Record, + runnables: Record, + summary: string + ): boolean { + if (this.entries.length === 0) return true + + const lastEntry = this.entries[this.entries.length - 1] + return ( + !deepEqual(lastEntry.files, files) || + !deepEqual(lastEntry.runnables, runnables) || + lastEntry.summary !== summary + ) + } + + /** + * Add a snapshot to the main timeline + */ + addSnapshot(entry: HistoryEntry): void { + if (this.isCreatingSnapshot) { + console.warn('Snapshot already in progress, skipping') + return + } + + this.isCreatingSnapshot = true + + try { + this.entries = [...this.entries, entry] + + // FIFO: Remove oldest entries when exceeding limit + if (this.entries.length > this.config.maxEntries) { + const removed = this.entries.slice(0, this.entries.length - this.config.maxEntries) + this.entries = this.entries.slice(-this.config.maxEntries) + // Clean up branches that reference removed entries + const removedIds = new Set(removed.map((e) => e.id)) + this.branches = this.branches.filter((b) => !removedIds.has(b.forkPointId)) + } + + this.hasPendingChanges = false + } finally { + this.isCreatingSnapshot = false + } + } + + /** + * Mark that there are pending changes + * When making changes from a historical position, create a branch from the "future" + */ + markPendingChanges(): void { + // If we're on a branch and making changes, that branch becomes main + if (this.currentBranchId !== undefined) { + this.promoteBranchToMain() + } + // If we're at a historical position on main timeline + else if (this.currentIndex !== -1 && this.currentIndex < this.entries.length - 1) { + this.createBranchFromFuture() + } + + this.hasPendingChanges = true + } + + /** + * Create a branch from the "future" entries when making changes from historical position + */ + private createBranchFromFuture(): void { + const forkEntry = this.entries[this.currentIndex] + const futureEntries = this.entries.slice(this.currentIndex + 1) + + if (futureEntries.length > 0) { + const newBranch: HistoryBranch = { + id: this.branchIdCounter++, + forkPointId: forkEntry.id, + entries: futureEntries + } + this.branches = [...this.branches, newBranch] + } + + // Truncate main timeline to current position + this.entries = this.entries.slice(0, this.currentIndex + 1) + this.currentIndex = -1 + } + + /** + * Promote current branch to main timeline + * The old main timeline (from fork point onwards) becomes a branch + */ + private promoteBranchToMain(): void { + const branch = this.branches.find((b) => b.id === this.currentBranchId) + if (!branch) return + + // Find fork point in main timeline + const forkIndex = this.entries.findIndex((e) => e.id === branch.forkPointId) + if (forkIndex === -1) return + + // Save the current main timeline's "future" as a new branch (if any entries after fork) + const mainFutureEntries = this.entries.slice(forkIndex + 1) + if (mainFutureEntries.length > 0) { + const oldMainBranch: HistoryBranch = { + id: this.branchIdCounter++, + forkPointId: branch.forkPointId, + entries: mainFutureEntries + } + this.branches = [...this.branches.filter((b) => b.id !== this.currentBranchId), oldMainBranch] + } else { + // Just remove the current branch from branches list + this.branches = this.branches.filter((b) => b.id !== this.currentBranchId) + } + + // New main timeline: entries up to fork point + branch entries up to selected index + const branchEntriesUpToSelection = branch.entries.slice(0, this.currentBranchEntryIndex + 1) + this.entries = [...this.entries.slice(0, forkIndex + 1), ...branchEntriesUpToSelection] + + // Reset selection state + this.currentBranchId = undefined + this.currentBranchEntryIndex = -1 + this.currentIndex = -1 + } + + getId(): number { + return this.entryIdCounter + } + + /** + * Manually create and add a snapshot + * @param force - If true, create snapshot even if state hasn't changed + */ + manualSnapshot( + files: Record, + runnables: Record, + summary: string, + force = false + ): HistoryEntry | undefined { + if (!force && !this.hasStateChanged(files, runnables, summary)) { + return + } + + const entry = this.createSnapshot(files, runnables, summary) + this.addSnapshot(entry) + return entry + } + + /** + * Start automatic snapshot timer + */ + startAutoSnapshot( + getState: () => { + files: Record + runnables: Record + summary: string + } + ): void { + this.stopAutoSnapshot() + this.getStateFn = getState + + if (!this.config.autoSnapshotInterval) return + + this.autoSnapshotTimer = setInterval(() => { + if (this.getStateFn && this.currentIndex === -1 && this.currentBranchId === undefined) { + const { files, runnables, summary } = this.getStateFn() + this.manualSnapshot(files, runnables, summary) + } + }, this.config.autoSnapshotInterval) as unknown as number + } + + /** + * Stop automatic snapshot timer + */ + stopAutoSnapshot(): void { + if (this.autoSnapshotTimer !== undefined) { + clearInterval(this.autoSnapshotTimer) + this.autoSnapshotTimer = undefined + } + } + + /** + * Select an entry (on main timeline or a branch) + * If there are pending changes, a snapshot should be created first by the caller + */ + selectEntry(id: number): HistoryEntry | undefined { + // Check main timeline first + const mainIndex = this.entries.findIndex((e) => e.id === id) + if (mainIndex !== -1) { + this.currentIndex = mainIndex + this.currentBranchId = undefined + this.currentBranchEntryIndex = -1 + this.hasPendingChanges = false + return this.entries[mainIndex] + } + + // Check branches + for (const branch of this.branches) { + const branchIndex = branch.entries.findIndex((e) => e.id === id) + if (branchIndex !== -1) { + this.currentBranchId = branch.id + this.currentBranchEntryIndex = branchIndex + this.currentIndex = -1 + this.hasPendingChanges = false + return branch.entries[branchIndex] + } + } + + return undefined + } + + /** + * Clear selection (go back to latest state) + */ + clearSelection(): void { + this.currentIndex = -1 + this.currentBranchId = undefined + this.currentBranchEntryIndex = -1 + } + + /** + * Get entry by ID (searches main timeline and branches) + */ + getEntryById(id: number): HistoryEntry | undefined { + const mainEntry = this.entries.find((e) => e.id === id) + if (mainEntry) return mainEntry + + for (const branch of this.branches) { + const branchEntry = branch.entries.find((e) => e.id === id) + if (branchEntry) return branchEntry + } + + return undefined + } + + /** + * Get branch that contains an entry + */ + getBranchForEntry(id: number): HistoryBranch | undefined { + return this.branches.find((b) => b.entries.some((e) => e.id === id)) + } + + /** + * Undo to previous state + */ + undo(): HistoryEntry | null { + if (!this.canUndo) return null + + if (this.currentIndex === -1) { + this.currentIndex = this.entries.length - 2 + } else { + this.currentIndex-- + } + + this.hasPendingChanges = false + return this.entries[this.currentIndex] + } + + /** + * Redo to next state + */ + redo(): HistoryEntry | null { + if (!this.canRedo) return null + + this.currentIndex++ + this.hasPendingChanges = false + return this.entries[this.currentIndex] + } + + /** + * Clear all history + */ + clearHistory(): void { + this.entries = [] + this.branches = [] + this.currentIndex = -1 + this.currentBranchId = undefined + this.currentBranchEntryIndex = -1 + } + + /** + * Cleanup resources + */ + destroy(): void { + this.stopAutoSnapshot() + this.clearHistory() + } +} diff --git a/frontend/src/lib/components/raw_apps/RawAppInlineScriptPanelList.svelte b/frontend/src/lib/components/raw_apps/RawAppInlineScriptPanelList.svelte index 481cf940f3..64230a812e 100644 --- a/frontend/src/lib/components/raw_apps/RawAppInlineScriptPanelList.svelte +++ b/frontend/src/lib/components/raw_apps/RawAppInlineScriptPanelList.svelte @@ -45,7 +45,7 @@ } - + {#snippet action()}
- -
+
+ {/snippet} + + +{/if}