diff --git a/backend/.sqlx/query-1438e8dc5738fc69bc6601eb11729610f671b7df0ab25da058e16c6654279d61.json b/backend/.sqlx/query-1438e8dc5738fc69bc6601eb11729610f671b7df0ab25da058e16c6654279d61.json
new file mode 100644
index 0000000000..452510a97f
--- /dev/null
+++ b/backend/.sqlx/query-1438e8dc5738fc69bc6601eb11729610f671b7df0ab25da058e16c6654279d61.json
@@ -0,0 +1,24 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        SELECT flow_status->'user_states'->$1\n        FROM queue\n        WHERE id = $2 AND workspace_id = $3\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Jsonb"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "1438e8dc5738fc69bc6601eb11729610f671b7df0ab25da058e16c6654279d61"
+}
diff --git a/backend/.sqlx/query-a21844a8a73420a1f25dac7815e525a27eda3ca6c69d690e8aa437b425c90963.json b/backend/.sqlx/query-a21844a8a73420a1f25dac7815e525a27eda3ca6c69d690e8aa437b425c90963.json
new file mode 100644
index 0000000000..f43754ea67
--- /dev/null
+++ b/backend/.sqlx/query-a21844a8a73420a1f25dac7815e525a27eda3ca6c69d690e8aa437b425c90963.json
@@ -0,0 +1,25 @@
+{
+  "db_name": "PostgreSQL",
+  "query": "\n        UPDATE queue SET flow_status = JSONB_SET(flow_status,  ARRAY['user_states'], JSONB_SET(COALESCE(flow_status->'user_states', '{}'::jsonb), ARRAY[$1], $2))\n        WHERE id = $3 AND workspace_id = $4 AND job_kind IN ('flow', 'flowpreview') RETURNING 1\n        ",
+  "describe": {
+    "columns": [
+      {
+        "ordinal": 0,
+        "name": "?column?",
+        "type_info": "Int4"
+      }
+    ],
+    "parameters": {
+      "Left": [
+        "Text",
+        "Jsonb",
+        "Uuid",
+        "Text"
+      ]
+    },
+    "nullable": [
+      null
+    ]
+  },
+  "hash": "a21844a8a73420a1f25dac7815e525a27eda3ca6c69d690e8aa437b425c90963"
+}
diff --git a/backend/src/main.rs b/backend/src/main.rs
index 0380c3c2f6..f22473b3c0 100644
--- a/backend/src/main.rs
+++ b/backend/src/main.rs
@@ -304,7 +304,7 @@ Windmill Community Edition {GIT_VERSION}
             default_base_internal_url.clone()
         };
 
-        initial_load(&db, killpill_tx.clone(), worker_mode, server_mode).await;
+        initial_load(&db, killpill_tx.clone(), worker_mode, server_mode, is_agent).await;
 
         monitor_db(&db, &base_internal_url, rsmq.clone(), server_mode, true).await;
 
@@ -446,7 +446,7 @@ Windmill Community Edition {GIT_VERSION}
                                                     reload_job_default_timeout_setting(&db).await
                                                 },
                                                 #[cfg(feature = "parquet")]
