fix(backend): rework dependency_map handling (#6598)

* v0 Signed-off-by: pyranota <pyra@duck.com> * optimize relocks * make it work with relative relative imports Signed-off-by: pyranota <pyra@duck.com> * use fallback Signed-off-by: pyranota <pyra@duck.com> * remove dbg and todos Signed-off-by: pyranota <pyra@duck.com> * future proof a bit Signed-off-by: pyranota <pyra@duck.com> * cleanup Signed-off-by: pyranota <pyra@duck.com> * more cleanup Signed-off-by: pyranota <pyra@duck.com> * remove final TODO Signed-off-by: pyranota <pyra@duck.com> * do not use bytemuck Signed-off-by: pyranota <pyra@duck.com> * optimize hashing Signed-off-by: pyranota <pyra@duck.com> * implementation 1 Signed-off-by: pyranota <pyra@duck.com> * almost v0 Signed-off-by: pyranota <pyra@duck.com> * v0 Signed-off-by: pyranota <pyra@duck.com> * add comments and use fallback Signed-off-by: pyranota <pyra@duck.com> * call dissolve for apps Signed-off-by: pyranota <pyra@duck.com> * add comms Signed-off-by: pyranota <pyra@duck.com> * refactor v0 (partially tested + dirty) Signed-off-by: pyranota <pyra@duck.com> * finishing Signed-off-by: pyranota <pyra@duck.com> * remove TODO Signed-off-by: pyranota <pyra@duck.com> * Update SQLx metadata * silence unused argument Signed-off-by: pyranota <pyra@duck.com> * cleanup Signed-off-by: pyranota <pyra@duck.com> * implement rebuild_map endpoint Signed-off-by: pyranota <pyra@duck.com> * update windmill api client Signed-off-by: pyranota <pyra@duck.com> * almost finish with tests Signed-off-by: pyranota <pyra@duck.com> * add proper testing Signed-off-by: pyranota <pyra@duck.com> * remove unused fixtures Signed-off-by: pyranota <pyra@duck.com> * Update SQLx metadata * partial cleanup Signed-off-by: pyranota <pyra@duck.com> * Update backend/windmill-worker/src/scoped_dependency_map.rs Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com> * Update backend/windmill-common/src/scripts.rs Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com> * evil doings Signed-off-by: pyranota <pyra@duck.com> * more cleanup * Update SQLx metadata * more cleanup Signed-off-by: pyranota <pyra@duck.com> * fixing CI Signed-off-by: pyranota <pyra@duck.com> * remove python from default features Signed-off-by: pyranota <pyra@duck.com> --------- Signed-off-by: pyranota <pyra@duck.com> Co-authored-by: Pyra <92104930+pyranye@users.noreply.github.com> Co-authored-by: GitHub Action <action@github.com> Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com> Co-authored-by: ellipsis-dev[bot] <65095814+ellipsis-dev[bot]@users.noreply.github.com>
2026-08-18 16:02:10 +00:00 · 2025-09-24 15:40:39 +02:00
parent ef75ed3df7
commit ed806bf9d0
29 changed files with 1973 additions and 284 deletions
@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT path, versions[array_upper(versions, 1)] as version FROM flow WHERE workspace_id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "version",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      null
+    ]
+  },
+  "hash": "094587579285fc5656f6104716dca1805e3a1530fc64bf95eaee5b645e885251"
+}
@@ -0,0 +1,46 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT workspace_id, importer_path, importer_kind::text, imported_path, importer_node_id\n        FROM dependency_map WHERE workspace_id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "workspace_id",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "importer_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 2,
+        "name": "importer_kind",
+        "type_info": "Text"
+      },
+      {
+        "ordinal": 3,
+        "name": "imported_path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 4,
+        "name": "importer_node_id",
+        "type_info": "Varchar"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false,
+      null,
+      false,
+      false
+    ]
+  },
+  "hash": "13297889361ac6839d6c4bd0b8ae121305d63cfcd88f69700125d17fb2c56a1f"
+}
@@ -0,0 +1,22 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT value FROM app_version WHERE id = $1 LIMIT 1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "value",
+        "type_info": "Json"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Int8"
+      ]
+    },
+    "nullable": [
+      false
+    ]
+  },
+  "hash": "1492b88c75722465b1a5c138729e6bb2782e1f8ef5f9fe752b356927a8605100"
+}
@@ -0,0 +1,12 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "DELETE FROM dependency_map WHERE workspace_id = 'test-workspace'",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": []
+  },
+  "hash": "3c5b6001aac7fb58ec9bfad1bfd7418c16f96f307bafa926a18316055b7c90c4"
+}
@@ -1,16 +0,0 @@
-{
-  "db_name": "PostgreSQL",
-  "query": "DELETE FROM dependency_map\n                 WHERE importer_path = $1 AND importer_kind = $3::text::IMPORTER_KIND\n                 AND workspace_id = $2",
-  "describe": {
-    "columns": [],
-    "parameters": {
-      "Left": [
-        "Text",
-        "Text",
-        "Text"
-      ]
-    },
-    "nullable": []
-  },
-  "hash": "6c962f9471b0b1fe385a93789ec46bee53a07c8d1264eeb44bc94233bc06bbfd"
-}
@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT path, hash FROM script WHERE workspace_id = $1 AND archived = false",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "hash",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      false
+    ]
+  },
+  "hash": "79624ae15f909bd6ab4f015e32345bd1c24f93db32970d9c06623b1e335138ba"
+}
@@ -1,18 +1,18 @@
 {
  "db_name": "PostgreSQL",
-  "query": "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id)\n                 VALUES ($1, $2, $4::text::IMPORTER_KIND, $3, $5) ON CONFLICT DO NOTHING",
+  "query": "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id)\n                     VALUES ($1, $2, $3::text::IMPORTER_KIND, $4, $5) ON CONFLICT DO NOTHING",
  "describe": {
    "columns": [],
    "parameters": {
      "Left": [
-        "Varchar",
        "Varchar",
        "Varchar",
        "Text",
+        "Varchar",
        "Varchar"
      ]
    },
    "nullable": []
  },
-  "hash": "958ed17dafffdd37e636ccd244dc4ca60cbf562e6f6a371d5f9a9943fb30254c"
+  "hash": "e32d6c6ae4e0d824c4cf19128182d67f36d1fd87fe4cf4f002e18367088c497c"
 }
@@ -0,0 +1,18 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        DELETE FROM dependency_map\n        WHERE workspace_id = $1\n            AND importer_path = $2\n            AND importer_kind = $3::text::IMPORTER_KIND\n            AND importer_node_id = $4\n            AND imported_path = $5\n            ",
+  "describe": {
+    "columns": [],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Text",
+        "Text",
+        "Text",
+        "Text"
+      ]
+    },
+    "nullable": []
+  },
+  "hash": "ead84a63cb965e36155605434c9e3670e50344f39b0d61eef8e60c2cdb57ae1f"
+}
@@ -0,0 +1,20 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT value::text FROM app_version WHERE id = 0 AND app_id = 2",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "value",
+        "type_info": "Text"
+      }
+    ],
+    "parameters": {
+      "Left": []
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "f03d52c091d10d27a274cebaf370a4665a518da0560fd9ed7add05c66da3898f"
+}
@@ -0,0 +1,28 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "SELECT path, versions[array_upper(versions, 1)] as version FROM app WHERE workspace_id = $1",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "path",
+        "type_info": "Varchar"
+      },
+      {
+        "ordinal": 1,
+        "name": "version",
+        "type_info": "Int8"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text"
+      ]
+    },
+    "nullable": [
+      false,
+      null
+    ]
+  },
+  "hash": "fd403acc343182fdab100263f8ef8067c1b43a97ea96845d92e2cbf85fbd6311"
+}
@@ -1166,6 +1166,7 @@ checksum = "edca88bc138befd0323b20752846e6587272d3b03b0343c8ea28a6f819e6e71f"
 dependencies = [
 "async-trait",
 "axum-core",
+ "axum-macros",
 "bytes",
 "futures-util",
 "http 1.3.1",
@@ -1214,6 +1215,17 @@ dependencies = [
 "tracing",
 ]

+[[package]]
+name = "axum-macros"
+version = "0.4.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "57d123550fa8d071b7255cb0cc04dc302baa6c8c4a79f55701552684d8399bce"
+dependencies = [
+ "proc-macro2",
+ "quote",
+ "syn 2.0.106",
+]
+
 [[package]]
 name = "az"
 version = "1.2.1"
@@ -207,7 +207,7 @@ reqwest-middleware = { version = "^0", features = ["json"] }

 bitflags = "2.9.4"
 memchr = "2.7.4"
-axum = { version = "^0.7", features = ["multipart"] }
+axum = { version = "^0.7", features = ["multipart", "macros"] }
 headers = "^0"
 hyper = { version = "^1", features = ["full"] }
 tokio = { version = "=1.46.1", features = ["full", "tracing", "time"] }
@@ -22,7 +22,6 @@ def main():
 '',
 'f/system_relative/different_folder_script', 12347, 'python3', '');

-
 INSERT INTO public.script(workspace_id, created_by, content, schema, summary, description, path, hash, language, lock) VALUES (
 'test-workspace',
 'test-user',
@@ -38,4 +37,4 @@ def main():
 '{"$schema":"https://json-schema.org/draft/2020-12/schema","properties":{},"required":[],"type":"object"}',
 '',
 '',
-'f/system_relative/nested_script', 12348, 'python3', '');
+'f/system_relative/nested_script', 12348, 'python3', '');
@@ -1,8 +1,8 @@
 mod common;
 use crate::common::*;
-use sqlx::Pool;
 use sqlx::postgres::Postgres;
-use windmill_common::scripts::{ ScriptLang};
+use sqlx::Pool;
+use windmill_common::scripts::ScriptLang;

 #[cfg(feature = "python")]
 #[sqlx::test(fixtures("base", "lockfile_python"))]
