noded: a log chunk's job record counts the log it was read from

CHANGELOG.md

@@ -178,6 +178,13 @@
   the accept loop was busy, had SYNs dropped: those clients waited 1, 3, 7 s for
   retransmits and could pass the gateway's 5 s connect timeout. The held
   job-read test hit this in CI (2995 of 3000 reads held at its deadline).
+- **A ucloud-noded log chunk's `job` record counts the bytes the chunk was read
+  from.** The record sized the running job's log, then the chunk opened it
+  again, so output written in between could come back as data past the
+  record's `stdout_bytes`/`stderr_bytes` (and its `cursor`), which the gateway
+  cached as the job's state. Held reads never returned early; the held
+  job-read test hit this in CI (a logs read answered with the byte it waited
+  for and `stdout_bytes` 0).
 - **Buffered worker RPCs never reuse a connection the gateway asked to close**
   (RFC 9112 9.6). aiohttp pooled such a connection unless the answer also said
   `Connection: close`. ucloud-noded and the Python agent always say so, so

runtime/noded/src/jobs/files.rs

@@ -394,6 +394,22 @@ impl JobRecord {
         if stream == "stderr" { self.state.stderr_bytes } else { self.state.stdout_bytes }
     }
 
+    /// The record a log chunk carries. The record counted the log just
+    /// before the chunk's own open, and the job may have written in between;
+    /// while it runs, the stream's count is the size the chunk was read
+    /// against, so the data never runs past the record that comes with it.
+    /// Once terminal, the persisted counts stand: the logs no longer grow.
+    pub fn read_against(mut self, stream: &str, size: u64) -> JobRecord {
+        if !self.terminal() {
+            if stream == "stderr" {
+                self.state.stderr_bytes = size;
+            } else {
+                self.state.stdout_bytes = size;
+            }
+        }
+        self
+    }
+
     /// `ManagedProcessRecord.to_dict()`, in its key order, plus `cursor`.
     pub fn to_json(&self) -> Value {
         let state = &self.state;
@@ -490,6 +506,24 @@ mod tests {
         );
     }
 
+    #[test]
+    fn a_chunks_record_counts_the_log_it_was_read_from() {
+        let record = |state: &str| JobRecord {
+            sandbox_id: "sb".into(),
+            sandbox_generation: 3,
+            state: JobState::parse(state.as_bytes()).unwrap(),
+            updated_at: "2026-10-11T20:00:02.000001+00:00".into(),
+        };
+        // The job wrote one byte between the record's read and the chunk's.
+        let running = record(GO_RUNNING).read_against("stdout", 1);
+        assert_eq!((running.state.stdout_bytes, running.state.stderr_bytes, running.cursor()), (1, 0, "2.1.0".into()));
+        let running = record(GO_RUNNING).read_against("stderr", 4);
+        assert_eq!((running.state.stdout_bytes, running.state.stderr_bytes), (0, 4));
+        // Terminal: the persisted counts stand.
+        let exited = GO_RUNNING.replace(r#""state":"running""#, r#""state":"exited","exit_code":0"#).replace(r#""stdout_bytes":0"#, r#""stdout_bytes":7"#);
+        assert_eq!(record(&exited).read_against("stdout", 9).state.stdout_bytes, 7);
+    }
+
     #[test]
     fn base64_matches_python() {
         // python3 -c 'import base64; print(base64.b64encode(b"..."))'

runtime/noded/src/jobs/mod.rs

@@ -324,11 +324,14 @@ fn read_pass(
             return Err(PassError::NotFound);
         }
         let ReadKind::Logs { stream, offset, limit } = kind else { return Ok(Pass { record, chunk: None }) };
-        let chunk = match dir.log(stream).map_err(PassError::Read)? {
+        let (record, chunk) = match dir.log(stream).map_err(PassError::Read)? {
             // The supervisor's answer for a log it never opened.
-            None => (*offset, Vec::new(), *offset, false),
+            None => (record, (*offset, Vec::new(), *offset, false)),
             Some(log) if *offset > log.size => return Err(PassError::OffsetBeyond),
-            Some(log) => (*offset, log.read(*offset, *limit).map_err(|error| PassError::Read(ReadError::Io(error)))?, log.size, true),
+            Some(log) => {
+                let data = log.read(*offset, *limit).map_err(|error| PassError::Read(ReadError::Io(error)))?;
+                (record.read_against(stream, log.size), (*offset, data, log.size, true))
+            }
         };
         Ok(Pass { record, chunk: Some(chunk) })
     })();