-                                                OBJECT_STORE_CACHE_CONFIG_SETTING => {
+                                                OBJECT_STORE_CACHE_CONFIG_SETTING if !is_agent => {
                                                     reload_s3_cache_setting(&db).await
                                                 },
                                                 SCIM_TOKEN_SETTING => {
diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs
index b01c2b523d..ab514b957e 100644
--- a/backend/src/monitor.rs
+++ b/backend/src/monitor.rs
@@ -105,6 +105,7 @@ pub async fn initial_load(
     tx: tokio::sync::broadcast::Sender<()>,
     worker_mode: bool,
     server_mode: bool,
+    _is_agent: bool,
 ) {
     if let Err(e) = load_metrics_enabled(db).await {
         tracing::error!("Error loading expose metrics: {e}");
@@ -136,7 +137,9 @@ pub async fn initial_load(
     }
 
     #[cfg(feature = "parquet")]
-    reload_s3_cache_setting(&db).await;
+    if !_is_agent {
+        reload_s3_cache_setting(&db).await;
+    }
 
     if server_mode {
         reload_server_config(&db).await;
diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml
index 4507d8b89e..30cbf9b50d 100644
--- a/backend/windmill-api/openapi.yaml
+++ b/backend/windmill-api/openapi.yaml
@@ -5820,6 +5820,53 @@ paths:
               schema:
                 type: string
 
+  /w/{workspace}/jobs/flow/user_states/{id}/{key}:
+    post:
+      summary: set flow user state at a given key
+      operationId: setFlowUserState
+      tags:
+        - job
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/JobId"
+        - name: key
+          in: path
+          required: true
+          schema:
+            type: string
+      requestBody:
+        description: new value
+        required: true
+        content:
+          application/json:
+            schema: {}
+      responses:
+        "200":
+          description: flow user state updated
+          content:
+            text/plain:
+              schema:
+                type: string
+    get:
+      summary: get flow user state at a given key
+      operationId: getFlowUserState
+      tags:
+        - job
+      parameters:
+        - $ref: "#/components/parameters/WorkspaceId"
+        - $ref: "#/components/parameters/JobId"
+        - name: key
+          in: path
+          required: true
+          schema:
+            type: string
+      responses:
+        "200":
+          description: flow user state updated
+          content:
+            application/json:
+              schema: {}
+
   /w/{workspace}/jobs/flow/resume/{id}:
     post:
       summary: resume a job for a suspended flow as an owner
diff --git a/backend/windmill-api/src/jobs.rs b/backend/windmill-api/src/jobs.rs
index f51cba7fe9..477107dd13 100644
--- a/backend/windmill-api/src/jobs.rs
+++ b/backend/windmill-api/src/jobs.rs
@@ -201,6 +201,12 @@ pub fn workspaced_service() -> Router {
             "/job_signature/:job_id/:resume_id",
             get(create_job_signature).layer(cors.clone()),
         )
+        .route(
+            "/flow/user_states/:job_id/:key",
+            get(get_flow_user_state)
+                .post(set_flow_user_state)
+                .layer(cors.clone()),
+        )
         .route(
             "/resume_urls/:job_id/:resume_id",
             get(get_resume_urls).layer(cors.clone()),
@@ -1675,6 +1681,55 @@ pub async fn create_job_signature(
     create_signature(key, job_id, resume_id, approver.approver)
 }
 
+pub async fn get_flow_user_state(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
+    Path((w_id, job_id, key)): Path<(String, Uuid, String)>,
+) -> error::JsonResult<Option<serde_json::Value>> {
+    let mut tx = user_db.begin(&authed).await?;
+    let r = sqlx::query_scalar!(
+        r#"
+        SELECT flow_status->'user_states'->$1
+        FROM queue
+        WHERE id = $2 AND workspace_id = $3
+        "#,
+        key,
+        job_id,
+        w_id
+    )
+    .fetch_optional(&mut *tx)
+    .await?
+    .flatten();
+    Ok(Json(r))
+}
+
+pub async fn set_flow_user_state(
+    authed: ApiAuthed,
+    Extension(user_db): Extension<UserDB>,
+    Path((w_id, job_id, key)): Path<(String, Uuid, String)>,
+    Json(value): Json<serde_json::Value>,
+) -> error::Result<String> {
+    let mut tx = user_db.begin(&authed).await?;
+    let r = sqlx::query_scalar!(
+        r#"
+        UPDATE queue SET flow_status = JSONB_SET(flow_status,  ARRAY['user_states'], JSONB_SET(COALESCE(flow_status->'user_states', '{}'::jsonb), ARRAY[$1], $2))
+        WHERE id = $3 AND workspace_id = $4 AND job_kind IN ('flow', 'flowpreview') RETURNING 1
+        "#,
+        key,
+        value,
+        job_id,
+        w_id
+    )
+    .fetch_optional(&mut *tx)
+    .await?
+    .flatten();
+    if r.is_none() {
+        return Err(Error::NotFound("Flow job not found".to_string()));
+    }
+    tx.commit().await?;
+    Ok("Flow job state updated".to_string())
+}
+
 fn create_signature(
     key: String,
     job_id: Uuid,
diff --git a/backend/windmill-common/src/flow_status.rs b/backend/windmill-common/src/flow_status.rs
index 473d363e2b..f73accb10b 100644
--- a/backend/windmill-common/src/flow_status.rs
+++ b/backend/windmill-common/src/flow_status.rs
@@ -6,6 +6,7 @@
  * LICENSE-AGPL for a copy of the license.
  */
 
+use std::collections::HashMap;
 use std::time::Duration;
 
 use serde::{Deserialize, Serialize};
@@ -29,6 +30,10 @@ pub struct FlowStatus {
     pub step: i32,
     pub modules: Vec<FlowStatusModule>,
     pub failure_module: FlowStatusModuleWParent,
+
+    #[serde(skip_serializing_if = "HashMap::is_empty")]
+    #[serde(default)]
+    pub user_states: HashMap<String, serde_json::Value>,
     #[serde(default)]
     pub cleanup_module: FlowCleanupModule,
     #[serde(default)]
@@ -231,6 +236,7 @@ impl FlowStatus {
             cleanup_module: FlowCleanupModule { flow_jobs_to_clean: vec![] },
             retry: RetryStatus { fail_count: 0, failed_jobs: vec![] },
             restarted_from: None,
+            user_states: HashMap::new(),
         }
     }
 
diff --git a/backend/windmill-queue/src/jobs.rs b/backend/windmill-queue/src/jobs.rs
index eede33ad9b..ba853311cf 100644
--- a/backend/windmill-queue/src/jobs.rs
+++ b/backend/windmill-queue/src/jobs.rs
@@ -2879,15 +2879,16 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
 
             let flow_status: FlowStatus = match restarted_from {
                 Some(restarted_from_val) => {
-                    let (_, _, step_n, truncated_modules, _) = restarted_flows_resolution(
-                        _db,
-                        workspace_id,
-                        Some(value.clone()),
-                        restarted_from_val.flow_job_id,
-                        restarted_from_val.step_id.as_str(),
-                        restarted_from_val.branch_or_iteration_n,
-                    )
-                    .await?;
+                    let (_, _, step_n, truncated_modules, _, user_states, cleanup_module) =
+                        restarted_flows_resolution(
+                            _db,
+                            workspace_id,
+                            Some(value.clone()),
+                            restarted_from_val.flow_job_id,
+                            restarted_from_val.step_id.as_str(),
+                            restarted_from_val.branch_or_iteration_n,
+                        )
+                        .await?;
                     FlowStatus {
                         step: step_n,
                         modules: truncated_modules,
@@ -2898,7 +2899,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
                                 id: "failure".to_string(),
                             },
                         },
-                        cleanup_module: FlowCleanupModule { flow_jobs_to_clean: vec![] },
+                        cleanup_module,
                         // retry status is reset
                         retry: RetryStatus { fail_count: 0, failed_jobs: vec![] },
                         // TODO: for now, flows with approval conditions aren't supported for restart
@@ -2908,6 +2909,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
                             step_id: restarted_from_val.step_id,
                             branch_or_iteration_n: restarted_from_val.branch_or_iteration_n,
                         }),
+                        user_states,
                     }
                 }
                 _ => FlowStatus::new(&value), // this is a new flow being pushed, flow_status is set to flow_value
@@ -3023,16 +3025,23 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
             )
         }
         JobPayload::RestartedFlow { completed_job_id, step_id, branch_or_iteration_n } => {
-            let (flow_path, raw_flow, step_n, truncated_modules, priority) =
-                restarted_flows_resolution(
-                    _db,
-                    workspace_id,
-                    None,
-                    completed_job_id,
-                    step_id.as_str(),
-                    branch_or_iteration_n,
-                )
-                .await?;
+            let (
+                flow_path,
+                raw_flow,
+                step_n,
+                truncated_modules,
+                priority,
+                user_states,
+                cleanup_module,
+            ) = restarted_flows_resolution(
+                _db,
+                workspace_id,
+                None,
+                completed_job_id,
+                step_id.as_str(),
+                branch_or_iteration_n,
+            )
+            .await?;
             let restarted_flow_status = FlowStatus {
                 step: step_n,
                 modules: truncated_modules,
@@ -3043,7 +3052,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
                         id: "failure".to_string(),
                     },
                 },
-                cleanup_module: FlowCleanupModule { flow_jobs_to_clean: vec![] },
+                cleanup_module,
                 // retry status is reset
                 retry: RetryStatus { fail_count: 0, failed_jobs: vec![] },
                 // TODO: for now, flows with approval conditions aren't supported for restart
@@ -3053,6 +3062,7 @@ pub async fn push<'c, T: Serialize + Send + Sync, R: rsmq_async::RsmqConnection
                     step_id,
                     branch_or_iteration_n,
                 }),