@@ -162,7 +162,6 @@ use windmill_common::jobs::RawCode;
 #[cfg(feature = "python")]
 #[sqlx::test(fixtures("base"))]
 async fn test_python_job(db: Pool<Postgres>) -> anyhow::Result<()> {
-
    initialize_tracing().await;
    let server = ApiServer::start(db.clone()).await?;
    let port = server.addr.port();
@@ -211,7 +210,6 @@ async fn test_python_global_site_packages(db: Pool<Postgres>) -> anyhow::Result<

    // 3.12
    {
-
        let content = r#"# py: ==3.12
 #requirements:
 #
@@ -355,7 +353,6 @@ def main():
    Ok(())
 }

-
 #[cfg(feature = "python")]
 #[sqlx::test(fixtures("base", "relative_python"))]
 async fn test_relative_imports_python(db: Pool<Postgres>) -> anyhow::Result<()> {
@@ -391,4 +388,3 @@ def main():
    run_preview_relative_imports(&db, content, ScriptLang::Python3).await?;
    Ok(())
 }
-
@@ -0,0 +1,549 @@
+// TODO: move all related logic here (if anything left anywhere in codebase)
+mod common;
+mod dependency_map {
+    use sqlx::{Pool, Postgres};
+    use tokio_stream::StreamExt;
+    use windmill_api_client::types::NewScript;
+
+    use crate::common::{in_test_worker, listen_for_completed_jobs, ApiServer};
+
+    pub async fn initialize_tracing() {
+        use std::sync::Once;
+
+        static ONCE: Once = Once::new();
+        ONCE.call_once(|| {
+            let _ = windmill_common::tracing_init::initialize_tracing(
+                "test",
+                &windmill_common::utils::Mode::Standalone,
+                "test",
+            );
+        });
+    }
+
+    async fn rebuild_dmap(client: &windmill_api_client::Client) -> bool {
+        client
+            .client()
+            .post(format!(
+                "{}/w/test-workspace/workspaces/rebuild_dependency_map",
+                client.baseurl()
+            ))
+            .send()
+            .await
+            .unwrap()
+            .status()
+            .is_success()
+    }
+
+    async fn init(db: Pool<Postgres>) -> (windmill_api_client::Client, u16, ApiServer) {
+        initialize_tracing().await;
+        let server = ApiServer::start(db).await.unwrap();
+        let port = server.addr.port();
+        let client = windmill_api_client::create_client(
+            &format!("http://localhost:{port}"),
+            "SECRET_TOKEN".to_string(),
+        );
+        (client, port, server)
+    }
+
+    async fn _clear_dmap(db: &Pool<Postgres>) {
+        sqlx::query!("DELETE FROM dependency_map WHERE workspace_id = 'test-workspace'")
+            .execute(db)
+            .await
+            .unwrap();
+    }
+
+    /// Corrects map according to provided replacements.
+    /// Only changes importer_path and/or id
+    /// Does not affect imported_path nor kind!
+    fn corrected_dmap(replacements: Vec<(&str, &str)>) -> Vec<(String, String, String, String)> {
+        CORRECT_DMAP
+            .clone()
+            .into_iter()
+            .map(|e| {
+                let mut r = (
+                    e.0.to_owned(),
+                    e.1.to_owned(),
+                    e.2.to_owned(),
+                    e.3.to_owned(),
+                );
+                for (from, to) in &replacements {
+                    r = (
+                        r.0.replace(from, to),
+                        r.1, // Kind should be immutable
+                        r.2, // Imported path should be immutable
+                        // We do not modify script contents in test, so we can assume scripts always import the same path
+                        // Modification of kind or imported path considered to be incorrect.
+                        r.3.replace(from, to),
+                    );
+                }
+                r
+            })
+            .collect()
+    }
+
+    async fn assert_dmap(
+        db: &Pool<Postgres>,
+        importer: Option<String>,
+        expected: Vec<(
+            impl Into<String>,
+            impl Into<String>,
+            impl Into<String>,
+            impl Into<String>,
+        )>,
+    ) {
+        let dmap = sqlx::query_as::<_, (String, String, String, String)>(
+            "SELECT importer_path, importer_kind::text, imported_path, importer_node_id FROM dependency_map WHERE workspace_id = 'test-workspace' AND ($1::text IS NULL OR importer_path = $1::text)",
+        )
+        .bind(importer)
+        .fetch_all(db)
+        .await
+        .unwrap();
+
+        assert_eq!(
+            dmap,
+            expected
+                .into_iter()
+                .map(|(f, s, t, fo)| (f.into(), s.into(), t.into(), fo.into()))
+                .collect::<Vec<(String, String, String, String)>>()
+        );
+    }
+
+    fn quick_ns(
+        content: &str,
+        language: windmill_api_client::types::ScriptLang,
+        path: &str,
+        lock: Option<String>,
+        parent_hash: Option<String>,
+    ) -> NewScript {
+        NewScript {
+            content: content.into(),
+            language,
+            lock,
+            parent_hash,
+            path: path.into(),
+            concurrent_limit: None,
+            concurrency_time_window_s: None,
+            cache_ttl: None,
+            dedicated_worker: None,
+            description: "".to_string(),
+            draft_only: None,
+            envs: vec![],
+            is_template: None,
+            kind: None,
+            summary: "".to_string(),
+            tag: None,
+            schema: std::collections::HashMap::new(),
+            ws_error_handler_muted: Some(false),
+            priority: None,
+            delete_after_use: None,
+            timeout: None,
+            restart_unless_cancelled: None,
+            deployment_message: None,
+            concurrency_key: None,
+            visible_to_runner_only: None,
+            no_main_func: None,
+            codebase: None,
+            has_preprocessor: None,
+            on_behalf_of_email: None,
+            assets: vec![],
+        }
+    }
+
+    lazy_static::lazy_static! {
+    pub static ref CORRECT_DMAP: Vec<(&'static str, &'static str, &'static str, &'static str)> =  vec![
+        ("f/rel/branch", "script", "f/rel/leaf_1", ""),
+        ("f/rel/root_script", "script", "f/rel/branch", ""),
+        ("f/rel/root_script", "script", "f/rel/leaf_1", ""),
+        ("f/rel/root_script", "script", "f/rel/leaf_2", ""),
+        ("f/rel/root_app", "app", "f/rel/leaf_2", "dontpressmeplz"),
+        ("f/rel/root_flow", "flow", "f/rel/branch", "nstep1"),
+        ("f/rel/root_flow", "flow", "f/rel/leaf_1", "nstep1"),
+        ("f/rel/root_flow", "flow", "f/rel/leaf_2", "nstep1"),
+        ("f/rel/root_flow", "flow", "f/rel/leaf_2", "nstep2_2"),
+        ("f/rel/root_flow", "flow", "f/rel/branch", "nstep4_1"),
+        ("f/rel/root_flow", "flow", "f/rel/branch", "nstep5_1"),
+        ("f/rel/root_flow", "flow", "f/rel/leaf_1", "nstep5_1"),
+        ("f/rel/root_flow", "flow", "f/rel/leaf_2", "nstep5_1"),
+        ("f/rel/root_app", "app", "f/rel/branch", "pressmeplz"),
+        ("f/rel/root_app", "app", "f/rel/leaf_1", "pressmeplz"),
+        ("f/rel/root_app", "app", "f/rel/leaf_2", "pressmeplz"),
+        ("f/rel/root_app", "app", "f/rel/branch", "youcanpressme")];
+    }
+
+    #[cfg(feature = "python")]
+    #[sqlx::test(fixtures("base", "dependency_map"))]
+    async fn relative_imports_test_rebuild_correctness(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let (client, _port, _s) = init(db.clone()).await;
+        assert_dmap(&db, None, CORRECT_DMAP.clone()).await;
+        // rebuild map
+        assert!(rebuild_dmap(&client).await);
+        assert_dmap(&db, None, CORRECT_DMAP.clone()).await;
+        Ok(())
+    }
+
+    #[cfg(feature = "python")]
+    #[sqlx::test(fixtures("base", "dependency_map"))]
+    async fn relative_imports_test_rebuild_lock(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let (client, _port, _s) = init(db.clone()).await;
+
+        // Spawn first rebuild
+        let handle = {
+            let client = client.clone();
+            tokio::spawn(async move { rebuild_dmap(&client).await })
+        };
+
+        // Immidiately spawn another
+        let res = client
+            .client()
+            .post(format!(
+                "{}/w/test-workspace/workspaces/rebuild_dependency_map",
+                client.baseurl()
+            ))
+            .send()
+            .await
+            .unwrap()
+            .text()
+            .await
+            .unwrap();
+
+        // Should tell us there is already rebuilt in progress
+        // Or if it is too fast we will be able to trigger it second time
+        assert!(&res == "There is already one task pending, try again later." || &res == "Success");
+
+        assert!(handle.await.unwrap());
+        Ok(())
+    }
+
+    // If you deploy from cli and you use raw requirements you don't want the script be included in dmap
+    // Otherwise script will be overwritten once any relative import is updated
+    #[cfg(feature = "python")]
+    #[sqlx::test(fixtures("base", "dependency_map"))]
+    async fn relative_imports_test_with_requirements_txt(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let (client, _port, _s) = init(db.clone()).await;
+
+        client
+            .create_script(
+                "test-workspace",
+                &quick_ns(
+                    "
+from f.rel.branch import main as br;
+from f.rel.leaf_1 import main as lf_1;
+from f.rel.leaf_2 import main as lf_2;
+
+def main():
+    return [br(), lf_1(), lf_2];
+                            ",
+                    windmill_api_client::types::ScriptLang::Python3,
+                    "f/rel/root_script",
+                    Some(format!("# from requirements.txt")),
+                    Some("000000000005165B".into()),
+                ),
+            )
+            .await
+            .unwrap();
+
+        assert_dmap(
+            &db,
+            Some("f/rel/root_script".into()),
+            vec![
+                ("f/rel/root_script", "script", "f/rel/branch", ""),
+                ("f/rel/root_script", "script", "f/rel/leaf_1", ""),
+                ("f/rel/root_script", "script", "f/rel/leaf_2", ""),
+            ],
+        )
+        .await;
+
+        tokio::time::sleep(std::time::Duration::from_secs(13)).await;
+
+        assert_dmap(
+            &db,
+            Some("f/rel/root_script".into()),
+            Vec::<(String, String, String, String)>::new(),
+        )
+        .await;
+
+        Ok(())
+    }
+
+    #[cfg(feature = "python")]
+    #[sqlx::test(fixtures("base", "dependency_map"))]
+    async fn relative_imports_test_without_requirements_txt(
+        db: Pool<Postgres>,
+    ) -> anyhow::Result<()> {
+        let (client, _port, _s) = init(db.clone()).await;
+
+        client
+            .create_script(
+                "test-workspace",
+                &quick_ns(
+                    "
+from f.rel.branch import main as br;
+from f.rel.leaf_1 import main as lf_1;
+from f.rel.leaf_2 import main as lf_2;
+
+def main():
+    return [br(), lf_1(), lf_2];
+                            ",
+                    windmill_api_client::types::ScriptLang::Python3,
+                    "f/rel/root_script",
+                    // We still want to pass lock to it.
+                    Some(format!("# py311")),
+                    Some("000000000005165B".into()),
+                ),
+            )
+            .await
+            .unwrap();
+        assert_dmap(&db, None, CORRECT_DMAP.clone()).await;
+        tokio::time::sleep(std::time::Duration::from_secs(13)).await;
+        assert_dmap(&db, None, CORRECT_DMAP.clone()).await;
+        Ok(())
+    }
+    // Consider simple one. Only referenced directly. No deep connections
+    #[cfg(feature = "python")]
+    #[sqlx::test(fixtures("base", "dependency_map"))]
+    async fn relative_imports_test_rename_leaf_2(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let (client, port, _s) = init(db.clone()).await;
+        client
+            .create_script(
+                "test-workspace",
+                &quick_ns(
+                    "
+def main():
+    return 'leaf3';
+                    ",
+                    windmill_api_client::types::ScriptLang::Python3,
+                    "f/rel/leaf_2_renamed",
+                    None,
+                    Some("0000000000051659".into()),
+                ),
+            )
+            .await
+            .unwrap();
+
+        let mut completed = listen_for_completed_jobs(&db).await;
+        in_test_worker(&db, completed.next(), port).await;
+
+        // Changing leafs should not change dependency map
+        assert_dmap(&db, None, CORRECT_DMAP.clone()).await;
+        Ok(())
+    }
+
+    // Consider hard one. Referenced deeply and exists in double references.
+    #[cfg(feature = "python")]
+    #[sqlx::test(fixtures("base", "dependency_map"))]
+    async fn relative_imports_test_rename_leaf_1(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let (client, port, _s) = init(db.clone()).await;
+        client
+            .create_script(
+                "test-workspace",
+                &quick_ns(
+                    "
+def main():
+    return 'leaf1';
+                    ",
+                    windmill_api_client::types::ScriptLang::Python3,
+                    "f/rel/leaf_1_renamed",
+                    None,
+                    Some("0000000000051658".into()),
+                ),
+            )
+            .await
+            .unwrap();
+
+        let mut completed = listen_for_completed_jobs(&db).await;
+        in_test_worker(&db, completed.next(), port).await;
+
+        // Changing leafs should not change dependency map
+        assert_dmap(&db, None, CORRECT_DMAP.clone()).await;
+        Ok(())
+    }
+
+    #[cfg(feature = "python")]
+    #[sqlx::test(fixtures("base", "dependency_map"))]
+    async fn relative_imports_test_rename_branch(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let (client, port, _s) = init(db.clone()).await;
+        client
+            .create_script(
+                "test-workspace",
+                &quick_ns(
+                    "
+from f.rel.leaf_1 import main as lf_1;
+
+def main():
+    return lf_1();
+                    ",
+                    windmill_api_client::types::ScriptLang::Python3,
+                    "f/rel/branch_renamed",
+                    None,
+                    Some("000000000005165A".into()),
+                ),
+            )
+            .await
+            .unwrap();
+
+        let mut completed = listen_for_completed_jobs(&db).await;
+        in_test_worker(&db, completed.next(), port).await;
+
+        // Changing branches SHOULD change dependency map
+        // Though it should only change branch item in dmap when it is importer.
+        // All entries when branch is imported should not change.
+        let mut corrected_dmap = CORRECT_DMAP.clone();
+        // Corresponds to importer path of branch entry
+        corrected_dmap[0].0 = "f/rel/branch_renamed";
+        assert_dmap(&db, None, corrected_dmap).await;
+        Ok(())
+    }
+
+    #[cfg(feature = "python")]
+    #[sqlx::test(fixtures("base", "dependency_map"))]
+    async fn relative_imports_test_rename_primary_script(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let (client, port, _s) = init(db.clone()).await;
+
+        client
+            .create_script(
+                "test-workspace",
+                &quick_ns(
+                    "
+from f.rel.branch import main as br;
+from f.rel.leaf_1 import main as lf_1;
+from f.rel.leaf_2 import main as lf_2;
+
+def main():
+    return [br(), lf_1(), lf_2];
+                            ",
+                    windmill_api_client::types::ScriptLang::Python3,
+                    "f/rel/root_script_renamed",
+                    None,
+                    Some("000000000005165B".into()),
+                ),
+            )
+            .await
+            .unwrap();
+
+        let corrected_dmap = corrected_dmap(vec![("root_script", "root_script_renamed")]);
+        let mut completed = listen_for_completed_jobs(&db).await;
+        in_test_worker(&db, completed.next(), port).await;
+        assert_dmap(&db, None, corrected_dmap.clone()).await;
+        Ok(())
+    }
+
+    #[cfg(feature = "python")]
+    #[sqlx::test(fixtures("base", "dependency_map"))]
+    async fn relative_imports_test_rename_primary_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
+        use windmill_common::{cache::flow::fetch_version, flows::NewFlow};
+
+        let (client, port, _s) = init(db.clone()).await;
+        let flow = fetch_version(&db, 1443253234253454).await.unwrap();
+        let res = client
+            .client()
+            .post(format!(
+                "{}/w/test-workspace/flows/update/{}",
+                client.baseurl(),
+                "f/rel/root_flow" // encode_path()
+            ))
+            .json(&NewFlow {
+                path: "f/rel/root_flow_renamed".into(),
+                summary: "".into(),
+                description: None,
+                value: serde_json::from_str(
+                    &serde_json::to_string(flow.value())
+                        .unwrap()
+                        .replace("nstep1", "Foxes")
+                        .replace("nstep2_2", "like")
+                        .replace("nstep_4_1", "Emeralds"),
+                )
+                .unwrap(),
+                schema: None,
+                draft_only: None,
+                tag: None,
+                dedicated_worker: None,
+                timeout: None,
+                deployment_message: None,
+                visible_to_runner_only: None,
+                on_behalf_of_email: None,
+            })
+            .send()
+            .await
+            .unwrap();
+
+        assert_eq!(res.text().await.unwrap(), "f/rel/root_flow_renamed");
+
+        let mut completed = listen_for_completed_jobs(&db).await;
+        in_test_worker(&db, completed.next(), port).await;
+
+        assert_dmap(
+            &db,
+            None,
+            corrected_dmap(vec![
+                ("f/rel/root_flow", "f/rel/root_flow_renamed"),
+                ("nstep1", "Foxes"),
+                ("nstep2_2", "like"),
+                ("nstep_4_1", "Emeralds"),
+            ]),
+        )
+        .await;
+        Ok(())
+    }
+
+    #[cfg(feature = "python")]
+    #[sqlx::test(fixtures("base", "dependency_map"))]
+    async fn relative_imports_test_rename_primary_app(db: Pool<Postgres>) -> anyhow::Result<()> {
+        let (client, port, _s) = init(db.clone()).await;
+
+        let app_value: String =
+            sqlx::query_scalar!("SELECT value::text FROM app_version WHERE id = 0 AND app_id = 2")
+                .fetch_one(&db)
+                .await
+                .unwrap()
+                .unwrap();
+
+        // TODO: There is:
+        // 1. update app
+        // 2. create app
+        // 3. update app raw
+        // Ideally all of them should be handled
+        let res = client
+            .client()
+            .post(format!(
+                "{}/w/test-workspace/apps/update/{}",
+                client.baseurl(),
+                "f/rel/root_app" // encode_path()
+            ))
+            .json(&windmill_api::EditApp {
+                path: Some("f/rel/root_app_renamed".into()),
+                summary: None,
+                value: serde_json::from_str(
+                    &app_value
+                        .replace("dontpressmeplz", "Apps")
+                        .replace("youcanpressme", "Work"),
+                )
+                .unwrap(),
+                policy: None,
+                deployment_message: None,
+                custom_path: None,
+            })
+            .send()
+            .await
+            .unwrap();
+
+        assert_eq!(
+            res.text().await.unwrap(),
+            "app f/rel/root_app updated (npath: \"f/rel/root_app_renamed\")"
+        );
+
+        let mut completed = listen_for_completed_jobs(&db).await;
+        in_test_worker(&db, completed.next(), port).await;
+
+        assert_dmap(
+            &db,
+            None,
+            corrected_dmap(vec![
+                ("f/rel/root_app", "f/rel/root_app_renamed"),
+                ("dontpressmeplz", "Apps"),
+                ("youcanpressme", "Work"),
+            ]),
+        )
+        .await;
+        Ok(())
+    }
+}
@@ -1,6 +1,5 @@
 use serde::de::DeserializeOwned;

