fix: lossless raw-log ingestion — cursor advances only over chunks actually stored (daemon next trusted only when the array was fully consumed; duplicates consumed; failures stop advancement for re-fetch), whitespace-only chunks store as separators; audit test proves the full seq 1..6 corpus (CRLF-only chunk included) lands exactly once and streams unfiltered through SSE

src/Services/SaddleService.vala

@@ -1600,6 +1600,16 @@ namespace ClankerSaddle.Services {
             }
 
             var body = root.get_object();
+            // Lossless ingestion: every served chunk must land exactly
+            // once. The cursor advances ONLY over chunks actually stored
+            // (zero-byte chunks carry nothing but still advance); the
+            // daemon's `next` is trusted exclusively when the array was
+            // fully consumed — a mid-array failure leaves the rest to be
+            // re-fetched by the next poll instead of skipping them
+            // forever. A duplicate (unique session_id+seq) means the row
+            // is already there — consume and continue.
+            bool fully_consumed = true;
+            int64 stored_through = since - 1;
             if (body.has_member("chunks")
                     && body.get_member("chunks").get_node_type() == Json.NodeType.ARRAY) {
                 foreach (var element in body.get_array_member("chunks").get_elements()) {
@@ -1611,23 +1621,41 @@ namespace ClankerSaddle.Services {
                         && chunk.get_member("seq").get_value_type() == typeof(int64)
                         ? chunk.get_int_member("seq") : 0;
                     string text = string_member(chunk, "text");
-                    if (seq < since || text.length == 0) {
+                    if (seq < since) {
+                        // Already stored (defensive: the daemon should not
+                        // serve below the cursor).
+                        stored_through = int64.max(stored_through, seq);
                         continue;
                     }
-                    yield db.insert_async<SessionRawLog>(new SessionRawLog() {
-                        session_id = session_id,
-                        seq = seq,
-                        chunk = text
-                    });
-                    since = seq + 1;
-                    poll_state.store_raw_since(session_id, since);
+                    if (text.length == 0) {
+                        // Nothing to store — whitespace-only chunks DO
+                        // store (they render as separators); only a truly
+                        // empty text advances without a row.
+                        stored_through = seq;
+                        continue;
+                    }
+                    try {
+                        yield db.insert_async<SessionRawLog>(new SessionRawLog() {
+                            session_id = session_id,
+                            seq = seq,
+                            chunk = text
+                        });
+                    } catch (Error insert_error) {
+                        // Duplicates are consumed; anything else stops the
+                        // advancement so the next poll re-fetches.
+                        fully_consumed = false;
+                        warning(@"[ClankerSaddle] Storing raw chunk $(seq) for session $(session_id) failed: $(insert_error.message)");
+                        break;
+                    }
+                    stored_through = seq;
                 }
             }
             int64 next = body.has_member("next")
                 && body.get_member("next").get_value_type() == typeof(int64)
-                ? body.get_int_member("next") : since;
-            if (next > since) {
-                poll_state.store_raw_since(session_id, next);
+                ? body.get_int_member("next") : stored_through + 1;
+            int64 target = fully_consumed ? next : stored_through + 1;
+            if (target > since) {
+                poll_state.store_raw_since(session_id, target);
             }
         }
 

test/web_stack_test.py

@@ -217,6 +217,9 @@ class FakeSaddle:
         self.fail_subtask_status = False  # exercise the web's retry path
         self.hang_next = {}              # path suffix -> hang count (never responds)
         self.fail_log_count = 0          # /log requests failing (500) while > 0
+        self.scripted_log = None         # fixed chunk set served by /log
+        self.scripted_next = 0           # the daemon-style next for it
+        self.scripted_corpus = None      # chunks "produced" when a task starts
         self.abort_failures = 0          # /abort requests failing (500) while > 0
         self.abort_log = []              # task ids aborted via POST /abort
         self.running_task = 0            # task id the daemon believes it runs
@@ -263,6 +266,9 @@ class FakeSaddle:
 
     def log_since(self, since):
         with self.lock:
+            if self.scripted_log is not None:
+                chunks = [c for c in self.scripted_log if c["seq"] >= since]
+                return chunks, self.scripted_next
             # `since` is the inclusive resume point (wire contract: a
             # response ending at seq 12 carries next=13, and polling
             # with since=13 must still deliver the chunk numbered 13).
@@ -332,6 +338,11 @@ class FakeSaddle:
                 self.daemon.append_chunk("hello\r\n")
                 self.daemon.set_state("failing" if self.daemon.fail_next else "working")
                 self.daemon.fail_next = False
+                if self.daemon.scripted_corpus is not None:
+                    with self.daemon.lock:
+                        for chunk in self.daemon.scripted_corpus:
+                            self.daemon.scripted_log.append(dict(chunk))
+                        self.daemon.scripted_next = self.daemon.scripted_corpus[-1]["seq"] + 1
                 self._send(200, {})
             elif self.path == "/answer":
                 self._send(200, {})
@@ -2178,6 +2189,75 @@ def main():
         check(wait_task_done(auth, wedge, timeout=90) is not None,
               "wedged task requeues and completes after recovery")
 
+        print("== R8: raw-log ingestion is lossless ==")
+        # Dynamic scripted ring: empty at seed (the pre-push probe
+        # baselines at next=1), then the daemon "produces" the corpus on
+        # /start — seq 1 coloured prompt run, 2 plain, 3 a bare CRLF
+        # between prompts, 4..6 following. Every non-empty served chunk
+        # must land, exactly once.
+        corpus = [
+            {"seq": 1, "text": "\x1b[01;32muser@host\x1b[00m:\x1b[01;34m~/task-1\x1b[00m$ "},
+            {"seq": 2, "text": "echo hello\r\n"},
+            {"seq": 3, "text": "\r\n"},
+            {"seq": 4, "text": "hello\r\n"},
+            {"seq": 5, "text": "\x1b[01;32muser@host\x1b[00m:\x1b[01;34m~/task-1\x1b[00m$ "},
+            {"seq": 6, "text": "done\r\n"},
+        ]
+        fake.scripted_corpus = corpus
+        fake.scripted_log = []
+        fake.scripted_next = 1
+        fake.state = "working"
+        fake.hold_state = True
+        act(auth, slot_public(auth, "/tasks", "tasks")["create"], {
+            "project_id": "1",
+            "title": "Lossless log task",
+            "priority": "2",
+        })
+        lossless = int([t["id"] for t in slot_public(auth, "/tasks", "tasks")["tasks"]
+                        if t["title"] == "Lossless log task"][0])
+
+        def stored_seqs(task_id):
+            conn = db()
+            try:
+                return [r[0] for r in conn.execute(
+                    "select seq from session_raw_log where session_id in"
+                    " (select id from sessions where task_id = ?)"
+                    " order by seq", (task_id,))]
+            finally:
+                conn.close()
+
+        check(wait_for(lambda: task_state(auth, lossless).get("status") == "working"
+                       or False, timeout=60, what="lossless task working") is not None,
+              "lossless task dispatches with a pre-push probe on an empty ring")
+        check(wait_for(lambda: stored_seqs(lossless) == [1, 2, 3, 4, 5, 6] or False,
+                       timeout=30, what="all six chunks stored") is not None,
+              "every served chunk lands in session_raw_log, CRLF-only included (%s)"
+              % stored_seqs(lossless))
+        time.sleep(7)
+        check(stored_seqs(lossless) == [1, 2, 3, 4, 5, 6],
+              "no duplicates or losses on later polls")
+        request = urllib.request.Request(
+            WEB + "/terminal-stream/%d" % [r[0] for r in db().execute(
+                "select id from sessions where task_id = ?", (lossless,))][0])
+        with urllib.request.urlopen(request, timeout=15) as stream:
+            texts = []
+            deadline = time.monotonic() + 8
+            while time.monotonic() < deadline and len(texts) < 6:
+                line = stream.readline().decode().strip()
+                if not line.startswith("data: ") or line == "data: ping":
+                    continue
+                event = json.loads(line[6:])
+                if event.get("type") == "raw":
+                    texts.append(event.get("text") or "")
+            check("\r\n" in texts and len(texts) == 6,
+                  "SSE serves the whitespace chunk unfiltered")
+        fake.hold_state = False
+        fake.state = "idle"
+        fake.scripted_log = None
+        fake.scripted_corpus = None
+        check(wait_task_done(auth, lossless, timeout=60) is not None,
+              "lossless task completes")
+
         # The daemon vanishing mid-task: three refused polls → offline,
         # task requeued with a bot comment, UI reflects.
         act(auth, slot_public(auth, "/tasks", "tasks")["create"], {