Support multi-environment apply_patch selection (#21617)

## Summary
- add multi-environment apply_patch routing for both freeform and
function-call tool flows
- parse and reconcile the optional environment selector in the main
apply_patch parser, then verify against the selected environment in the
handler
- carry environment_id through runtime and approval surfaces so
remote-targeted patches stay explicit end to end

## Testing
- just fmt
- remote exec-server e2e: `cargo test -p codex-core --test all
apply_patch_multi_environment_uses_remote_executor -- --nocapture` on
dev via `scripts/test-remote-env.sh`

---------

Co-authored-by: Codex <noreply@openai.com>

codex-rs/apply-patch/src/streaming_parser.rs

@@ -16,6 +16,8 @@ use crate::parser::UpdateFileChunk;
 use Hunk::*;
 use ParseError::*;
 
+const ENVIRONMENT_ID_MARKER: &str = "*** Environment ID: ";
+
 #[derive(Debug, Default, Clone)]
 pub struct StreamingPatchParser {
     line_buffer: String,
@@ -29,7 +31,7 @@ struct StreamingParserState {
     hunks: Vec<Hunk>,
 }
 
-#[derive(Debug, Default, Clone)]
+#[derive(Debug, Default, Clone, Copy)]
 enum StreamingParserMode {
     #[default]
     NotStarted,
@@ -43,6 +45,13 @@ enum StreamingParserMode {
 }
 
 impl StreamingPatchParser {
+    // The live streaming parser only needs to keep the patch preview flowing.
+    // Environment selection and validation happen on the final tool invocation,
+    // so here we just tolerate and skip the optional preamble line.
+    fn is_environment_id_preamble_line(&self, line: &str) -> bool {
+        line.starts_with(ENVIRONMENT_ID_MARKER)
+    }
+
     fn ensure_update_hunk_is_not_empty(&self, line: &str) -> Result<(), ParseError> {
         if let Some(UpdateFile { path, chunks, .. }) = self.state.hunks.last() {
             if chunks.is_empty()
@@ -150,7 +159,7 @@ impl StreamingPatchParser {
 
     fn process_line(&mut self, line: &str) -> Result<(), ParseError> {
         let trimmed = line.trim();
-        match self.state.mode.clone() {
+        match self.state.mode {
             StreamingParserMode::NotStarted => {
                 if trimmed == BEGIN_PATCH_MARKER {
                     self.state.mode = StreamingParserMode::StartedPatch;
@@ -161,6 +170,9 @@ impl StreamingPatchParser {
                 ))
             }
             StreamingParserMode::StartedPatch => {
+                if self.is_environment_id_preamble_line(line) {
+                    return Ok(());
+                }
                 if self.handle_hunk_headers_and_end_patch(trimmed)? {
                     return Ok(());
                 }
@@ -431,6 +443,32 @@ mod tests {
         );
     }
 
+    #[test]
+    fn test_streaming_patch_parser_environment_id_mode() {
+        let patch = "\
+*** Begin Patch
+*** Environment ID: remote
+*** Add File: src/hello.txt
++hello
+*** End Patch
+";
+
+        let mut parser = StreamingPatchParser::default();
+        assert_eq!(
+            parser.push_delta(patch),
+            Ok(vec![AddFile {
+                path: PathBuf::from("src/hello.txt"),
+                contents: "hello\n".to_string(),
+            }])
+        );
+
+        let mut parser = StreamingPatchParser::default();
+        assert_eq!(
+            parser.push_delta("*** Begin Patch\n*** Environment ID:   \n"),
+            Ok(vec![])
+        );
+    }
+
     #[test]
     fn test_streaming_patch_parser_large_patch_split_by_character() {
         let patch = "\