-
 #[cfg(feature = "enterprise")]
 use chrono::Timelike;

@@ -20,13 +19,12 @@ use windmill_api_client::types::{EditSchedule, NewSchedule, ScriptArgs};
 use windmill_common::flows::InputTransform;

 #[cfg(any(feature = "python", feature = "deno_core"))]
-use windmill_common::flow_status::{RestartedFrom};
+use windmill_common::flow_status::RestartedFrom;

 use windmill_common::{
-    flows::{ FlowValue},
-    jobs::{ JobPayload, RawCode},
-    scripts::{ScriptLang},
-
+    flows::FlowValue,
+    jobs::{JobPayload, RawCode},
+    scripts::ScriptLang,
 };
 mod common;
 use common::*;
@@ -34,7 +32,6 @@ use common::*;
 #[cfg(feature = "enterprise")]
 use futures::StreamExt;

-
 // async fn _print_job(id: Uuid, db: &Pool<Postgres>) -> Result<(), anyhow::Error> {
 //     tracing::info!(
 //         "{:#?}",
@@ -45,7 +42,6 @@ use futures::StreamExt;
 //     Ok(())
 // }

-
 #[cfg(feature = "deno_core")]
 #[sqlx::test(fixtures("base"))]
 async fn test_iteration(db: Pool<Postgres>) -> anyhow::Result<()> {
@@ -167,8 +163,6 @@ async fn test_iteration_parallel(db: Pool<Postgres>) -> anyhow::Result<()> {
    Ok(())
 }

-
-
 #[cfg(feature = "deno_core")]
 #[sqlx::test(fixtures("base"))]
 async fn test_deno_flow(db: Pool<Postgres>) -> anyhow::Result<()> {
@@ -341,7 +335,6 @@ use windmill_common::flows::FlowModuleValue;
 #[cfg(feature = "deno_core")]
 #[sqlx::test(fixtures("base"))]
 async fn test_deno_flow_same_worker(db: Pool<Postgres>) -> anyhow::Result<()> {
-
    initialize_tracing().await;

    let server = ApiServer::start(db.clone()).await?;
@@ -1138,8 +1131,6 @@ public class Main {
    Ok(())
 }

-
-
 #[sqlx::test(fixtures("base"))]
 async fn test_bun_job_datetime(db: Pool<Postgres>) -> anyhow::Result<()> {
    initialize_tracing().await;
@@ -2627,7 +2618,6 @@ async fn test_flow_schedule_handlers(db: Pool<Postgres>) -> anyhow::Result<()> {
    Ok(())
 }

-
 #[sqlx::test(fixtures("base", "relative_bun"))]
 async fn test_relative_imports_bun(db: Pool<Postgres>) -> anyhow::Result<()> {
    let content = r#"
@@ -2699,8 +2689,6 @@ export async function main() {
    Ok(())
 }

-
-
 #[sqlx::test(fixtures("base", "result_format"))]
 async fn test_result_format(db: Pool<Postgres>) -> anyhow::Result<()> {
    let ordered_result_job_id = "1eecb96a-c8b0-4a3d-b1b6-087878c55e41";
@@ -2929,4 +2917,3 @@ async fn test_workflow_as_code(db: Pool<Postgres>) -> anyhow::Result<()> {
    .await;
    Ok(())
 }
-
@@ -8,9 +8,80 @@ pub mod types {
    #[allow(unused_imports)]
    use std::convert::TryFrom;
    #[derive(Clone, Debug, Deserialize, Serialize)]
+    pub struct AiAgent {
+        pub input_transforms: std::collections::HashMap<String, InputTransform>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub parallel: Option<bool>,
+        pub tools: Vec<FlowModule>,
+        #[serde(rename = "type")]
+        pub type_: AiAgentType,
+    }
+    impl From<&AiAgent> for AiAgent {
+        fn from(value: &AiAgent) -> Self {
+            value.clone()
+        }
+    }
+    #[derive(
+        Clone,
+        Copy,
+        Debug,
+        Deserialize,
+        Eq,
+        Hash,
+        Ord,
+        PartialEq,
+        PartialOrd,
+        Serialize
+    )]
+    pub enum AiAgentType {
+        #[serde(rename = "aiagent")]
+        Aiagent,
+    }
+    impl From<&AiAgentType> for AiAgentType {
+        fn from(value: &AiAgentType) -> Self {
+            value.clone()
+        }
+    }
+    impl ToString for AiAgentType {
+        fn to_string(&self) -> String {
+            match *self {
+                Self::Aiagent => "aiagent".to_string(),
+            }
+        }
+    }
+    impl std::str::FromStr for AiAgentType {
+        type Err = &'static str;
+        fn from_str(value: &str) -> Result<Self, &'static str> {
+            match value {
+                "aiagent" => Ok(Self::Aiagent),
+                _ => Err("invalid value"),
+            }
+        }
+    }
+    impl std::convert::TryFrom<&str> for AiAgentType {
+        type Error = &'static str;
+        fn try_from(value: &str) -> Result<Self, &'static str> {
+            value.parse()
+        }
+    }
+    impl std::convert::TryFrom<&String> for AiAgentType {
+        type Error = &'static str;
+        fn try_from(value: &String) -> Result<Self, &'static str> {
+            value.parse()
+        }
+    }
+    impl std::convert::TryFrom<String> for AiAgentType {
+        type Error = &'static str;
+        fn try_from(value: String) -> Result<Self, &'static str> {
+            value.parse()
+        }
+    }
+    #[derive(Clone, Debug, Deserialize, Serialize)]
    pub struct AiConfig {
        #[serde(default, skip_serializing_if = "Option::is_none")]
        pub code_completion_model: Option<AiProviderModel>,
+        #[serde(default, skip_serializing_if = "std::collections::HashMap::is_empty")]
+        pub custom_prompts: std::collections::HashMap<String, String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        pub default_model: Option<AiProviderModel>,
        #[serde(default, skip_serializing_if = "std::collections::HashMap::is_empty")]
@@ -1295,8 +1366,8 @@ pub mod types {
        Websocket,
        #[serde(rename = "kafka")]
        Kafka,
-        #[serde(rename = "email")]
-        Email,
+        #[serde(rename = "default_email")]
+        DefaultEmail,
        #[serde(rename = "nats")]
        Nats,
        #[serde(rename = "postgres")]
@@ -1307,6 +1378,8 @@ pub mod types {
        Mqtt,
        #[serde(rename = "gcp")]
        Gcp,
+        #[serde(rename = "email")]
+        Email,
    }
    impl From<&CaptureTriggerKind> for CaptureTriggerKind {
        fn from(value: &CaptureTriggerKind) -> Self {
@@ -1320,12 +1393,13 @@ pub mod types {
                Self::Http => "http".to_string(),
                Self::Websocket => "websocket".to_string(),
                Self::Kafka => "kafka".to_string(),
-                Self::Email => "email".to_string(),
+                Self::DefaultEmail => "default_email".to_string(),
                Self::Nats => "nats".to_string(),
                Self::Postgres => "postgres".to_string(),
                Self::Sqs => "sqs".to_string(),
                Self::Mqtt => "mqtt".to_string(),
                Self::Gcp => "gcp".to_string(),
+                Self::Email => "email".to_string(),
            }
        }
    }
@@ -1337,12 +1411,13 @@ pub mod types {
                "http" => Ok(Self::Http),
                "websocket" => Ok(Self::Websocket),
                "kafka" => Ok(Self::Kafka),
-                "email" => Ok(Self::Email),
+                "default_email" => Ok(Self::DefaultEmail),
                "nats" => Ok(Self::Nats),
                "postgres" => Ok(Self::Postgres),
                "sqs" => Ok(Self::Sqs),
                "mqtt" => Ok(Self::Mqtt),
                "gcp" => Ok(Self::Gcp),
+                "email" => Ok(Self::Email),
                _ => Err("invalid value"),
            }
        }
@@ -1493,6 +1568,8 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
        Flownode,
        #[serde(rename = "appscript")]
        Appscript,
+        #[serde(rename = "aiagent")]
+        Aiagent,
    }
    impl From<&CompletedJobJobKind> for CompletedJobJobKind {
        fn from(value: &CompletedJobJobKind) -> Self {
@@ -1516,6 +1593,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
                Self::Flowscript => "flowscript".to_string(),
                Self::Flownode => "flownode".to_string(),
                Self::Appscript => "appscript".to_string(),
+                Self::Aiagent => "aiagent".to_string(),
            }
        }
    }
@@ -1537,6 +1615,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
                "flowscript" => Ok(Self::Flowscript),
                "flownode" => Ok(Self::Flownode),
                "appscript" => Ok(Self::Appscript),
+                "aiagent" => Ok(Self::Aiagent),
                _ => Err("invalid value"),
            }
        }
@@ -1705,6 +1784,21 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
        }
    }
    #[derive(Clone, Debug, Deserialize, Serialize)]
+    pub struct CreateWorkspaceFork {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub color: Option<String>,
+        pub id: String,
+        pub name: String,
+        pub parent_workspace_id: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub username: Option<String>,
+    }
+    impl From<&CreateWorkspaceFork> for CreateWorkspaceFork {
+        fn from(value: &CreateWorkspaceFork) -> Self {
+            value.clone()
+        }
+    }
+    #[derive(Clone, Debug, Deserialize, Serialize)]
    pub struct CriticalAlert {
        ///Acknowledgment status of the alert, can be true, false, or null if not set
        #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -1799,6 +1893,24 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
        }
    }
    #[derive(Clone, Debug, Deserialize, Serialize)]
