Require reviewed workflow source hash on start

codex-rs/app-server-protocol/src/protocol/common.rs

@@ -4597,6 +4597,8 @@ mod tests {
             params: v2::ThreadWorkflowRunStartParams {
                 thread_id: "thr_123".to_string(),
                 workflow_record_id: "workflow_123".to_string(),
+                expected_source_yaml_sha256:
+                    "0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef".to_string(),
                 idempotency_key: None,
             },
         };

codex-rs/app-server-protocol/src/protocol/v2/workflow.rs

@@ -311,6 +311,7 @@ pub struct ThreadWorkflowRunGetResponse {
 pub struct ThreadWorkflowRunStartParams {
     pub thread_id: String,
     pub workflow_record_id: String,
+    pub expected_source_yaml_sha256: String,
     #[ts(optional = nullable)]
     pub idempotency_key: Option<String>,
 }

codex-rs/app-server/src/request_processors/thread_workflow_processor.rs

@@ -270,6 +270,9 @@ impl ThreadWorkflowRequestProcessor {
         let state_db = self.state_db_for_materialized_thread(thread_id).await?;
         let idempotency_key = params.idempotency_key.and_then(normalize_optional_string);
         let workflow_record_id = params.workflow_record_id;
+        let expected_source_yaml_sha256 =
+            normalize_optional_string(params.expected_source_yaml_sha256)
+                .ok_or_else(|| invalid_request("expectedSourceYamlSha256 is required"))?;
         let workflow = retry_transient_sqlite_busy("read thread workflow before run start", || {
             state_db
                 .workflows()
@@ -287,6 +290,7 @@ impl ThreadWorkflowRequestProcessor {
         let start_request = codex_workflows_extension::WorkflowStartRequest {
             workflow_record_id,
             source_thread_id: thread_id,
+            expected_source_yaml_sha256,
             idempotency_key: idempotency_key.clone(),
             activation_config: crate::extensions::workflow_activation_config(&self.config),
         };

codex-rs/app-server/src/workflow_provider_credit.rs

@@ -1345,7 +1345,8 @@ mod tests {
         state
             .workflows()
             .create_workflow_run(WorkflowRunCreateParams {
-                workflow_record_id: spec.workflow_record_id,
+                workflow_record_id: spec.workflow_record_id.clone(),
+                expected_source_yaml_sha256: spec.source_yaml_sha256.clone(),
                 source_thread_id: None,
                 idempotency_key: None,
             })

codex-rs/app-server/tests/suite/v2/workflow.rs

@@ -359,6 +359,7 @@ async fn workflow_run_lifecycle_projects_tasks_and_returns_sanitized_state() ->
             Some(json!({
                 "threadId": thread_id.as_str(),
                 "workflowRecordId": workflow.workflow_record_id.as_str(),
+                "expectedSourceYamlSha256": workflow.source_yaml_sha256.as_str(),
                 "idempotencyKey": "run-lifecycle",
             })),
         )
@@ -521,6 +522,158 @@ async fn workflow_run_lifecycle_projects_tasks_and_returns_sanitized_state() ->
     Ok(())
 }
 
+#[tokio::test]
+async fn workflow_create_canonicalizes_terminal_line_endings_in_shared_store() -> Result<()> {
+    let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
+    let codex_home = TempDir::new()?;
+    create_config_toml(codex_home.path(), &server.uri(), WorkflowsFeature::Enabled)?;
+    let thread_id = create_materialized_thread(codex_home.path(), "workflow source canonical")?;
+
+    let mut mcp = TestAppServer::new_without_managed_config(codex_home.path()).await?;
+    initialize(&mut mcp, ExperimentalApiCapability::Enabled).await?;
+
+    for (index, case) in [
+        "no-final-line-ending",
+        "final-lf",
+        "final-crlf",
+        "two-final-lfs",
+    ]
+    .into_iter()
+    .enumerate()
+    {
+        let marker = codex_home
+            .path()
+            .join(format!("workflow-source-canonical-{index}"));
+        let base = valid_workflow_yaml(&marker, &format!("wf_app_server_canonical_{index}"));
+        let base = base.trim_end_matches(['\r', '\n']).to_string();
+        let (source_yaml, expected_source_yaml) = match case {
+            "no-final-line-ending" => (base.clone(), base.clone()),
+            "final-lf" => (format!("{base}\n"), base.clone()),
+            "final-crlf" => (format!("{base}\r\n"), base.clone()),
+            "two-final-lfs" => (format!("{base}\n\n"), format!("{base}\n")),
+            _ => unreachable!("case list is exhaustive"),
+        };
+
+        let request_id =
+            send_workflow_create(&mut mcp, thread_id.as_str(), source_yaml.as_str()).await?;
+        let response = read_response(&mut mcp, request_id).await?;
+        let ThreadWorkflowCreateResponse { workflow } = to_response(response)?;
+        let runtime = open_state_runtime(codex_home.path()).await?;
+        let stored = runtime
+            .workflows()
+            .get_thread_workflow_spec(
+                parse_thread_id(thread_id.as_str())?,
+                workflow.workflow_record_id.as_str(),
+            )
+            .await?
+            .ok_or_else(|| anyhow::anyhow!("{case} workflow spec was not persisted"))?;
+
+        assert_eq!(
+            expected_source_yaml.as_bytes(),
+            stored.source_yaml.as_bytes()
+        );
+        assert_eq!(stored.source_yaml_sha256, workflow.source_yaml_sha256);
+    }
+
+    Ok(())
+}
+
+#[tokio::test]
+async fn workflow_run_start_rejects_stale_expected_source_sha_before_state_changes() -> Result<()> {
+    let server = create_mock_responses_server_sequence_unchecked(Vec::new()).await;
+    let codex_home = TempDir::new()?;
+    create_config_toml(codex_home.path(), &server.uri(), WorkflowsFeature::Enabled)?;
+    let thread_id = create_materialized_thread(codex_home.path(), "workflow stale source")?;
+    let marker = codex_home.path().join("workflow-stale-source-command-ran");
+    let yaml = valid_workflow_yaml(&marker, "wf_app_server_stale_source");
+
+    let mut mcp = TestAppServer::new_without_managed_config(codex_home.path()).await?;
+    initialize(&mut mcp, ExperimentalApiCapability::Enabled).await?;
+
+    let create_id = send_workflow_create(&mut mcp, thread_id.as_str(), yaml.as_str()).await?;
+    let create_resp = read_response(&mut mcp, create_id).await?;
+    let ThreadWorkflowCreateResponse { workflow } =
+        to_response::<ThreadWorkflowCreateResponse>(create_resp)?;
+    let updated_yaml = yaml.replace(
+        "Build a serious workflow without leaking",
+        "Build an updated serious workflow without leaking",
+    );
+    let update_id =
+        send_workflow_create(&mut mcp, thread_id.as_str(), updated_yaml.as_str()).await?;
+    let update_resp = read_response(&mut mcp, update_id).await?;
+    let ThreadWorkflowCreateResponse { workflow: updated } =
+        to_response::<ThreadWorkflowCreateResponse>(update_resp)?;
+    assert_eq!(workflow.workflow_record_id, updated.workflow_record_id);
+    assert_ne!(workflow.source_yaml_sha256, updated.source_yaml_sha256);
+
+    let stale_id = mcp
+        .send_raw_request(
+            "thread/workflow/run/start",
+            Some(json!({
+                "threadId": thread_id.as_str(),
+                "workflowRecordId": workflow.workflow_record_id.as_str(),
+                "expectedSourceYamlSha256": workflow.source_yaml_sha256.as_str(),
+                "idempotencyKey": "stale-source",
+            })),
+        )
+        .await?;
+    let stale = read_error(&mut mcp, stale_id).await?;
+    assert_eq!(stale.error.code, -32600);
+    assert!(
+        stale.error.message.contains("source YAML SHA mismatch"),
+        "unexpected stale-start error: {}",
+        stale.error.message
+    );
+    assert_no_workflow_runs(codex_home.path(), parse_thread_id(thread_id.as_str())?).await?;
+    assert_no_execution_side_effects(codex_home.path(), parse_thread_id(thread_id.as_str())?)
+        .await?;
+
+    let missing_id = mcp
+        .send_raw_request(
+            "thread/workflow/run/start",
+            Some(json!({
+                "threadId": thread_id.as_str(),
+                "workflowRecordId": workflow.workflow_record_id.as_str(),
+                "idempotencyKey": "missing-source",
+            })),
+        )
+        .await?;
+    let missing = read_error(&mut mcp, missing_id).await?;
+    assert_eq!(missing.error.code, -32600);
+    assert!(
+        missing
+            .error
+            .message
+            .contains("Invalid request: missing field")
+            && missing.error.message.contains("expectedSourceYamlSha256"),
+        "unexpected missing-source error: {}",
+        missing.error.message
+    );
+    assert_no_workflow_runs(codex_home.path(), parse_thread_id(thread_id.as_str())?).await?;
+
+    let matching_id = mcp
+        .send_raw_request(
+            "thread/workflow/run/start",
+            Some(json!({
+                "threadId": thread_id.as_str(),
+                "workflowRecordId": updated.workflow_record_id.as_str(),
+                "expectedSourceYamlSha256": updated.source_yaml_sha256.as_str(),
+                "idempotencyKey": "matching-source",
+            })),
+        )
+        .await?;
+    let matching_resp = read_response(&mut mcp, matching_id).await?;
+    let matching = to_response::<ThreadWorkflowRunStartResponse>(matching_resp)?;
+    assert_eq!(ThreadWorkflowRunStatus::Running, matching.run.run.status);
+    assert_eq!(
+        updated.source_yaml_sha256,
+        matching.run.run.source_yaml_sha256
+    );
+    assert!(!marker.exists());
+
+    Ok(())
+}
+
 #[tokio::test(flavor = "multi_thread", worker_threads = 2)]
 async fn workflow_start_activates_paired_real_workers_and_verifiers() -> Result<()> {
     let codex_home = TempDir::new()?;
@@ -589,6 +742,7 @@ async fn workflow_start_activates_paired_real_workers_and_verifiers() -> Result<
             Some(json!({
                 "threadId": thread_id.as_str(),
                 "workflowRecordId": workflow.workflow_record_id.as_str(),
+                "expectedSourceYamlSha256": workflow.source_yaml_sha256.as_str(),
                 "idempotencyKey": "actual-worker-activation",
             })),
         )
@@ -906,6 +1060,17 @@ async fn assert_no_execution_side_effects(codex_home: &Path, thread_id: ThreadId
     Ok(())
 }
 
+async fn assert_no_workflow_runs(codex_home: &Path, thread_id: ThreadId) -> Result<()> {
+    let runtime = open_state_runtime(codex_home).await?;
+    let page = runtime
+        .workflows()
+        .list_thread_workflow_runs_page(thread_id, /*cursor*/ None, /*limit*/ 10)
+        .await?;
+    assert_eq!(Vec::<codex_state::WorkflowRunSnapshot>::new(), page.data);
+    assert_eq!(None, page.next_cursor);
+    Ok(())
+}
+
 async fn assert_initial_execution_side_effects(
     codex_home: &Path,
     thread_id: ThreadId,