move s3 cred resolution later (#2954)

<!-- CURSOR_SUMMARY -->
> [!NOTE]
> Move S3 credential handling to runtime with
`resolve_s3_credentials_from_env`, add `resolve_credentials` to
`load_from_user_code`, and set context-specific env/flags across CLI,
planning, and TS/Py libs.
> 
> - **Core (InfrastructureMap)**:
> - Add `resolve_s3_credentials_from_env()` to resolve S3/S3Queue creds
from env at runtime and recalc `engine_params_hash`.
> - Change `load_from_user_code(project, resolve_credentials)` and
optionally resolve creds; propagate throughout callers.
> - Planning (`plan.rs`): load prebuilt JSON (no creds) in prod when
present, then ALWAYS resolve creds at runtime; when building from code,
pass `true`.
> - **Partial Map (DMV2)**:
> - Keep S3/S3Queue env markers in engine configs; remove build-time
secret resolution helper.
> - **CLI commands**:
>   - Pass resolve flag appropriately:
> - Runtime ops use `true`: `remote_plan`, `remote_gen_migration`,
`remote_refresh`, `migrate`, workflows, seed data.
> - Structure-only ops use `false`: `check` (and JSON write), `db pull`,
`ls`, `kafka pull`.
> - **Executors**:
> - Set `IS_LOADING_INFRA_MAP=true` when running TS `dmv2-serializer`
and Python serializer to emit markers during infra loading.
> - **Language libs (secrets)**:
> - TS/Py `mooseRuntimeEnv.get()`/`get()` now return markers when
`IS_LOADING_INFRA_MAP=true`, otherwise fetch real env vars (error if
unset); tests updated.
> 
> <sup>Written by [Cursor
Bugbot](https://cursor.com/dashboard?tab=bugbot) for commit
67dc4cc. This will update automatically
on new commits. Configure
[here](https://cursor.com/dashboard?tab=bugbot).</sup>
<!-- /CURSOR_SUMMARY -->

Author: phiSgr

apps/framework-cli/src/cli.rs

@@ -482,7 +482,8 @@ pub async fn top_command_handler(
 
             let infra_map = if project_arc.features.data_model_v2 {
                 debug!("Loading InfrastructureMap from user code (DMV2)");
-                InfrastructureMap::load_from_user_code(&project_arc)
+                // Don't resolve credentials for moose check - avoids baking into Docker
+                InfrastructureMap::load_from_user_code(&project_arc, false)
                     .await
                     .map_err(|e| {
                         RoutineFailure::error(Message {

apps/framework-cli/src/cli/routines.rs

@@ -888,7 +888,7 @@ async fn legacy_remote_plan_logic(
     // Build the inframap from the local project
     let local_infra_map = if project.features.data_model_v2 {
         debug!("Loading InfrastructureMap from user code (DMV2)");
-        InfrastructureMap::load_from_user_code(project).await?
+        InfrastructureMap::load_from_user_code(project, true).await?
     } else {
         debug!("Loading InfrastructureMap from primitives");
         let primitive_map = PrimitiveMap::load(project).await?;
@@ -994,7 +994,7 @@ pub async fn remote_plan(
     // Build the inframap from the local project
     let local_infra_map = if project.features.data_model_v2 {
         debug!("Loading InfrastructureMap from user code (DMV2)");
-        InfrastructureMap::load_from_user_code(project).await?
+        InfrastructureMap::load_from_user_code(project, true).await?
     } else {
         debug!("Loading InfrastructureMap from primitives");
         let primitive_map = PrimitiveMap::load(project).await?;
@@ -1114,9 +1114,10 @@ pub async fn remote_gen_migration(
     use anyhow::Context;
 
     // Build the inframap from the local project
+    // Resolve credentials for generating migration DDL with S3 tables
     let local_infra_map = if project.features.data_model_v2 {
         debug!("Loading InfrastructureMap from user code (DMV2)");
-        InfrastructureMap::load_from_user_code(project).await?
+        InfrastructureMap::load_from_user_code(project, true).await?
     } else {
         debug!("Loading InfrastructureMap from primitives");
         let primitive_map = PrimitiveMap::load(project).await?;
@@ -1235,7 +1236,7 @@ pub async fn remote_refresh(
     // Build the inframap from the local project
     let local_infra_map = if project.features.data_model_v2 {
         debug!("Loading InfrastructureMap from user code (DMV2)");
-        InfrastructureMap::load_from_user_code(project).await?
+        InfrastructureMap::load_from_user_code(project, true).await?
     } else {
         debug!("Loading InfrastructureMap from primitives");
         let primitive_map = PrimitiveMap::load(project).await?;

apps/framework-cli/src/cli/routines/code_generation.rs

@@ -493,7 +493,8 @@ pub async fn db_pull(
     let (client, db) = create_client_and_db(remote_url).await?;
 
     debug!("Loading InfrastructureMap from user code (DMV2)");
-    let infra_map = InfrastructureMap::load_from_user_code(project)
+    // Don't resolve credentials for code generation - only needs structure
+    let infra_map = InfrastructureMap::load_from_user_code(project, false)
         .await
         .map_err(|e| {
             RoutineFailure::error(Message::new(

apps/framework-cli/src/cli/routines/kafka_pull.rs

@@ -63,7 +63,8 @@ pub async fn write_external_topics(
         .filter(|n| inc.is_match(n) && !exc.is_match(n))
         .collect();
 
-    let infra_map = InfrastructureMap::load_from_user_code(project)
+    // Don't resolve credentials - only checking which topics are managed
+    let infra_map = InfrastructureMap::load_from_user_code(project, false)
         .await
         .map_err(|e| {
             RoutineFailure::error(Message::new(

apps/framework-cli/src/cli/routines/ls.rs

@@ -293,7 +293,8 @@ pub async fn ls_dmv2(
     name: Option<&str>,
     json: bool,
 ) -> Result<RoutineSuccess, RoutineFailure> {
-    let infra_map = InfrastructureMap::load_from_user_code(project)
+    // Don't resolve credentials for ls command - only inspects structure
+    let infra_map = InfrastructureMap::load_from_user_code(project, false)
         .await
         .map_err(|e| {
             RoutineFailure::new(

apps/framework-cli/src/cli/routines/migrate.rs

@@ -498,8 +498,8 @@ pub async fn execute_migration(
 
         let current_tables = &current_infra_map.tables;
 
-        // Load target state from current code
-        let target_infra_map = InfrastructureMap::load_from_user_code(project)
+        // Load target state from current code with credentials resolved for migration DDL
+        let target_infra_map = InfrastructureMap::load_from_user_code(project, true)
             .await
             .map_err(|e| {
                 RoutineFailure::new(

apps/framework-cli/src/cli/routines/scripts.rs

@@ -125,7 +125,8 @@ pub async fn run_workflow_and_get_run_ids(
     name: &str,
     input: Option<String>,
 ) -> Result<WorkflowStartInfo, RoutineFailure> {
-    let infra_map = InfrastructureMap::load_from_user_code(project)
+    // Resolve credentials for workflows that may interact with S3 tables
+    let infra_map = InfrastructureMap::load_from_user_code(project, true)
         .await
         .map_err(|e| {
             RoutineFailure::new(

apps/framework-cli/src/cli/routines/seed_data.rs

@@ -106,7 +106,8 @@ fn build_count_query(
 /// Loads the infrastructure map based on project configuration
 async fn load_infrastructure_map(project: &Project) -> Result<InfrastructureMap, RoutineFailure> {
     if project.features.data_model_v2 {
-        InfrastructureMap::load_from_user_code(project)
+        // Resolve credentials for seeding data into S3-backed tables
+        InfrastructureMap::load_from_user_code(project, true)
             .await
             .map_err(|e| {
                 RoutineFailure::error(Message {

apps/framework-cli/src/framework/core/infrastructure_map.rs

@@ -2222,6 +2222,108 @@ impl InfrastructureMap {
         Ok(infra_map)
     }
 
+    /// Resolves S3 credentials from environment variables at runtime.
+    ///
+    /// This method iterates through all tables in the infrastructure map and resolves
+    /// any environment variable markers (like `MOOSE_ENV::AWS_ACCESS_KEY_ID`) in S3 and S3Queue
+    /// engine configurations to their actual values from the environment.
+    ///
+    /// This MUST be called at runtime (dev/prod mode start) rather than at build time
+    /// to avoid baking credentials into Docker images.
+    ///
+    /// # Returns
+    /// A Result indicating success or containing an error message if credential resolution fails
+    pub fn resolve_s3_credentials_from_env(&mut self) -> Result<(), String> {
+        use crate::infrastructure::olap::clickhouse::queries::ClickhouseEngine;
+        use crate::utilities::secrets::resolve_optional_runtime_env;
+
+        for table in self.tables.values_mut() {
+            let mut should_recalc_hash = false;
+
+            if let Some(engine) = &mut table.engine {
+                match engine {
+                    ClickhouseEngine::S3Queue {
+                        aws_access_key_id,
+                        aws_secret_access_key,
+                        ..
+                    } => {
+                        // Resolve environment variable markers for AWS credentials
+                        let resolved_access_key = resolve_optional_runtime_env(aws_access_key_id)
+                            .map_err(|e| {
+                            format!(
+                                "Failed to resolve runtime environment variable for table '{}' field 'awsAccessKeyId': {}",
+                                table.name, e
+                            )
+                        })?;
+
+                        let resolved_secret_key =
+                            resolve_optional_runtime_env(aws_secret_access_key).map_err(|e| {
+                                format!(
+                                    "Failed to resolve runtime environment variable for table '{}' field 'awsSecretAccessKey': {}",
+                                    table.name, e
+                                )
+                            })?;
+
+                        *aws_access_key_id = resolved_access_key;
+                        *aws_secret_access_key = resolved_secret_key;
+                        should_recalc_hash = true;
+
+                        log::debug!(
+                            "Resolved S3Queue credentials for table '{}' at runtime",
+                            table.name
+                        );
+                    }
+                    ClickhouseEngine::S3 {
+                        aws_access_key_id,
+                        aws_secret_access_key,
+                        ..
+                    } => {
+                        // Resolve environment variable markers for AWS credentials
+                        let resolved_access_key = resolve_optional_runtime_env(aws_access_key_id)
+                            .map_err(|e| {
+                            format!(
+                                "Failed to resolve runtime environment variable for table '{}' field 'awsAccessKeyId': {}",
+                                table.name, e
+                            )
+                        })?;
+
+                        let resolved_secret_key =
+                            resolve_optional_runtime_env(aws_secret_access_key).map_err(|e| {
+                                format!(
+                                    "Failed to resolve runtime environment variable for table '{}' field 'awsSecretAccessKey': {}",
+                                    table.name, e
+                                )
+                            })?;
+
+                        *aws_access_key_id = resolved_access_key;
+                        *aws_secret_access_key = resolved_secret_key;
+                        should_recalc_hash = true;
+
+                        log::debug!(
+                            "Resolved S3 credentials for table '{}' at runtime",
+                            table.name
+                        );
+                    }
+                    _ => {
+                        // No credentials to resolve for other engine types
+                    }
+                }
+            }
+
+            // Recalculate engine_params_hash after resolving credentials
+            if should_recalc_hash {
+                table.engine_params_hash =
+                    table.engine.as_ref().map(|e| e.non_alterable_params_hash());
+                log::debug!(
+                    "Recalculated engine_params_hash for table '{}' after credential resolution",
+                    table.name
+                );
+            }
+        }
+
+        Ok(())
+    }
+
     /// Stores the infrastructure map in Redis for persistence and sharing
     ///
     /// Serializes the map to protocol buffers and stores it in Redis using
@@ -2472,10 +2574,16 @@ impl InfrastructureMap {
     ///
     /// # Arguments
     /// * `project` - The project to load the infrastructure map from
+    /// * `resolve_credentials` - Whether to resolve S3 credentials from environment variables.
+    ///   Set to `false` for build-time operations like `moose check` to avoid baking credentials
+    ///   into Docker images. Set to `true` for runtime operations that need to interact with infrastructure.
     ///
     /// # Returns
     /// A Result containing the infrastructure map or an error
-    pub async fn load_from_user_code(project: &Project) -> anyhow::Result<Self> {
+    pub async fn load_from_user_code(
+        project: &Project,
+        resolve_credentials: bool,
+    ) -> anyhow::Result<Self> {
         let partial = if project.language == SupportedLanguages::Typescript {
             let process = crate::framework::typescript::export_collectors::collect_from_index(
                 project,
@@ -2486,12 +2594,19 @@ impl InfrastructureMap {
         } else {
             load_main_py(project, &project.project_location).await?
         };
-        let infra_map = partial.into_infra_map(
+        let mut infra_map = partial.into_infra_map(
             project.language,
             &project.main_file(),
             &project.clickhouse_config.db_name,
         )?;
 
+        // Resolve S3 credentials at runtime if requested
+        if resolve_credentials {
+            infra_map
+                .resolve_s3_credentials_from_env()
+                .map_err(|e| anyhow::anyhow!("Failed to resolve S3 credentials: {}", e))?;
+        }
+
         // Provide explicit feedback when streams are defined but streaming engine is disabled
         if !project.features.streaming_engine && infra_map.uses_streaming() {
             show_message_wrapper(

apps/framework-cli/src/framework/core/partial_infrastructure_map.rs

@@ -65,7 +65,7 @@ use crate::{
         scripts::Workflow, versions::Version,
     },
     infrastructure::olap::clickhouse::queries::ClickhouseEngine,
-    utilities::{constants, secrets::resolve_optional_runtime_env},
+    utilities::constants,
 };
 
 /// Defines how Moose manages the lifecycle of database resources when code changes.
@@ -455,34 +455,6 @@ pub struct PartialInfrastructureMap {
     web_apps: HashMap<String, PartialWebApp>,
 }
 
-/// Helper function to resolve S3 credentials from environment variables
-/// This ensures consistent credential handling between S3 and S3Queue engines
-fn resolve_s3_credentials(
-    aws_access_key_id: &Option<String>,
-    aws_secret_access_key: &Option<String>,
-    table_name: &str,
-) -> Result<(Option<String>, Option<String>), DmV2LoadingError> {
-    // Resolve environment variable markers for AWS credentials at runtime
-    // This must happen before the infrastructure diff to support credential rotation
-    let resolved_access_key = resolve_optional_runtime_env(aws_access_key_id).map_err(|e| {
-        DmV2LoadingError::RuntimeEnvResolution {
-            table_name: table_name.to_string(),
-            field: "awsAccessKeyId".to_string(),
-            error: e.to_string(),
-        }
-    })?;
-
-    let resolved_secret_key = resolve_optional_runtime_env(aws_secret_access_key).map_err(|e| {
-        DmV2LoadingError::RuntimeEnvResolution {
-            table_name: table_name.to_string(),
-            field: "awsSecretAccessKey".to_string(),
-            error: e.to_string(),
-        }
-    })?;
-
-    Ok((resolved_access_key, resolved_secret_key))
-}
-
 impl PartialInfrastructureMap {
     /// Creates a new [`PartialInfrastructureMap`] by executing and reading from a subprocess.
     ///
@@ -802,37 +774,25 @@ impl PartialInfrastructureMap {
             })),
 
             Some(EngineConfig::S3Queue(config)) => {
-                // Resolve S3 credentials using shared helper
-                let (resolved_access_key, resolved_secret_key) = resolve_s3_credentials(
-                    &config.aws_access_key_id,
-                    &config.aws_secret_access_key,
-                    &partial_table.name,
-                )?;
-
+                // Keep environment variable markers as-is - credentials will be resolved at runtime
                 // S3Queue settings are handled in table_settings, not in the engine
                 Ok(Some(ClickhouseEngine::S3Queue {
                     s3_path: config.s3_path.clone(),
                     format: config.format.clone(),
                     compression: config.compression.clone(),
                     headers: config.headers.clone(),
-                    aws_access_key_id: resolved_access_key,
-                    aws_secret_access_key: resolved_secret_key,
+                    aws_access_key_id: config.aws_access_key_id.clone(),
+                    aws_secret_access_key: config.aws_secret_access_key.clone(),
                 }))
             }
 
             Some(EngineConfig::S3(config)) => {
-                // Resolve S3 credentials using shared helper
-                let (resolved_access_key, resolved_secret_key) = resolve_s3_credentials(
-                    &config.aws_access_key_id,
-                    &config.aws_secret_access_key,
-                    &partial_table.name,
-                )?;
-
+                // Keep environment variable markers as-is - credentials will be resolved at runtime
                 Ok(Some(ClickhouseEngine::S3 {
                     path: config.path.clone(),
                     format: config.format.clone(),
-                    aws_access_key_id: resolved_access_key,
-                    aws_secret_access_key: resolved_secret_key,
+                    aws_access_key_id: config.aws_access_key_id.clone(),
+                    aws_secret_access_key: config.aws_secret_access_key.clone(),
                     compression: config.compression.clone(),
                     partition_strategy: config.partition_strategy.clone(),
                     partition_columns_in_data_file: config.partition_columns_in_data_file.clone(),