+    pub struct DependencyMap {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub imported_path: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub importer_kind: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub importer_node_id: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub importer_path: Option<String>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub workspace_id: Option<String>,
+    }
+    impl From<&DependencyMap> for DependencyMap {
+        fn from(value: &DependencyMap) -> Self {
+            value.clone()
+        }
+    }
+    #[derive(Clone, Debug, Deserialize, Serialize)]
    pub struct DucklakeSettings {
        pub ducklakes: std::collections::HashMap<String, DucklakeSettingsDucklakesValue>,
    }
@@ -1909,6 +2021,27 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
        }
    }
    #[derive(Clone, Debug, Deserialize, Serialize)]
+    pub struct EditEmailTrigger {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub error_handler_args: Option<ScriptArgs>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub error_handler_path: Option<String>,
+        pub is_flow: bool,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub local_part: Option<String>,
+        pub path: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub retry: Option<Retry>,
+        pub script_path: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub workspaced_local_part: Option<bool>,
+    }
+    impl From<&EditEmailTrigger> for EditEmailTrigger {
+        fn from(value: &EditEmailTrigger) -> Self {
+            value.clone()
+        }
+    }
+    #[derive(Clone, Debug, Deserialize, Serialize)]
    pub struct EditHttpTrigger {
        pub authentication_method: AuthenticationMethod,
        #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -2234,6 +2367,23 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
        }
    }
    #[derive(Clone, Debug, Deserialize, Serialize)]
+    pub struct EmailTrigger {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub error_handler_args: Option<ScriptArgs>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub error_handler_path: Option<String>,
+        pub local_part: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub retry: Option<Retry>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub workspaced_local_part: Option<bool>,
+    }
+    impl From<&EmailTrigger> for EmailTrigger {
+        fn from(value: &EmailTrigger) -> Self {
+            value.clone()
+        }
+    }
+    #[derive(Clone, Debug, Deserialize, Serialize)]
    pub struct EndpointTool {
        ///JSON schema for request body
        #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -2481,7 +2631,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
        #[serde(default, skip_serializing_if = "Option::is_none")]
        pub suspend: Option<FlowModuleSuspend>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
-        pub timeout: Option<f64>,
+        pub timeout: Option<InputTransform>,
        pub value: FlowModuleValue,
    }
    impl From<&FlowModule> for FlowModule {
@@ -2555,6 +2705,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
        BranchOne(BranchOne),
        BranchAll(BranchAll),
        Identity(Identity),
+        AiAgent(AiAgent),
    }
    impl From<&FlowModuleValue> for FlowModuleValue {
        fn from(value: &FlowModuleValue) -> Self {
@@ -2601,6 +2752,11 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
            Self::Identity(value)
        }
    }
+    impl From<AiAgent> for FlowModuleValue {
+        fn from(value: AiAgent) -> Self {
+            Self::AiAgent(value)
+        }
+    }
    #[derive(Clone, Debug, Deserialize, Serialize)]
    pub struct FlowPreview {
        pub args: ScriptArgs,
@@ -2648,6 +2804,10 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
    }
    #[derive(Clone, Debug, Deserialize, Serialize)]
    pub struct FlowStatusModule {
+        #[serde(default, skip_serializing_if = "Vec::is_empty")]
+        pub agent_actions: Vec<std::collections::HashMap<String, serde_json::Value>>,
+        #[serde(default, skip_serializing_if = "Vec::is_empty")]
+        pub agent_actions_success: Vec<bool>,
        #[serde(default, skip_serializing_if = "Vec::is_empty")]
        pub approvers: Vec<FlowStatusModuleApproversItem>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -4810,6 +4970,26 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
        }
    }
    #[derive(Clone, Debug, Deserialize, Serialize)]
+    pub struct NewEmailTrigger {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub error_handler_args: Option<ScriptArgs>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub error_handler_path: Option<String>,
+        pub is_flow: bool,
+        pub local_part: String,
+        pub path: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub retry: Option<Retry>,
+        pub script_path: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub workspaced_local_part: Option<bool>,
+    }
+    impl From<&NewEmailTrigger> for NewEmailTrigger {
+        fn from(value: &NewEmailTrigger) -> Self {
+            value.clone()
+        }
+    }
+    #[derive(Clone, Debug, Deserialize, Serialize)]
    pub struct NewHttpTrigger {
        pub authentication_method: AuthenticationMethod,
        #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -5907,6 +6087,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
    #[derive(Clone, Debug, Deserialize, Serialize)]
    pub struct Preview {
        pub args: ScriptArgs,
+        ///The code to run
        #[serde(default, skip_serializing_if = "Option::is_none")]
        pub content: Option<String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -5917,8 +6098,10 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
        pub language: Option<ScriptLang>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        pub lock: Option<String>,
+        ///The path to the script
        #[serde(default, skip_serializing_if = "Option::is_none")]
        pub path: Option<String>,
+        ///The hash of the script
        #[serde(default, skip_serializing_if = "Option::is_none")]
        pub script_hash: Option<String>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -6127,6 +6310,8 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
        Flownode,
        #[serde(rename = "appscript")]
        Appscript,
+        #[serde(rename = "aiagent")]
+        Aiagent,
    }
    impl From<&QueuedJobJobKind> for QueuedJobJobKind {
        fn from(value: &QueuedJobJobKind) -> Self {
@@ -6150,6 +6335,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
                Self::Flowscript => "flowscript".to_string(),
                Self::Flownode => "flownode".to_string(),
                Self::Appscript => "appscript".to_string(),
+                Self::Aiagent => "aiagent".to_string(),
            }
        }
    }
@@ -6171,6 +6357,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
                "flowscript" => Ok(Self::Flowscript),
                "flownode" => Ok(Self::Flownode),
                "appscript" => Ok(Self::Appscript),
+                "aiagent" => Ok(Self::Aiagent),
                _ => Err("invalid value"),
            }
        }
@@ -6676,6 +6863,8 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
        pub constant: Option<RetryConstant>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        pub exponential: Option<RetryExponential>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub retry_if: Option<RetryRetryIf>,
    }
    impl From<&Retry> for Retry {
        fn from(value: &Retry) -> Self {
@@ -6710,6 +6899,15 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
            value.clone()
        }
    }
+    #[derive(Clone, Debug, Deserialize, Serialize)]
+    pub struct RetryRetryIf {
+        pub expr: String,
+    }
+    impl From<&RetryRetryIf> for RetryRetryIf {
+        fn from(value: &RetryRetryIf) -> Self {
+            value.clone()
+        }
+    }
    #[derive(
        Clone,
        Copy,