+                user_states,
             };
             (
                 None,
@@ -3383,6 +3393,8 @@ async fn restarted_flows_resolution(
         i32,
         Vec<FlowStatusModule>,
         Option<i16>,
+        HashMap<String, serde_json::Value>,
+        FlowCleanupModule,
     ),
     Error,
 > {
@@ -3541,6 +3553,7 @@ async fn restarted_flows_resolution(
             }?;
         }
     }
+
     if !dependent_module {
         // step not found in flow.
         return Err(Error::InternalErr(format!(
@@ -3555,5 +3568,7 @@ async fn restarted_flows_resolution(
         step_n,
         truncated_modules,
         completed_job.priority,
+        flow_status.user_states,
+        flow_status.cleanup_module,
     ));
 }
diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs
index 2cc4f361a1..89ef736765 100644
--- a/backend/windmill-worker/src/worker.rs
+++ b/backend/windmill-worker/src/worker.rs
@@ -1313,7 +1313,9 @@ pub async fn run_worker<R: rsmq_async::RsmqConnection + Send + Sync + Clone + 's
         if let Err(e) =
             queue_init_bash_maybe(db, same_worker_tx.clone(), &worker_name, rsmq.clone()).await
         {
+            killpill_tx.send(()).unwrap_or_default();
             tracing::error!("Error queuing init bash script for worker {worker_name}: {e}");
+            return;
         }
     }
 
diff --git a/python-client/wmill/wmill/client.py b/python-client/wmill/wmill/client.py
index 773a4c4918..d7debbcae1 100644
--- a/python-client/wmill/wmill/client.py
+++ b/python-client/wmill/wmill/client.py
@@ -346,6 +346,24 @@ class Windmill:
     def set_state(self, value: Any):
         self.set_resource(value, path=self.state_path, resource_type="state")
 
+    def set_flow_user_state(self, key: str, value: Any) -> None:
+        """Set the user state of a flow at a given key"""
+        flow_id = self.get_root_job_id()
+        r = client.post(f"/w/{self.workspace}/jobs/flow/user_states/{flow_id}/{key}", json=value,  raise_for_status=False)
+        if r.status_code == 404:
+            print(f"Job {flow_id} does not exist or is not a flow")
+
+
+    def get_flow_user_state(self, key: str) -> Any:
+        """Get the user state of a flow at a given key"""
+        flow_id = self.get_root_job_id()
+        r = client.get(f"/w/{self.workspace}/jobs/flow/user_states/{flow_id}/{key}", raise_for_status=False)
+        if r.status_code == 404:
+            print(f"Job {flow_id} does not exist or is not a flow")
+            return None
+        else:
+            return r.json()
+
     @property
     def version(self):
         return self.get("version").text
@@ -851,6 +869,20 @@ def set_variable(path: str, value: str, is_secret: bool = False) -> None:
     return _client.set_variable(path, value, is_secret)
 
 
+@init_global_client
+def get_flow_user_state(key: str) -> Any:
+    """
+    Get the user state of a flow at a given key
+    """
+    return _client.get_flow_user_state(key)
+
+def set_flow_user_state(key: str, value: Any) -> None:
+    """
+    Set the user state of a flow at a given key
+    """
+    return _client.set_flow_user_state(key, value)
+
+
 @init_global_client
 def get_state_path() -> str:
     return _client.state_path
diff --git a/typescript-client/build.sh b/typescript-client/build.sh
index e2bc2ffea0..e4dcc5b3d0 100755
--- a/typescript-client/build.sh
+++ b/typescript-client/build.sh
@@ -11,4 +11,4 @@ npx --yes openapi-typescript-codegen --input "${script_dirpath}/../backend/windm
 cp "${script_dirpath}/client.ts" "${script_dirpath}/src/"
 cp "${script_dirpath}/s3Types.ts" "${script_dirpath}/src/"
 echo 'export type { S3Object, DenoS3LightClientSettings } from "./s3Types";' >> "${script_dirpath}/src/index.ts"
-echo 'export { type Base64, setClient, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, getState, getIdToken, denoS3LightClientSettings, loadS3FileStream, loadS3File, writeS3File, task, runScript, runScriptAsync, waitJob, getRootJobId } from "./client";' >> "${script_dirpath}/src/index.ts"
+echo 'export { type Base64, setClient, getVariable, setVariable, getResource, setResource, getResumeUrls, setState, getState, getIdToken, denoS3LightClientSettings, loadS3FileStream, loadS3File, writeS3File, task, runScript, runScriptAsync, waitJob, getRootJobId, setFlowUserState, getFlowUserState } from "./client";' >> "${script_dirpath}/src/index.ts"
diff --git a/typescript-client/client.ts b/typescript-client/client.ts
index a2df2ae7d4..159b19ef95 100644
--- a/typescript-client/client.ts
+++ b/typescript-client/client.ts
@@ -332,6 +332,63 @@ export async function setState(state: any): Promise<void> {
   await setResource(state, undefined, "state");
 }
 
+
+/**
+ * Set a flow user state 
+ * @param key key of the state 
+ * @param value value of the state
+
+ */
+export async function setFlowUserState(
+  key: string,
+  value: any,
+  errorIfNotPossible?: boolean
+): Promise<void> {
+  !clientSet && setClient();
+  const workspace = getWorkspace();
+  try {
+    await JobService.setFlowUserState({
+      workspace,
+      id: await getRootJobId(),
+      key,
+      requestBody: value,
+    });
+  } catch (e: any) {
+    if (errorIfNotPossible) {
+      throw Error(`Error setting flow user state at ${key}: ${e.body}`);
+    } else {
+      console.error(`Error setting flow user state at ${key}: ${e.body}`);
+    }
+  }
+}
+
+/**
+ * Get a flow user state
+ * @param path path of the variable
+
+ */
+export async function getFlowUserState(
+  key: string,
+  errorIfNotPossible?: boolean
+): Promise<void> {
+  !clientSet && setClient();
+  const workspace = getWorkspace();
+  try {
+    await JobService.getFlowUserState({
+      workspace,
+      id: await getRootJobId(),
+      key,
+    });
+  } catch (e: any) {
+    if (errorIfNotPossible) {
+      throw Error(`Error setting flow user state at ${key}: ${e.body}`);
+    } else {
+      console.error(`Error setting flow user state at ${key}: ${e.body}`);
+    }
+  }
+}
+
+
 // /**
 //  * Set the shared state
 //  * @param state state to set