Kill the survivors: descriptor placement as a testable function, token bound at its edge, zero-use window kept, readout identity and frame edge pinned

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>

crates/claude-connector/src/backend/tests.rs

@@ -1956,7 +1956,7 @@ fn rate_limit_event_updates_quota_cell() {
 
 #[test]
 fn a_rejected_turn_still_reports_its_windows_and_a_bad_window_costs_only_itself() {
-    let event = r#"{"type":"rate_limit_event","rate_limit_info":{"status":"rejected","unifiedWindows":{"five_hour":{"utilization":0.9996,"resetsAt":1793307000,"status":"rejected"},"seven_day":{"resetsAt":1793656800,"status":"allowed"},"odd":{"utilization":"high","resetsAt":1,"status":"allowed"},"negative":{"utilization":-0.5,"resetsAt":1,"status":"allowed"}}}}"#;
+    let event = r#"{"type":"rate_limit_event","rate_limit_info":{"status":"rejected","unifiedWindows":{"five_hour":{"utilization":0.9996,"resetsAt":1793307000,"status":"rejected"},"seven_day":{"resetsAt":1793656800,"status":"allowed"},"idle":{"utilization":0,"resetsAt":1793307000,"status":"allowed"},"odd":{"utilization":"high","resetsAt":1,"status":"allowed"},"negative":{"utilization":-0.5,"resetsAt":1,"status":"allowed"}}}}"#;
     let rig = rig(vec![Script::lines(vec![
         init_line(MODEL, SESSION),
         event.to_string(),
@@ -1968,7 +1968,11 @@ fn a_rejected_turn_still_reports_its_windows_and_a_bad_window_costs_only_itself(
     rejected.status = "rejected".to_string();
     // Unknown is never zero: a window missing a field or of another type is
     // skipped, not read as empty.
-    assert_eq!(quota_of(&rig).windows, vec![rejected]);
+    // A window at exactly zero use is a reading, and is kept.
+    assert_eq!(
+        quota_of(&rig).windows,
+        vec![rejected, window("idle", 0, 1_793_307_000_000)]
+    );
 }
 
 #[test]
@@ -2261,7 +2265,7 @@ mod real {
     }
 
     #[test]
-    fn a_token_too_large_for_a_pipe_is_refused_not_hung() {
+    fn a_token_too_large_for_a_pipe_is_refused_at_its_bound_not_hung() {
         let fake = fake("exit 0", limits());
         let config = config(fake.dir.path(), INSTANCE, limits());
         let env = child_env(&config, &request(vec![user("x")]));
@@ -2273,9 +2277,13 @@ mod real {
                 .to_string_lossy()
                 .into_owned(),
         ];
-        let huge = "t".repeat(1 << 20);
-        let refused = port.spawn(&argv, &env, fake.dir.path(), Vec::new(), &huge);
-        assert!(refused.is_err());
+        let spawn = |token: &str| port.spawn(&argv, &env, fake.dir.path(), Vec::new(), token);
+        let at_bound = spawn(&"t".repeat(4096)).expect("4096 bytes fit the bound");
+        drop(at_bound);
+        assert_eq!(
+            spawn(&"t".repeat(4097)).err().map(|e| e.kind()),
+            Some(io::ErrorKind::InvalidInput)
+        );
     }
 
     #[test]

crates/claude-connector/src/process.rs

@@ -67,21 +67,30 @@ mod unix {
     /// hundred bytes, so anything near the pipe's capacity is not one.
     const MAX_TOKEN_BYTES: usize = 4096;
 
-    /// Puts the pipe's read end at [`TOKEN_FD`] in the child. Runs between
-    /// fork and exec, so it calls nothing but `dup2` and `fcntl`.
-    fn place_token_fd(source: RawFd) -> io::Result<()> {
+    fn cvt(result: libc::c_int) -> io::Result<libc::c_int> {
+        if result < 0 {
+            Err(io::Error::last_os_error())
+        } else {
+            Ok(result)
+        }
+    }
+
+    /// Makes `source` also descriptor `target`, open across exec. Runs between
+    /// fork and exec in the child, so it calls nothing but `dup2` and `fcntl`.
+    fn place_fd(source: RawFd, target: RawFd) -> io::Result<()> {
         // SAFETY: dup2 and fcntl take integers only and are async-signal-safe.
         unsafe {
-            if source == TOKEN_FD {
+            if source == target {
                 // dup2 onto itself changes nothing, so the close-on-exec flag
                 // the pipe was created with must be cleared by hand.
-                let flags = libc::fcntl(TOKEN_FD, libc::F_GETFD);
-                if flags < 0 || libc::fcntl(TOKEN_FD, libc::F_SETFD, flags & !libc::FD_CLOEXEC) < 0
-                {
-                    return Err(io::Error::last_os_error());
-                }
-            } else if libc::dup2(source, TOKEN_FD) < 0 {
-                return Err(io::Error::last_os_error());
+                let flags = cvt(libc::fcntl(target, libc::F_GETFD))?;
+                cvt(libc::fcntl(
+                    target,
+                    libc::F_SETFD,
+                    flags & !libc::FD_CLOEXEC,
+                ))?;
+            } else {
+                cvt(libc::dup2(source, target))?;
             }
         }
         Ok(())
@@ -126,13 +135,13 @@ mod unix {
                 .process_group(0);
             // The write end closes here, so the child sees the token and then
             // the end of the pipe. Both ends are close-on-exec; only the copy
-            // `place_token_fd` makes reaches the child.
+            // `place_fd` makes reaches the child.
             let (reader, mut writer) = std::io::pipe()?;
             writer.write_all(token.as_bytes())?;
             drop(writer);
             let source = reader.as_raw_fd();
             // SAFETY: the closure calls only async-signal-safe functions.
-            unsafe { command.pre_exec(move || place_token_fd(source)) };
+            unsafe { command.pre_exec(move || place_fd(source, TOKEN_FD)) };
             let mut child = command.spawn()?;
             drop(reader);
             let (stdin_pipe, stdout, stderr) =
@@ -271,4 +280,70 @@ mod unix {
             self.wait();
         }
     }
+
+    #[cfg(test)]
+    mod tests {
+        use super::*;
+
+        fn cloexec(fd: RawFd) -> bool {
+            // SAFETY: fcntl on an integer descriptor.
+            let flags = unsafe { libc::fcntl(fd, libc::F_GETFD) };
+            assert!(flags >= 0, "descriptor {fd} is not open");
+            flags & libc::FD_CLOEXEC != 0
+        }
+
+        /// A descriptor at or above 300 on the same pipe, so a test never
+        /// touches a low number another thread may be using.
+        fn high_dup(source: RawFd, close_on_exec: bool) -> RawFd {
+            let command = if close_on_exec {
+                libc::F_DUPFD_CLOEXEC
+            } else {
+                libc::F_DUPFD
+            };
+            // SAFETY: fcntl on an integer descriptor.
+            let fd = unsafe { libc::fcntl(source, command, 300) };
+            assert!(fd >= 300);
+            fd
+        }
+
+        fn close(fd: RawFd) {
+            // SAFETY: closing a descriptor the test owns.
+            unsafe { libc::close(fd) };
+        }
+
+        #[test]
+        fn placing_a_descriptor_on_itself_clears_close_on_exec() {
+            let (reader, _writer) = std::io::pipe().unwrap();
+            let fd = high_dup(reader.as_raw_fd(), true);
+            assert!(cloexec(fd));
+            place_fd(fd, fd).unwrap();
+            assert!(!cloexec(fd), "it must survive exec");
+            // One that already survives is left alone, not refused.
+            place_fd(fd, fd).unwrap();
+            assert!(!cloexec(fd));
+            close(fd);
+        }
+
+        #[test]
+        fn placing_a_descriptor_elsewhere_copies_it_open_across_exec() {
+            let (reader, _writer) = std::io::pipe().unwrap();
+            let source = high_dup(reader.as_raw_fd(), true);
+            let target = high_dup(reader.as_raw_fd(), true);
+            assert!(cloexec(target));
+            place_fd(source, target).unwrap();
+            assert!(!cloexec(target), "the copy survives exec");
+            assert!(cloexec(source), "the source does not");
+            close(source);
+            close(target);
+        }
+
+        #[test]
+        fn a_descriptor_that_is_not_open_is_refused_either_way() {
+            let (reader, _writer) = std::io::pipe().unwrap();
+            let closed = high_dup(reader.as_raw_fd(), true);
+            close(closed);
+            assert!(place_fd(closed, closed).is_err());
+            assert!(place_fd(closed, closed + 1).is_err());
+        }
+    }
 }

crates/model-connector/src/client.rs

@@ -256,4 +256,56 @@ mod fixture_tests {
             "the client must not even connect"
         );
     }
+
+    #[cfg(feature = "daemon")]
+    #[test]
+    fn a_readout_is_accepted_only_for_the_identity_that_asked() {
+        use crate::readout::PlanQuotaSnapshot;
+        // (caller the answer is sealed for, request id override, accepted)
+        let cases = [
+            ("caller-1", None, true),
+            ("caller-2", None, false),
+            ("caller-1", Some("some-other-request"), false),
+        ];
+        for (answer_caller, answer_request, accepted) in cases {
+            let listener = std::net::TcpListener::bind("127.0.0.1:0").unwrap();
+            let endpoint = listener.local_addr().unwrap();
+            let key = security("readout-client-key");
+            let server = std::thread::spawn(move || {
+                let (mut stream, _) = listener.accept().unwrap();
+                let request = read_transport_frame(&mut stream, 64 * 1024).unwrap();
+                let RequestFrame::SpendReadout(asked) = decode_request_frame(&request).unwrap()
+                else {
+                    panic!("the client sent something other than a readout request");
+                };
+                let request_id = answer_request.unwrap_or(&asked.request_id);
+                let readout = SpendReadout::new(1, Vec::new(), PlanQuotaSnapshot::default());
+                let sealed =
+                    encrypt_spend_readout(answer_caller, request_id, &readout, &key).unwrap();
+                write_transport_frame(
+                    &mut stream,
+                    &encode_spend_readout_response_frame(&sealed).unwrap(),
+                    64 * 1024,
+                )
+                .unwrap();
+            });
+            let client = ConnectorClient::new(
+                endpoint,
+                "readout-client-key",
+                64 * 1024,
+                Some(Duration::from_secs(1)),
+            )
+            .unwrap();
+            let answer = client.spend_readout("caller-1");
+            if accepted {
+                assert_eq!(answer.unwrap().at_unix_ms, 1);
+            } else {
+                assert!(matches!(
+                    answer,
+                    Err(ConnectorClientError::Transport(ServiceError::OuterIdentity))
+                ));
+            }
+            server.join().unwrap();
+        }
+    }
 }

crates/model-connector/src/envelope.rs

@@ -1307,4 +1307,29 @@ mod tests {
             ServiceError::Encryption
         );
     }
+
+    #[cfg(feature = "daemon")]
+    #[test]
+    fn a_response_is_read_as_its_own_operation_only() {
+        use crate::readout::PlanQuotaSnapshot;
+        let key = TransportKey::from_connection_secret("readout-test-key").unwrap();
+        let readout = SpendReadout::new(1, Vec::new(), PlanQuotaSnapshot::default());
+        let sealed = encrypt_spend_readout("caller-1", "readout-1", &readout, &key).unwrap();
+        let readout_frame = encode_spend_readout_response_frame(&sealed).unwrap();
+        let invoke_frame = encode_result_frame(&sealed).unwrap();
+        assert_eq!(
+            decode_spend_readout_response_frame(&readout_frame).unwrap(),
+            sealed
+        );
+        assert_eq!(decode_result_frame(&invoke_frame).unwrap(), sealed);
+        assert_eq!(
+            decode_result_frame(&readout_frame),
+            Err(ServiceError::Carriage)
+        );
+        assert_eq!(
+            decode_spend_readout_response_frame(&invoke_frame),
+            Err(ServiceError::Carriage)
+        );
+        assert_eq!(decrypt_spend_readout(&sealed, &key).unwrap(), readout);
+    }
 }

crates/model-connector/src/service.rs

@@ -3837,10 +3837,19 @@ mod tests {
     fn a_readout_that_cannot_fit_the_frame_is_a_frame_bound_failure() {
         let store = TestStore::new();
         let service = open_service(&store, 4);
+        let asked = encrypt_spend_readout_request(CALLER, "readout-1", &security(KEY)).unwrap();
+        let answer = service
+            .spend_readout(&asked, 1_000, PlanQuotaSnapshot::default(), FRAME)
+            .unwrap();
+        let exact = encode_spend_readout_response_frame(&answer).unwrap().len();
+        assert!(readout_as(&service, CALLER, KEY, 1_000, exact).is_ok());
+        assert_eq!(
+            readout_as(&service, CALLER, KEY, 1_000, exact - 1).unwrap_err(),
+            ServiceError::FrameBound
+        );
         assert_eq!(
             readout_as(&service, CALLER, KEY, 1_000, 64).unwrap_err(),
             ServiceError::FrameBound
         );
-        assert!(readout_as(&service, CALLER, KEY, 1_000, FRAME).is_ok());
     }
 }