@@ -7861,6 +8059,8 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
    }
    #[derive(Clone, Debug, Deserialize, Serialize)]
    pub struct TriggersCount {
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub default_email_count: Option<f64>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        pub email_count: Option<f64>,
        #[serde(default, skip_serializing_if = "Option::is_none")]
@@ -8069,10 +8269,14 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
    #[derive(Clone, Debug, Deserialize, Serialize)]
    pub struct UserWorkspaceListWorkspacesItem {
        pub color: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub created_by: Option<String>,
        pub id: String,
        pub name: String,
        #[serde(default, skip_serializing_if = "Option::is_none")]
        pub operator_settings: Option<OperatorSettings>,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub parent_workspace_id: Option<String>,
        pub username: String,
    }
    impl From<&UserWorkspaceListWorkspacesItem> for UserWorkspaceListWorkspacesItem {
@@ -8479,6 +8683,8 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
        pub id: String,
        pub name: String,
        pub owner: String,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub parent_workspace_id: Option<String>,
    }
    impl From<&Workspace> for Workspace {
        fn from(value: &Workspace) -> Self {
@@ -8550,6 +8756,8 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
        pub email: String,
        pub is_admin: bool,
        pub operator: bool,
+        #[serde(default, skip_serializing_if = "Option::is_none")]
+        pub parent_workspace_id: Option<String>,
        pub workspace_id: String,
    }
    impl From<&WorkspaceInvite> for WorkspaceInvite {
@@ -8561,7 +8769,7 @@ the execution of this script will be permissioned_as and by extension its DT_TOK
 #[derive(Clone, Debug)]
 /**Client for Windmill API

-Version: 1.526.1*/
+Version: 1.543.0*/
 pub struct Client {
    pub(crate) baseurl: String,
    pub(crate) client: reqwest::Client,
@@ -8607,7 +8815,7 @@ impl Client {
    /// This string is pulled directly from the source OpenAPI
    /// document and may be in any format the API selects.
    pub fn api_version(&self) -> &'static str {
-        "1.526.1"
+        "1.543.0"
    }
 }
 impl Client {
@@ -2134,6 +2134,40 @@ paths:
              schema:
                type: string

+  /w/{workspace}/workspaces/rebuild_dependency_map:
+    post:
+      summary: rebuild dependency map
+      operationId: rebuildDependencyMap
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+      responses:
+        "200":
+          description: status
+          content:
+            text/plain:
+              schema:
+                type: string
+
+  /w/{workspace}/workspaces/get_dependency_map:
+    get:
+      summary: get dependency map
+      operationId: getDependencyMap
+      tags:
+        - workspace
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+      responses:
+        "200":
+          description: dmap
+          content:
+            application/json:
+              schema:
+                type: array
+                items:
+                  $ref: "#/components/schemas/DependencyMap"
+
  /w/{workspace}/workspaces/edit_slack_command:
    post:
      summary: edit slack command
@@ -17645,6 +17679,25 @@ components:
        - owner
        - created_at

+    DependencyMap:
+      type: object
+      properties:
+        workspace_id:
+          type: string
+          nullable: true
+        importer_path:
+          type: string
+          nullable: true
+        importer_kind:
+          type: string
+          nullable: true
+        imported_path:
+          type: string
+          nullable: true
+        importer_node_id:
+          type: string
+          nullable: true
+
    WorkspaceInvite:
      type: object
      properties:
@@ -268,7 +268,7 @@ pub struct CreateApp {
    pub custom_path: Option<String>,
 }

-#[derive(Deserialize)]
+#[derive(Serialize, Deserialize)]
 pub struct EditApp {
    pub path: Option<String>,
    pub summary: Option<String>,
@@ -87,7 +87,7 @@ pub mod ee;
 pub mod ee_oss;
 pub mod embeddings;
 mod favorite;
-mod flows;
+pub mod flows;
 mod folders;
 mod granular_acls;
 mod groups;
@@ -182,6 +182,7 @@ mod workspaces_oss;
 #[cfg(feature = "mcp")]
 mod mcp;

+pub use apps::EditApp;
 pub const DEFAULT_BODY_LIMIT: usize = 2097152 * 100; // 200MB

 lazy_static::lazy_static! {
@@ -1036,6 +1036,9 @@ async fn create_script_internal<'c>(
            let content = ns.content.clone();
            let language = ns.language.clone();
            tokio::spawn(async move {
+                // TODO: I don't think we want this. We might want to send dependency job. But skip any calculations if lock is already present.
+                // It will allow us to make code more consistent and predictable.
+
                // wait for 10 seconds to make sure the script is deployed and that the CLI sync that pushed it (f one) is complete
                tokio::time::sleep(std::time::Duration::from_secs(10)).await;
                if let Err(e) = process_relative_imports(
@@ -51,6 +51,7 @@ use windmill_common::{
    utils::{paginate, rd_string, require_admin, Pagination},
 };
 use windmill_git_sync::{handle_deployment_metadata, DeployedObject};
+use windmill_worker::scoped_dependency_map::{DependencyMap, ScopedDependencyMap};

 #[cfg(feature = "enterprise")]
 use windmill_common::utils::require_admin_or_devops;
@@ -78,6 +79,8 @@ pub fn workspaced_service() -> Router {
        .route("/invite_user", post(invite_user))
        .route("/add_user", post(add_user))
        .route("/delete_invite", post(delete_invite))
+        .route("/rebuild_dependency_map", post(rebuild_dependency_map))
+        .route("/get_dependency_map", get(get_dependency_map))
        .route("/get_settings", get(get_settings))
        .route("/get_deploy_to", get(get_deploy_to))
        .route("/edit_slack_command", post(edit_slack_command))
@@ -3351,6 +3354,42 @@ async fn get_workspace_name(
    Ok(workspace)
 }

+async fn get_dependency_map(
+    authed: ApiAuthed,
+    Path(w_id): Path<String>,
+    Extension(user_db): Extension<UserDB>,
+) -> JsonResult<Vec<DependencyMap>> {
+    require_admin(authed.is_admin, &authed.username)?;
+
+    let mut tx = user_db.begin(&authed).await?;
+    let dmap = sqlx::query_as!(
+        DependencyMap,
+        "
+        SELECT workspace_id, importer_path, importer_kind::text, imported_path, importer_node_id
+        FROM dependency_map WHERE workspace_id = $1",
+        &w_id
+    )
+    .fetch_all(&mut *tx)
+    .await?;
+
+    tx.commit().await?;
+
+    Ok(Json(dmap))
+}
+
+#[axum::debug_handler]
+async fn rebuild_dependency_map(
+    Extension(db): Extension<DB>,
+    Path(w_id): Path<String>,
+    authed: ApiAuthed,
+) -> Result<String> {
+    require_admin(authed.is_admin, &authed.username)?;
+    if *CLOUD_HOSTED {
+        return Err(Error::BadRequest("Disabled on Cloud".into()));
+    }
+    ScopedDependencyMap::rebuild_map(&w_id, &db).await
+}
+
 #[derive(Deserialize)]
 struct ChangeWorkspaceName {
    new_name: String,
@@ -9,8 +9,11 @@
 use std::{collections::HashMap, sync::Arc};

 use serde::{Deserialize, Serialize};
+use serde_json::{from_value, Value};
 use tokio::sync::RwLock;

+use crate::{error, scripts::ScriptLang};
+
 lazy_static::lazy_static! {
    pub static ref APP_WORKSPACED_ROUTE: Arc<RwLock<bool>> = Arc::new(RwLock::new(false));
 }
@@ -33,3 +36,70 @@ pub struct ListAppQuery {
 pub struct RawAppValue {
    pub files: HashMap<String, String>,
 }
+
+pub struct AppInlineScript {
+    pub language: Option<ScriptLang>,
+    pub content: String,
+    pub lock: Option<String>,
+}
+
+/// Traverse FlowValue while invoking provided by caller callback on leafs
+// #[async_recursion::async_recursion(?Send)]
+pub fn traverse_app_inline_scripts<
+    C: FnMut(AppInlineScript, Option<String>) -> error::Result<()>,
+>(
+    value: &Value,
+    // Set to None.
+    container_id: Option<String>,
+    cb: &mut C,
+) -> error::Result<()> {
+    match value {
+        Value::Object(object) => {
+            if let Some(Value::Object(script)) = object.get("inlineScript") {
+                let (language, lock, code) = (
+                    script
+                        .get("language")
+                        .cloned()
+                        .map(|v| from_value::<ScriptLang>(v).ok())
+                        .flatten(),
+                    script
+                        .get("lock")
+                        .and_then(Value::as_str)
+                        .map(str::to_owned),
+                    script
+                        .get("content")
+                        .and_then(Value::as_str)
+                        .map(str::to_owned)
+                        .ok_or(error::Error::internal_err(
+                            "Missing `content` in inlineScript".to_string(),
+                        ))?,
+                );
+                if language.is_some() {
+                    cb(
+                        AppInlineScript { language, content: code.to_owned(), lock },
+                        container_id.clone(),
+                    )?;
+                }
+            } else {
+                for (_, value) in object {
+                    traverse_app_inline_scripts(
+                        value,
+                        object
+                            .get("id")
+                            .and_then(Value::as_str)
+                            .map(str::to_owned)
+                            .or(container_id.clone()),
+                        cb,
+                    )?;
+                }
+            }
+        }
+        Value::Array(array) => {
+            for value in array {
+                traverse_app_inline_scripts(value, container_id.clone(), cb)?;
+            }
+        }
+        _ => {}
+    }
+    Ok(())
+}
@@ -93,7 +93,7 @@ pub struct ListableFlow {
    pub deployment_msg: Option<String>,
 }

-#[derive(Debug, Deserialize, sqlx::FromRow)]
+#[derive(Debug, Serialize, Deserialize, sqlx::FromRow)]
 pub struct NewFlow {
    pub path: String,
    pub summary: String,
@@ -158,6 +158,34 @@ impl FlowValue {

        flow_module
    }
+
+    /// Traverse FlowValue while invoking provided by caller callback on leafs
+    // #[async_recursion::async_recursion(?Send)]
+    // TODO: We may be want this async.
+    pub fn traverse_leafs<C: FnMut(&FlowModuleValue, &String) -> crate::error::Result<()>>(
+        modules: &Vec<FlowModule>,
+        cb: &mut C,
+    ) -> crate::error::Result<()> {
+        use FlowModuleValue::*;
+        for module in modules {
+            match serde_json::from_str::<FlowModuleValue>(module.value.get())? {
+                s @ (Script { .. }
+                | RawScript { .. }
+                | Flow { .. }
+                | FlowScript { .. }
+                | Identity) => cb(&s, &module.id)?,
+                ForloopFlow { modules, .. }
+                | WhileloopFlow { modules, .. }
+                | AIAgent { tools: modules, .. } => Self::traverse_leafs(&modules, cb)?,
+                BranchOne { branches, .. } | BranchAll { branches, .. } => {
+                    for branch in branches {
+                        Self::traverse_leafs(&branch.modules, cb)?;
+                    }
+                }
+            }
+        }
+        Ok(())
+    }
 }

 #[derive(Debug, Copy, Clone)]
@@ -565,6 +593,7 @@ pub struct Branch {
    rename_all(serialize = "lowercase", deserialize = "lowercase")
 )]
 pub enum FlowModuleValue {
+    /// Reference to another script on the workspace
    Script {
        #[serde(default)]
        #[serde(alias = "input_transform")]
@@ -577,12 +606,16 @@ pub enum FlowModuleValue {
        #[serde(skip_serializing_if = "Option::is_none")]
        is_trigger: Option<bool>,
    },
+
+    /// Reference to another flow on the workspace
    Flow {
        #[serde(default)]
        #[serde(alias = "input_transform")]
        input_transforms: HashMap<String, InputTransform>,
        path: String,
    },
+
+    /// For loop node
    ForloopFlow {
        iterator: InputTransform,
        modules: Vec<FlowModule>,
@@ -594,6 +627,8 @@ pub enum FlowModuleValue {
        #[serde(skip_serializing_if = "Option::is_none")]
        parallelism: Option<u16>,
    },
+
+    /// While loop node
    WhileloopFlow {
        modules: Vec<FlowModule>,
        #[serde(skip_serializing_if = "Option::is_none")]
@@ -601,17 +636,24 @@ pub enum FlowModuleValue {
        #[serde(default = "default_false")]
        skip_failures: bool,
    },
+
+    /// Branch-one node
    BranchOne {
        branches: Vec<Branch>,
        default: Vec<FlowModule>,
        #[serde(skip_serializing_if = "Option::is_none")]
        default_node: Option<FlowNodeId>,
    },
+
+    /// Branch-all node
    BranchAll {
        branches: Vec<Branch>,
        #[serde(default = "default_true")]
        parallel: bool,
    },
+
+    /// Inline script node
+    /// Only exists if parsed from value from `flow_version` | `flow` table.
    RawScript {
        #[serde(default)]
        #[serde(alias = "input_transform", serialize_with = "ordered_map")]
@@ -635,8 +677,13 @@ pub enum FlowModuleValue {
        #[serde(skip_serializing_if = "Option::is_none")]
        assets: Option<Vec<AssetWithAltAccessType>>,
    },
+
+    /// Just a placeholder
    Identity,
-    // Internal only, never exposed to the frontend.
+
+    /// Also Inline script node, but instead of being baked into flow, it references `flow_node`
+    /// Internal only, never exposed to the frontend.
+    /// Only exists if parsed from value from `flow_version_lite` table.
    FlowScript {
        #[serde(default)]
        #[serde(alias = "input_transform", serialize_with = "ordered_map")]
@@ -656,6 +703,8 @@ pub enum FlowModuleValue {
        #[serde(skip_serializing_if = "Option::is_none")]
        assets: Option<Vec<AssetWithAltAccessType>>,
    },
+
+    // AI agent node
    AIAgent {
        input_transforms: HashMap<String, InputTransform>,
        tools: Vec<FlowModule>,
@@ -137,6 +137,12 @@ impl Into<u64> for ScriptHash {
    }
 }

+impl From<i64> for ScriptHash {
+    fn from(value: i64) -> Self {
+        Self(value)
+    }
+}
+
 #[derive(PartialEq, sqlx::Type)]
 #[sqlx(transparent, no_pg_array)]
 pub struct ScriptHashes(pub Vec<i64>);
@@ -160,7 +166,10 @@ impl<'de> Deserialize<'de> for ScriptHash {
        D: Deserializer<'de>,
    {
        let s = String::deserialize(deserializer)?;
-        let i = to_i64(&s).map_err(|e| D::Error::custom(format!("{}", e)))?;
+        let i = to_i64(&s).map_err(|e| {
+            tracing::error!("Could not deserialize ScriptHash. Note, input should be in Hex and digit amount should be divisible by 16 (can be padded). err: {}", &e);
+            D::Error::custom(format!("{}", e))
+        })?;
        Ok(ScriptHash(i))
    }
 }
@@ -329,7 +338,7 @@ impl Hash for Schema {
    }
 }

-#[derive(Serialize, Deserialize, Hash)]
+#[derive(Serialize, Deserialize, Hash, Debug)]
 pub struct NewScript {
    pub path: String,
    pub parent_hash: Option<ScriptHash>,
@@ -57,6 +57,7 @@ pub mod result_processor;
 mod rust_executor;
 mod sanitized_sql_params;
 mod schema;
+pub mod scoped_dependency_map;
 mod universal_pkg_installer;
 mod worker;
 mod worker_flow;
@@ -0,0 +1,445 @@
+use serde::Serialize;
+use tokio::sync::RwLock;
+use windmill_common::{
+    apps::traverse_app_inline_scripts,
+    cache,
+    error::{Error, Result},
+    flows::{FlowModuleValue, FlowValue},
+    scripts::ScriptLang,
+};
+
+use std::collections::HashSet;
+
+use crate::worker_lockfiles::{extract_relative_imports, LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT};
+
+// TODO: To be removed in future versions
+lazy_static::lazy_static! {
+    pub static ref WMDEBUG_NO_DMAP_DISSOLVE: bool = std::env::var("WMDEBUG_NO_DMAP_DISSOLVE").is_ok();
+}
+
+#[derive(Serialize)]
+pub struct DependencyMap {
+    pub workspace_id: Option<String>,
+    pub importer_path: Option<String>,
+    pub importer_kind: Option<String>,
+    pub imported_path: Option<String>,
+    pub importer_node_id: Option<String>,
+}
+
+#[derive(Debug)]
+pub struct ScopedDependencyMap {
+    dmap: HashSet<(String, String)>,
+    w_id: String,
+    importer_path: String,
+    importer_kind: String,
+}
+
+impl ScopedDependencyMap {
+    /// Calls DB, however is assumed to be called once per dependency job
+    /// AND is scoped to smaller subset of data
+    /// So it is not too expensive
+    pub(crate) async fn fetch_maybe_rearranged<'a>(
+        w_id: &str,
+        importer_path: &str,
+        importer_kind: &str,
+        parent_path: &Option<String>,
+        executor: impl sqlx::Executor<'a, Database = sqlx::Postgres>,
+    ) -> Result<Self> {
+        if parent_path
+            .as_ref()
+            .is_some_and(|x| !x.is_empty() && x != importer_path)
+        {
+            tracing::info!(
+                workspace_id = %w_id,
+                "detected top level rename from: {} to: {importer_path} on object of kind: {importer_kind}. reflecting in dependency_map.",
+                parent_path.clone().unwrap_or_default(),
+            );
+
+            let dmap = sqlx::query_as::<_, (String, String)>(
+                "
+UPDATE dependency_map
+    SET importer_path = $1
+    WHERE importer_path = $2
+        AND importer_kind = $3::text::IMPORTER_KIND
+        AND workspace_id = $4
+RETURNING importer_node_id, imported_path
+        ",
+            )
+            .bind(importer_path)
+            .bind(parent_path.clone().unwrap())
+            .bind(importer_kind)
+            .bind(w_id)
+            .fetch_all(executor)
+            .await?;
+            Ok(Self {
+                dmap: HashSet::from_iter(dmap.into_iter()),
+                w_id: w_id.to_owned(),
+                importer_path: importer_path.to_owned(),
+                importer_kind: importer_kind.to_owned(),
+            })
+        } else {
+            Self::fetch(w_id, importer_path, importer_kind, executor).await
+        }
+    }
+
+    /// Almost same as [[Self::fetch_maybe_rearranged]], however only reads values, thus a bit faster.
+    pub async fn fetch<'a>(
+        w_id: &str,
+        importer_path: &str,
+        importer_kind: &str,
+        executor: impl sqlx::Executor<'a, Database = sqlx::Postgres>,
+    ) -> Result<Self> {
+        let dmap = sqlx::query_as::<_, (String, String)>(
+            "
+SELECT importer_node_id, imported_path
+    FROM dependency_map
+    WHERE workspace_id = $1
+        AND importer_path = $2
+        AND importer_kind = $3::text::IMPORTER_KIND",
+        )
+        .bind(w_id)
+        .bind(importer_path)
+        .bind(importer_kind)
+        .fetch_all(executor)
+        .await?;
+
+        Ok(Self {
+            dmap: HashSet::from_iter(dmap.into_iter()),
+            w_id: w_id.to_owned(),
+            importer_path: importer_path.to_owned(),
+            importer_kind: importer_kind.to_owned(),
+        })
+    }
+
+    /// Add missing entries to `dependency_map`
+    /// Remove matching entries
+    pub(crate) async fn patch<'c>(
+        &mut self,
+        relative_imports: Option<Vec<String>>,
+        node_id: String, // Flow Step/Node ID
+        mut tx: sqlx::Transaction<'c, sqlx::Postgres>,
+    ) -> Result<sqlx::Transaction<'c, sqlx::Postgres>> {
+        self.patch_tx_ref(relative_imports, &node_id, &mut tx)
+            .await?;
+        Ok(tx)
+    }
+
+    pub(crate) async fn patch_tx_ref<'c>(
+        &mut self,
+        relative_imports: Option<Vec<String>>,
+        node_id: &str, // Flow Step/Node ID
+        tx: &mut sqlx::Transaction<'c, sqlx::Postgres>,
+    ) -> Result<()> {
+        let Some(mut relative_imports) = relative_imports else {
+            tracing::info!("relative imports are not found for: importer - {}, importer_node_id - {}, importer_kind - {}",
+                &self.importer_path,
+                &node_id,
+                &self.importer_kind,
+            );
+            return Ok(());
+        };
+
+        // This does:
+        // 1. remove all relative imports from relative_imports that ARE tracked in dependency_map
+        // 2. remove corresponding trackers from dependency_map
+        //
+        // After this operation `relative_imports` variable has only untracked imports.
+        // We will handle those in the next expression.
+        //
+        // After all `reduce`'s called ScopedDependencyMap has only extra/orphan imports
+        // these are going to be clean up by calling [dissolve]
+        // NOTE: `retain` iterates over vec and remove the ones whose closures returned false.
+        relative_imports.retain(|imported_path| {
+            !self
+                .dmap
+                // As dmap is HashSet, removing is O(1) operation
+                // thus making entire process very efficient
+                // NOTE: `remove` returns true if item was removed and false if wasn't.
+                .remove(&(node_id.to_owned(), imported_path.to_owned()))
+        });
+
+        // As mentioned above, usually this will always be empty.
+        if !relative_imports.is_empty() {
+            tracing::info!("adding missing entries to dependency_map: importer_node_id - {}, importer_kind - {}, new_imported_paths - {:?}",
+                &node_id,
+                &self.importer_kind,
+                &relative_imports,
+            );
+        }
+
+        for import in relative_imports {
+            sqlx::query!(
+                "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id)
+                     VALUES ($1, $2, $3::text::IMPORTER_KIND, $4, $5) ON CONFLICT DO NOTHING",
+                &self.w_id,
+                &self.importer_path,
+                &self.importer_kind,
+                import,
+                node_id
+            )
+            .execute(&mut **tx)
+            .await?;
+
+            tracing::info!("added entry to dependency_map: {import:?}");
+        }
+        Ok(())
+    }
+
+    /// clean orphan entries from `dependency_map`
+    pub(crate) async fn dissolve<'a>(
+        self,
+        mut tx: sqlx::Transaction<'a, sqlx::Postgres>,
+    ) -> sqlx::Transaction<'a, sqlx::Postgres> {
+        if *WMDEBUG_NO_DMAP_DISSOLVE {
+            tracing::warn!(
+                "WMDEBUG_NO_DMAP_DISSOLVE usually should not be used. Behavior might be unstable."
+            );
+            return tx;
+        }
+
+        tracing::info!("dissolving dependency_map: {:?}", &self);
+
+        // We _could_ shove it into single query, but this query is rarely called AND let's keep it simple for redability.
+        for (importer_node_id, imported_path) in self.dmap.into_iter() {
+            tracing::info!("cleaning orphan entry from dependency_map: importer_kind - {}, imported_path - {}, importer_node_id - {}",
+                &self.importer_kind,
+                &imported_path,
+                &importer_node_id,
+            );
+
+            // Dissolve MUST succeed. Error in dissolve MUST not block the execution.
+            if let Err(err) = sqlx::query!(
+                "
+        DELETE FROM dependency_map
+        WHERE workspace_id = $1
+            AND importer_path = $2
+            AND importer_kind = $3::text::IMPORTER_KIND
+            AND importer_node_id = $4
+            AND imported_path = $5
+            ",
+                &self.w_id,
+                &self.importer_path,
+                &self.importer_kind,
+                &importer_node_id,
+                &imported_path,
+            )
+            .execute(&mut *tx)
+            .await
+            {
+                tracing::error!(
+                    "error while cleaning dependency_map for: importer_node_id - {}, imported_path - {}, importer_path - {}: {err}",
+                    importer_node_id,
+                    imported_path,
+                    self.importer_path,
+                );
+            }
+        }
+        tx
+    }
+
+    /// Selectively clean dependency_map for object
+    /// If `importer_node_id` is None will clear all nodes.
+    pub(crate) async fn clear_map_for_item<'c>(
+        item_path: &str,
+        w_id: &str,
+        importer_kind: &str,
+        mut tx: sqlx::Transaction<'c, sqlx::Postgres>,
+        importer_node_id: &Option<String>,
+    ) -> sqlx::Transaction<'c, sqlx::Postgres> {
+        tracing::warn!(
+            importer = item_path,
+            kind = importer_kind,
+            node_id = importer_node_id,
+            workspace_id = w_id,
+            "discovered orphan entry in `dependency_map`. It will be healed automatically, however please report this issue to Windmill Team. It is also advised to rebuild maps in workspace settings in troubleshooting.",
+        );
+
+        // MUST succeed. Error MUST not block the execution.
+        if let Err(err) = sqlx::query!(
+            "DELETE FROM dependency_map
+                 WHERE importer_path = $1 AND importer_kind = $3::text::IMPORTER_KIND
+                 AND workspace_id = $2 AND ($4::text IS NULL OR importer_node_id = $4::text)",
+            item_path,
+            w_id,
+            importer_kind,
+            importer_node_id.clone(),
+        )
+        .execute(&mut *tx)
+        .await
+        {
+            tracing::error!(
+                workspace_id = w_id,
+                "error while clearing discovered orphan: {err}"
+            );
+        }
+        tx
+    }
+
+    /// Run if you want to rebuild maps on specific workspace.
+    /// Potentially takes much time
+    pub async fn rebuild_map(w_id: &str, db: &sqlx::Pool<sqlx::Postgres>) -> Result<String> {
+        async fn inner<'c>(w_id: &str, db: &sqlx::Pool<sqlx::Postgres>) -> Result<String> {
+            // Scripts
+            tracing::info!(workspace_id = w_id, "Rebuilding dependency map for scripts");
+            for r in sqlx::query!(
+                "SELECT path, hash FROM script WHERE workspace_id = $1 AND archived = false",
+                w_id
+            )
+            .fetch_all(db)
+            .await?
+            {
+                let (sd, smd) = cache::script::fetch(&db.clone().into(), r.hash.into()).await?;
+                let mut dmap = ScopedDependencyMap::fetch(w_id, &r.path, "script", db).await?;
+                let mut tx = db.begin().await?;
+
+                if (smd.language.is_some_and(|v| v == ScriptLang::Bun)
+                    && sd
+                        .lock
+                        .as_ref()
+                        .is_some_and(|v| v.contains("generatedFromPackageJson")))
+                    || (smd.language.is_some_and(|v| v == ScriptLang::Python3)
+                        && sd.lock.as_ref().is_some_and(|v| {
+                            v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT)
+                        }))
+                {
+                    // if the lock file is generated from a package.json/requirements.txt, we need to clear the dependency map
+                    // because we do not want to have dependencies be recomputed automatically. Empty relative imports passed
+                    // to update_script_dependency_map will clear the dependency map.
+                } else {
+                    tx = dmap
+                        .patch(
+                            extract_relative_imports(&sd.code, &r.path, &smd.language),
+                            "".into(),
+                            tx,
+                        )
+                        .await?;
+                }
+                if !*WMDEBUG_NO_DMAP_DISSOLVE {
+                    dmap.dissolve(tx).await.commit().await?;
+                }
+                tracing::info!(workspace_id = w_id, "Rebuilt for script {}", &r.path);
+            }
+
+            // Fetch only top level versions and paths
+            // It is not fetching value
+            tracing::info!(workspace_id = w_id, "Rebuilding dependency map for flows");
+            for r in sqlx::query!("SELECT path, versions[array_upper(versions, 1)] as version FROM flow WHERE workspace_id = $1", w_id).fetch_all(db).await? {
+                if let Some(version) = r.version {
+                    // To reduce stress on db try to fetch from cache
+                    // Since our flow versions are immutable it is safe to assume if we have cache for specific version/id it is up to date.
+                    let flow_data = cache::flow::fetch_version(&db.clone().into(), version).await?;
+
+                    // Create map for specific flow
+                    let mut dmap = ScopedDependencyMap::fetch(w_id, &r.path, "flow", db).await?;
+
+                    // Traverse retrieved flow modules
+                    let mut tx = db.begin().await?;
+                    let mut to_process = vec![];
+                    FlowValue::traverse_leafs(&flow_data.flow.modules, &mut |fmv, id| {
+                        match fmv {
+                            // Since we fetched from flow_version it is safe to assume all inline scripts are in form of RawScript.
+                            FlowModuleValue::RawScript { content, language, .. } => {
+                                to_process.push((
+                                    extract_relative_imports(
+                                        content,
+                                        &(r.path.clone() + "/flow"),
+                                        &Some(language.clone()),
+                                    ),
+                                    id.clone(),
+                                ));
+                            }
+                            // But just in case we will also handle other cases.
+                            FlowModuleValue::FlowScript { .. } => {
+                                // Abort will cancel transaction.
+                                return Err(Error::internal_err("FlowScript is not supposed to be in flow."));
+                            }
+                            _ => {}
+                        }
+                        Ok(())
+                    })?;
+
+                    for (ri, id) in to_process {
+                        tx = dmap.patch(ri, id, tx).await?;
+                    }
+
+                    if !*WMDEBUG_NO_DMAP_DISSOLVE {
+                        dmap.dissolve(tx).await.commit().await?;
+                    }
+
+                    tracing::info!(workspace_id = w_id, "Rebuilt for flow {}", &r.path);
+                } else {
+                    tracing::error!(workspace_id = w_id, "version is never supposed to be none. skipping flow.");
+                    return Err(Error::internal_err("version was none"));
+                }
+            }
+
+            // Apps
+            tracing::info!(workspace_id = w_id, "Rebuilding dependency map for apps");
+            for r in sqlx::query!("SELECT path, versions[array_upper(versions, 1)] as version FROM app WHERE workspace_id = $1", w_id).fetch_all(db).await? {
+                if let Some(version) = r.version {
+                    // TODO: Use cache when implemented.
+                    let value = sqlx::query_scalar!(
+                        "SELECT value FROM app_version WHERE id = $1 LIMIT 1",
+                        version
+                    )
+                    .fetch_one(db)
+                    .await?;
+
+                    let mut dmap = ScopedDependencyMap::fetch(w_id, &r.path, "app", db).await?;
+                    let mut tx = db.begin().await?;
+                    let mut to_process = vec![];
+                    traverse_app_inline_scripts(&value, None, &mut |ais, id| {
+                        to_process.push((
+                            extract_relative_imports(
+                                &ais.content,
+                                &(r.path.clone() + "/app"),
+                                &ais.language,
+                            ),
+                            id,
+                        ));
+
+                        Ok(())
+                    })?;
+                    for (ri, id) in to_process {
+                        tx = dmap.patch(ri, id.unwrap_or_default(), tx).await?;
+                    }
+                    if !*WMDEBUG_NO_DMAP_DISSOLVE {
+                        dmap.dissolve(tx).await.commit().await?;
+                    }
+                    tracing::info!(workspace_id = w_id, "Rebuilt for app {}", &r.path);
+                } else {
+                    tracing::error!(
+                        workspace_id = w_id,
+                        "version is never supposed to be none. skipping app."
+                    );
+                    return Err(Error::internal_err("version was none"));
+                }
+            }
+
+            Ok("Success".into())
+        }
+
+        lazy_static::lazy_static! {
+            pub static ref LOCKED: RwLock<bool> = RwLock::new(false);
+        }
+
+        if *LOCKED.read().await {
+            tracing::warn!(
+                workspace_id = w_id,
+                "Tried to rebuild dependency map. However rebuild is already in progress."
+            );
+            Ok("There is already one task pending, try again later.".into())
+        } else {
+            tracing::info!(workspace_id = w_id, "Rebuilding dependency map");
+
+            if *WMDEBUG_NO_DMAP_DISSOLVE {
+                tracing::warn!("WMDEBUG_NO_DMAP_DISSOLVE usually should not be used. Behavior might be unstable. Please contact Windmill Team for support.")
+            }
+
+            *LOCKED.write().await = true;
+            let r = inner(w_id, db).await;
+            *LOCKED.write().await = false;
+            r
+        }
+    }
+}
@@ -5,6 +5,7 @@ use std::path::{Component, Path, PathBuf};

 #[cfg(feature = "python")]
 use crate::ansible_executor::{get_git_repos_lock, AnsibleDependencyLocks};
+use crate::scoped_dependency_map::{ScopedDependencyMap, WMDEBUG_NO_DMAP_DISSOLVE};
 use async_recursion::async_recursion;
 use itertools::Itertools;
 use serde_json::value::RawValue;
@@ -69,115 +70,6 @@ use crate::{
    go_executor::install_go_dependencies,
 };

-pub async fn update_script_dependency_map(
-    job_id: &Uuid,
-    db: &DB,
-    w_id: &str,
-    parent_path: &Option<String>,
-    script_path: &str,
-    relative_imports: Vec<String>,
-) -> error::Result<()> {
-    let importer_kind = "script";
-
-    let mut tx = db.begin().await?;
-    tx = clear_dependency_parent_path(parent_path, script_path, w_id, importer_kind, tx).await?;
-
-    tx = clear_dependency_map_for_item(script_path, w_id, importer_kind, tx, &None).await?;
-
-    if !relative_imports.is_empty() {
-        let mut logs = "".to_string();
-        logs.push_str("\n--- RELATIVE IMPORTS ---\n\n");
-        logs.push_str(&relative_imports.join("\n"));
-
-        tx = add_relative_imports_to_dependency_map(
-            script_path,
-            w_id,
-            relative_imports,
-            importer_kind,
-            tx,
-            &mut logs,
-            None,
-        )
-        .await?;
-        append_logs(job_id, w_id, logs, &db.into()).await;
-    }
-    tx.commit().await?;
-
-    Ok(())
-}
-
-async fn add_relative_imports_to_dependency_map<'c>(
-    script_path: &str,
-    w_id: &str,
-    relative_imports: Vec<String>,
-    importer_kind: &str,
-    mut tx: sqlx::Transaction<'c, sqlx::Postgres>,
-    logs: &mut String,
-    node_id: Option<String>,
-) -> error::Result<sqlx::Transaction<'c, sqlx::Postgres>> {
-    for import in relative_imports {
-        sqlx::query!(
-            "INSERT INTO dependency_map (workspace_id, importer_path, importer_kind, imported_path, importer_node_id)
-                 VALUES ($1, $2, $4::text::IMPORTER_KIND, $3, $5) ON CONFLICT DO NOTHING",
-            w_id,
-            script_path,
-            import,
-            importer_kind,
-            node_id.clone().unwrap_or_default()
-        )
-        .execute(&mut *tx)
-        .await?;
-        logs.push_str(&format!("{}\n", import));
-    }
-    Ok(tx)
-}
-
-async fn clear_dependency_map_for_item<'c>(
-    item_path: &str,
-    w_id: &str,
-    importer_kind: &str,
-    mut tx: sqlx::Transaction<'c, sqlx::Postgres>,
-    importer_node_id: &Option<String>,
-) -> Result<sqlx::Transaction<'c, sqlx::Postgres>> {
-    sqlx::query!(
-        "DELETE FROM dependency_map
-                 WHERE importer_path = $1 AND importer_kind = $3::text::IMPORTER_KIND
-                 AND workspace_id = $2 AND ($4::text IS NULL OR importer_node_id = $4::text)",
-        item_path,
-        w_id,
-        importer_kind,
-        importer_node_id.clone()
-    )
-    .execute(&mut *tx)
-    .await?;
-    Ok(tx)
-}
-
-async fn clear_dependency_parent_path<'c>(
-    parent_path: &Option<String>,
-    item_path: &str,
-    w_id: &str,
-    importer_kind: &str,
-    mut tx: sqlx::Transaction<'c, sqlx::Postgres>,
-) -> Result<sqlx::Transaction<'c, sqlx::Postgres>> {
-    if parent_path
-        .as_ref()
-        .is_some_and(|x| !x.is_empty() && x != item_path)
-    {
-        sqlx::query!(
-            "DELETE FROM dependency_map
-                 WHERE importer_path = $1 AND importer_kind = $3::text::IMPORTER_KIND
-                 AND workspace_id = $2",
-            parent_path.clone().unwrap(),
-            w_id,
-            importer_kind
-        )
-        .execute(&mut *tx)
-        .await?;
-    }
-    Ok(tx)
-}
-
 fn try_normalize(path: &Path) -> Option<PathBuf> {
    let mut ret = PathBuf::new();

@@ -237,6 +129,7 @@ pub fn extract_relative_imports(
        _ => None,
    }
 }
+
 #[tracing::instrument(level = "trace", skip_all)]
 pub async fn handle_dependency_job(
    job: &MiniPulledJob,
@@ -369,6 +262,7 @@ pub async fn handle_dependency_job(
            // where Second will **always** do with lock being not null
            let deployed_hash = if script_info.lock.is_some() && !*WMDEBUG_NO_HASH_CHANGE_ON_DJ {
                let path = script_info.path.clone();
+
                let mut tx = db.begin().await?;
                // This entire section exists to solve following problem:
                //
@@ -439,7 +333,12 @@ pub async fn handle_dependency_job(
    FROM script WHERE hash = $2 AND workspace_id = $3;
            ",
                new_hash, current_hash.0, w_id, &content).execute(db).await?;
-                tracing::info!("Updated script at path {} with hash {} to new hash {}", path, current_hash.0, new_hash);
+                tracing::info!(
+                    "Updated script at path {} with hash {} to new hash {}",
+                    path,
+                    current_hash.0,
+                    new_hash
+                );
                // Archive current
                sqlx::query!(
                    "UPDATE script SET archived = true WHERE hash = $1 AND workspace_id = $2",
@@ -448,7 +347,11 @@ pub async fn handle_dependency_job(
                )
                .execute(&mut *tx)
                .await?;
-                tracing::info!("Archived script at path {} from dependency job {}", path, current_hash.0);
+                tracing::info!(
+                    "Archived script at path {} from dependency job {}",
+                    path,
+                    current_hash.0
+                );
                tx.commit().await?;

                ScriptHash(new_hash)
@@ -549,7 +452,7 @@ fn remove_ansi_codes(s: &str) -> String {

 pub async fn process_relative_imports(
    db: &sqlx::Pool<sqlx::Postgres>,
-    job_id: Option<Uuid>,
+    _job_id: Option<Uuid>,
    args: Option<&Json<HashMap<String, Box<RawValue>>>>,
    w_id: &str,
    script_path: &str,
@@ -562,40 +465,50 @@ pub async fn process_relative_imports(
    permissioned_as: &str,
    lock: Option<String>,
 ) -> error::Result<()> {
-    let relative_imports = extract_relative_imports(&code, script_path, script_lang);
-    if let Some(relative_imports) = relative_imports {
-        if (script_lang.is_some_and(|v| v == ScriptLang::Bun)
-            && lock
-                .as_ref()
-                .is_some_and(|v| v.contains("generatedFromPackageJson")))
-            || (script_lang.is_some_and(|v| v == ScriptLang::Python3)
+    // TODO: Should be moved into handle_dependency_job body to be more consistent with how flows and apps are handled
+    {
+        let relative_imports = extract_relative_imports(&code, script_path, script_lang);
+        if let Some(relative_imports) = relative_imports {
+            let mut tx = db.begin().await?;
+            let mut dependency_map = ScopedDependencyMap::fetch_maybe_rearranged(
+                &w_id,
+                script_path,
+                "script",
+                &parent_path,
+                db,
+            )
+            .await?;
+            if (script_lang.is_some_and(|v| v == ScriptLang::Bun)
                && lock
                    .as_ref()
-                    .is_some_and(|v| v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT)))
-        {
-            // if the lock file is generated from a package.json/requirements.txt, we need to clear the dependency map
-            // because we do not want to have dependencies be recomputed automatically. Empty relative imports passed
-            // to update_script_dependency_map will clear the dependency map.
-            update_script_dependency_map(
-                &job_id.unwrap_or_else(|| Uuid::nil()),
-                db,
-                w_id,
-                &parent_path,
-                script_path,
-                vec![],
-            )
-            .await?;
-        } else {
-            update_script_dependency_map(
-                &job_id.unwrap_or_else(|| Uuid::nil()),
-                db,
-                w_id,
-                &parent_path,
-                script_path,
-                relative_imports,
-            )
-            .await?;
+                    .is_some_and(|v| v.contains("generatedFromPackageJson")))
+                || (script_lang.is_some_and(|v| v == ScriptLang::Python3)
+                    && lock
+                        .as_ref()
+                        .is_some_and(|v| v.starts_with(LOCKFILE_GENERATED_FROM_REQUIREMENTS_TXT)))
+            {
+                // if the lock file is generated from a package.json/requirements.txt, we need to clear the dependency map
+                // because we do not want to have dependencies be recomputed automatically. Empty relative imports passed
+                // to update_script_dependency_map will clear the dependency map.
+
+                // TODO: Rework the logic for synchronized raw requirements PR.
+                // For now we will just do nothing and let dissolve clear every item related to this script.
+            } else {
+                tx = dependency_map
+                    .patch(
+                        Some(relative_imports),
+                        // Ideally should be None, but due to current implementation will use empty string to represent None.
+                        "".into(),
+                        tx,
+                    )
+                    .await?;
+            }
+            // If felt into first branch which did not call .patch(, this operation will clean dependency_map for this script.
+            dependency_map.dissolve(tx).await.commit().await?;
        }
+    }
+
+    {
        let already_visited = args
            .map(|x| {
                x.get("already_visited")
@@ -604,6 +517,9 @@ pub async fn process_relative_imports(
            })
            .flatten()
            .unwrap_or_default();
+
+        // But currently we will do this extra db call for every script regardless of whether they have relative imports or not
+        // Script might have no relative imports but still be referenced by someone else.
        if let Err(e) = trigger_dependents_to_recompute_dependencies(
            w_id,
            script_path,
@@ -620,6 +536,7 @@ pub async fn process_relative_imports(
            tracing::error!(%e, "error triggering dependents to recompute dependencies");
        }
    }
+
    Ok(())
 }

@@ -710,9 +627,9 @@ pub async fn trigger_dependents_to_recompute_dependencies(
                "SELECT versions[array_upper(versions, 1)] FROM flow WHERE path = $1 AND workspace_id = $2",
                s.importer_path,
                w_id,
-            ).fetch_one(&mut *flow_tx)
+            ).fetch_optional(&mut *flow_tx)
            .await
-            .map_err(to_anyhow);
+            .map_err(to_anyhow).map(Option::flatten);

            match r {
                // TODO: Fallback - remove eventually.
@@ -770,11 +687,22 @@ pub async fn trigger_dependents_to_recompute_dependencies(
                    }
                }
                Ok(None) => {
-                    tracing::error!(
-                        "no flow version found for path {path}",
-                        path = s.importer_path
-                    );
-                    // Do not commit the transaction. It will be dropped and rollbacked
+                    if *WMDEBUG_NO_DMAP_DISSOLVE {
+                        tracing::warn!("WMDEBUG_NO_DMAP_DISSOLVE usually should not be used. Behavior might be unstable.");
+                    } else {
+                        // Remember the path we used to query the flow was fetched just now from dependency_map
+                        // if dependency_map advertise unexistent path, as part of self-healing it should be removed
+                        ScopedDependencyMap::clear_map_for_item(
+                            &s.importer_path,
+                            w_id,
+                            "flow",
+                            flow_tx,
+                            &None,
+                        )
+                        .await
+                        .commit()
+                        .await?;
+                    }
                    continue;
                }
                Err(err) => {
@@ -792,6 +720,7 @@ pub async fn trigger_dependents_to_recompute_dependencies(

            args.insert(
                "components_to_relock".to_string(),
+                // TODO: unsafe. Importer Node Ids are not checked. They can simply be array of empty strings!
                to_raw_value(&s.importer_node_ids),
            );

@@ -799,9 +728,9 @@ pub async fn trigger_dependents_to_recompute_dependencies(
                "SELECT versions[array_upper(versions, 1)] FROM app WHERE path = $1 AND workspace_id = $2",
                s.importer_path,
                w_id,
-            ).fetch_one(&mut *tx)
+            ).fetch_optional(&mut *tx)
            .await
-            .map_err(to_anyhow);
+            .map_err(to_anyhow).map(Option::flatten);

            match r {
                // Get current version of current flow.
@@ -842,11 +771,22 @@ pub async fn trigger_dependents_to_recompute_dependencies(
                    }
                }
                Ok(None) => {
-                    tracing::error!(
-                        "no app version found for path {path}",
-                        path = s.importer_path
-                    );
-                    // Do not commit the transaction. It will be dropped and rollbacked
+                    if *WMDEBUG_NO_DMAP_DISSOLVE {
+                        tracing::warn!("WMDEBUG_NO_DMAP_DISSOLVE usually should not be used. Behavior might be unstable.");
+                    } else {
+                        // Remember the path we used to query the flow was fetched just now from dependency_map
+                        // if dependency_map advertise unexistent path, as part of self-healing it should be removed
+                        ScopedDependencyMap::clear_map_for_item(
+                            &s.importer_path,
+                            w_id,
+                            "app",
+                            tx,
+                            &None,
+                        )
+                        .await
+                        .commit()
+                        .await?;
+                    }
                    continue;
                }
                Err(err) => {
@@ -992,8 +932,15 @@ pub async fn handle_flow_dependency_job(
    .clone();

    let mut tx = db.begin().await?;
-    tx = clear_dependency_parent_path(&parent_path, &job_path, &job.workspace_id, "flow", tx)
-        .await?;
+
+    let mut dependency_map = ScopedDependencyMap::fetch_maybe_rearranged(
+        &job.workspace_id,
+        &job_path,
+        "flow",
+        &parent_path,
+        &mut *tx,
+    )
+    .await?;

    if !skip_flow_update {
        sqlx::query!(
@@ -1026,6 +973,7 @@ pub async fn handle_flow_dependency_job(
        occupancy_metrics,
        skip_flow_update,
        raw_deps,
+        &mut dependency_map,
    )
    .await?;

@@ -1083,6 +1031,7 @@ pub async fn handle_flow_dependency_job(
        tracing::error!(%job.id, %err, "error checking cancellation for job {0}: {err}", job.id);
        false
    }) {
+        // Drop tx and thus cancel any changes
        return Ok(to_raw_value_owned(json!({
            "status": "Flow lock generation was canceled",
        })));
@@ -1093,6 +1042,8 @@ pub async fn handle_flow_dependency_job(
            Error::internal_err("Flow Dependency requires script hash (flow version)".to_owned())
        })?;

+        tx = dependency_map.dissolve(tx).await;
+
        sqlx::query!(
            "UPDATE flow SET value = $1 WHERE path = $2 AND workspace_id = $3",
            &new_flow_value as &Json<Box<RawValue>>,
@@ -1111,6 +1062,7 @@ pub async fn handle_flow_dependency_job(

        // Compute a lite version of the flow value (`RawScript` => `FlowScript`).
        let mut value_lite = flow.clone();
+
        tx = reduce_flow(
            tx,
            &mut value_lite.modules,
@@ -1197,6 +1149,8 @@ struct LockModuleError {
    error: Error,
 }

+// TODO: Maybe use [FlowValue::traverse_leafs]
+// IMPORTANT: If updating this function, make sure you also update [FlowValue::traverse_leafs]
 async fn lock_modules<'c>(
    modules: Vec<FlowModule>,
    job: &MiniPulledJob,
@@ -1214,7 +1168,7 @@ async fn lock_modules<'c>(
    occupancy_metrics: &mut OccupancyMetrics,
    skip_flow_update: bool,
    raw_deps: Option<HashMap<String, String>>,
-    // (modules to replace old seq (even unmmodified ones), new transaction, modified ids) )
+    dependency_map: &mut ScopedDependencyMap, // (modules to replace old seq (even unmmodified ones), new transaction, modified ids) )
 ) -> Result<(
    Vec<FlowModule>,
    sqlx::Transaction<'c, sqlx::Postgres>,
@@ -1268,6 +1222,7 @@ async fn lock_modules<'c>(
                        occupancy_metrics,
                        skip_flow_update,
                        raw_deps.clone(),
+                        dependency_map,
                    ))
                    .await?;
                    e.value = FlowModuleValue::ForloopFlow {
@@ -1304,6 +1259,7 @@ async fn lock_modules<'c>(
                            occupancy_metrics,
                            skip_flow_update,
                            raw_deps.clone(),
+                            dependency_map,
                        ))
                        .await?;
                        nmodified_ids.extend(inner_modified_ids);
@@ -1332,6 +1288,7 @@ async fn lock_modules<'c>(
                        occupancy_metrics,
                        skip_flow_update,
                        raw_deps.clone(),
+                        dependency_map,
                    ))
                    .await?;
                    e.value = FlowModuleValue::WhileloopFlow {
@@ -1365,6 +1322,7 @@ async fn lock_modules<'c>(
                            occupancy_metrics,
                            skip_flow_update,
                            raw_deps.clone(),
+                            dependency_map,
                        ))
                        .await?;
                        nmodified_ids.extend(inner_modified_ids);
@@ -1391,6 +1349,7 @@ async fn lock_modules<'c>(
                        occupancy_metrics,
                        skip_flow_update,
                        raw_deps.clone(),
+                        dependency_map,
                    ))
                    .await?;
                    errors.extend(ninner_errors);
@@ -1442,8 +1401,19 @@ async fn lock_modules<'c>(
            .await?;
        }

+        let dep_path = path.clone().unwrap_or_else(|| job_path.to_string());
+        let relative_imports = extract_relative_imports(
+            &content,
+            &format!("{dep_path}/flow"),
+            &Some(language.clone()),
+        );
+
        if let Some(locks_to_reload) = locks_to_reload {
            if !locks_to_reload.contains(&e.id) {
+                tx = dependency_map
+                    .patch(relative_imports.clone(), e.id.clone(), tx)
+                    .await?;
+
                new_flow_modules.push(e);
                continue;
            }
@@ -1451,6 +1421,10 @@ async fn lock_modules<'c>(
            if lock.as_ref().is_some_and(|x| !x.trim().is_empty()) {
                let skip_creating_new_lock = skip_creating_new_lock(&language, &content);
                if skip_creating_new_lock {
+                    tx = dependency_map
+                        .patch(relative_imports.clone(), e.id.clone(), tx)
+                        .await?;
+
                    new_flow_modules.push(e);
                    continue;
                }
@@ -1498,36 +1472,9 @@ async fn lock_modules<'c>(
        //
        let lock = match new_lock {
            Ok(new_lock) => {
-                let dep_path = path.clone().unwrap_or_else(|| job_path.to_string());
-                tx = clear_dependency_map_for_item(
-                    &job_path,
-                    &job.workspace_id,
-                    "flow",
-                    tx,
-                    &Some(e.id.clone()),
-                )
-                .await?;
-                let relative_imports = extract_relative_imports(
-                    &content,
-                    &format!("{dep_path}/flow"),
-                    &Some(language.clone()),
-                );
-                if let Some(relative_imports) = relative_imports {
-                    let mut logs = "".to_string();
-                    logs.push_str(format!("\n\n--- RELATIVE IMPORTS of {} ---\n\n", e.id).as_str());
-
-                    tx = add_relative_imports_to_dependency_map(
-                        &dep_path,
-                        &job.workspace_id,
-                        relative_imports,
-                        "flow",
-                        tx,
-                        &mut logs,
-                        Some(e.id.clone()),
-                    )
+                tx = dependency_map
+                    .patch(relative_imports.clone(), e.id.clone(), tx)
                    .await?;
-                    append_logs(&job.id, &job.workspace_id, logs, &db.into()).await;
-                }

                if language == ScriptLang::Bun || language == ScriptLang::Bunnative {
                    let anns = windmill_common::worker::TypeScriptAnnotations::parse(&content);
@@ -1638,7 +1585,6 @@ async fn insert_flow_node<'c>(
    Ok((tx, FlowNodeId(id)))
 }

-// TODO: Clean up dependency map when moved/renamed?
 async fn insert_app_script(
    db: &sqlx::Pool<sqlx::Postgres>,
    path: &str,
@@ -1771,6 +1717,7 @@ async fn reduce_flow<'c>(
                    Some(language),
                )
                .await?;
+
                val = FlowScript {
                    input_transforms,
                    id,
@@ -1906,6 +1853,10 @@ fn skip_creating_new_lock(language: &ScriptLang, content: &str) -> bool {
    true
 }

+// TODO: Use transaction?
+// TODO: Use abstracted traverse function.
+//
+// IMPORTANT: If updating this function, make sure you also update [traverse_app_inline_scripts]
 #[async_recursion]
 async fn lock_modules_app(
    value: Value,
@@ -1923,6 +1874,7 @@ async fn lock_modules_app(
    locks_to_reload: &Option<Vec<String>>,
    // Represents the closest container id
    container_id: Option<String>,
+    dependency_map: &mut ScopedDependencyMap,
 ) -> Result<Value> {
    match value {
        Value::Object(mut m) => {
@@ -1957,6 +1909,12 @@ async fn lock_modules_app(
                                .to_string();
                            let mut logs = "".to_string();

+                            let relative_imports = extract_relative_imports(
+                                &content,
+                                &format!("{job_path}/app"),
+                                &Some(language.clone()),
+                            );
+
                            if let Some((l, id)) = locks_to_reload
                                .as_ref()
                                .zip(container_id.as_ref())
@@ -1970,6 +1928,15 @@ async fn lock_modules_app(
                                })
                            {
                                if !l.contains(id) {
+                                    dependency_map
+                                        .patch(
+                                            relative_imports.clone(),
+                                            container_id.unwrap_or_default(),
+                                            db.begin().await?,
+                                        )
+                                        .await?
+                                        .commit()
+                                        .await?;
                                    return Ok(Value::Object(m.clone()));
                                }
                            } else if v
@@ -1977,6 +1944,16 @@ async fn lock_modules_app(
                                .is_some_and(|x| !x.as_str().unwrap().trim().is_empty())
                            {
                                if skip_creating_new_lock(&language, &content) {
+                                    dependency_map
+                                        .patch(
+                                            relative_imports.clone(),
+                                            container_id.unwrap_or_default(),
+                                            db.begin().await?,
+                                        )
+                                        .await?
+                                        .commit()
+                                        .await?;
+
                                    logs.push_str(
                                        "Found already locked inline script. Skipping lock...\n",
                                    );
@@ -2007,44 +1984,15 @@ async fn lock_modules_app(
                                Ok(new_lock) => {
                                    append_logs(&job.id, &job.workspace_id, logs, &db.into()).await;

-                                    let mut tx = db.begin().await?;
-
-                                    tx = clear_dependency_map_for_item(
-                                        &job_path,
-                                        &job.workspace_id,
-                                        "app",
-                                        tx,
-                                        &container_id,
-                                    )
-                                    .await?;
-
-                                    let relative_imports = extract_relative_imports(
-                                        &content,
-                                        &format!("{job_path}/app"),
-                                        &Some(language.clone()),
-                                    );
-
-                                    if let Some(relative_imports) = relative_imports {
-                                        let mut logs = "".to_string();
-                                        logs.push_str(
-                                            format!("\n\n--- RELATIVE IMPORTS ---\n\n").as_str(),
-                                        );
-
-                                        tx = add_relative_imports_to_dependency_map(
-                                            &job_path,
-                                            &job.workspace_id,
-                                            relative_imports,
-                                            "app",
-                                            tx,
-                                            &mut logs,
-                                            container_id,
+                                    dependency_map
+                                        .patch(
+                                            relative_imports.clone(),
+                                            container_id.unwrap_or_default(),
+                                            db.begin().await?,
                                        )
+                                        .await?
+                                        .commit()
                                        .await?;
-                                        append_logs(&job.id, &job.workspace_id, logs, &db.into())
-                                            .await;
-                                    }
-
-                                    tx.commit().await?;

                                    let anns =
                                        windmill_common::worker::TypeScriptAnnotations::parse(
@@ -2104,6 +2052,7 @@ async fn lock_modules_app(
                            .and_then(Value::as_str)
                            .map(str::to_owned)
                            .or(container_id.clone()),
+                        dependency_map,
                    )
                    .await?,
                );
@@ -2129,6 +2078,7 @@ async fn lock_modules_app(
                        occupancy_metrics,
                        locks_to_reload,
                        container_id.clone(),
+                        dependency_map,
                    )
                    .await?,
                );
@@ -2192,6 +2142,17 @@ pub async fn handle_app_dependency_job(
        .await?
        .map(|record| (record.app_id, record.value));

+    let (_, parent_path) = get_deployment_msg_and_parent_path_from_args(job.args.clone());
+
+    let mut dependency_map = ScopedDependencyMap::fetch_maybe_rearranged(
+        &job.workspace_id,
+        &job_path,
+        "app",
+        &parent_path,
+        db,
+    )
+    .await?;
+
    // TODO: Use transaction for entire segment?
    if let Some((app_id, value)) = record {
        let value = lock_modules_app(
@@ -2209,9 +2170,17 @@ pub async fn handle_app_dependency_job(
            occupancy_metrics,
            &components_to_relock,
            None,
+            &mut dependency_map,
        )
        .await?;

+        // TODO: Dissolve in the end?
+        dependency_map
+            .dissolve(db.begin().await?)
+            .await
+            .commit()
+            .await?;
+
        // Compute a lite version of the app value (w/ `inlineScript.{lock,code}`).
        let mut value_lite = value.clone();
        reduce_app(db, &job_path, &mut value_lite, app_id).await?;