0.3.9 live-test lane: frontdoor resume de-chunk, Bedrock throttling ≠ overflow, silent file clicks, stale question cards, stdout log, activity colors

ops/hub/hub-frontdoor.py

@@ -1606,10 +1606,17 @@ def _dechunk(buf, state):
                 buf = buf[take:]
                 state["size"] -= take
                 if state["size"] == 0:
-                    state["mode"] = "size"
-                    if buf.startswith(b"\r\n"): buf = buf[2:]
+                    # the CRLF closing a chunk can straddle reads: consume it
+                    # in its own state, never as an (empty → unparseable)
+                    # size line — that wedged the stream for good
+                    state["mode"] = "crlf"
                     continue
                 break
+            if state["mode"] == "crlf":
+                if len(buf) < 2: break
+                if buf.startswith(b"\r\n"): buf = buf[2:]
+                state["mode"] = "size"
+                continue
             if state["mode"] == "done": break
         return out, buf
 
@@ -1649,8 +1656,18 @@ def _resume_loop(self, c, shard_id, addr, cursor):
             # Byte-level edits (gap frames, synthetic heartbeats, swallowed
             # frames) are only valid on a stream we fully own.
             _head, _, _body = pre.partition(b"\r\n\r\n")
-            _dcst = {"mode": "size", "size": 0}
-            _clean, _rest = self._dechunk(_body, _dcst)
+            # de-chunk ONLY a chunked body (an identity-encoded engine body —
+            # HTTP/1.0, or a future non-chunked transport — parses as-is; the
+            # unconditional de-chunk dropped every frame of such a body). The
+            # unconsumed tail (a chunk-size line split across reads) carries
+            # into the next read instead of being discarded.
+            _dcst = {"mode": "size", "size": 0,
+                     "chunked": b"transfer-encoding: chunked" in _head.lower()}
+            _pend = b""
+            if _dcst["chunked"]:
+                _clean, _pend = self._dechunk(_body, _dcst)
+            else:
+                _clean = _body
             reasm += _clean
             frames, reasm = _parse_frames(reasm)
             with self.lock:
@@ -1725,7 +1742,10 @@ def _resume_loop(self, c, shard_id, addr, cursor):
                     if chunk is None:
                         continue
                     _hb_ticks = 0
-                    _dc_out, _rest = self._dechunk(chunk, _dcst)
+                    if _dcst["chunked"]:
+                        _dc_out, _pend = self._dechunk(_pend + chunk, _dcst)
+                    else:
+                        _dc_out = chunk
                     reasm += _dc_out
                     frames, reasm = _parse_frames(reasm)
                     if b"server.cursor.unservable" in reasm or any(b"server.cursor.unservable" in f for f in frames):

ops/hub/test_frontdoor.py

@@ -2345,5 +2345,52 @@ def test_stalled_member_drop_does_not_wedge_the_group(self):
                                 "the surviving member must keep receiving (no wedged lock)")
 
 
+
+class TestResumeDechunk(unittest.TestCase):
+    """The resume path de-chunks the engine's chunked SSE body across
+    arbitrary read boundaries (83cd8afc follow-up). Group._dechunk is pulled
+    from the module source by AST — importing the module starts a server."""
+
+    @classmethod
+    def setUpClass(cls):
+        import ast, textwrap
+        src = open(os.path.join(os.path.dirname(os.path.abspath(__file__)), "hub-frontdoor.py")).read()
+        for node in ast.walk(ast.parse(src)):
+            if isinstance(node, ast.FunctionDef) and node.name == "_dechunk" and node.args.args and node.args.args[0].arg == "buf":
+                ns = {}
+                exec(textwrap.dedent(ast.get_source_segment(src, node)), ns)
+                cls.dechunk = staticmethod(ns["_dechunk"])
+                return
+        raise AssertionError("Group._dechunk not found")
+
+    @staticmethod
+    def chunked(*payloads):
+        return b"".join(b"%x\r\n%s\r\n" % (len(p), p) for p in payloads)
+
+    def feed(self, wire, cut):
+        st = {"mode": "size", "size": 0}
+        out, pend = b"", b""
+        for i in range(0, len(wire), cut):
+            o, pend = self.dechunk(pend + wire[i:i + cut], st)
+            out += o
+        return out, pend
+
+    def test_every_read_boundary_yields_the_exact_body(self):
+        frames = [b"id: 1\ndata: {\"a\":1}\n\n", b"id: 2\ndata: " + b"x" * 300 + b"\n\n", b": hb\n\n"]
+        wire = self.chunked(*frames)
+        for cut in range(1, len(wire) + 1):
+            out, _ = self.feed(wire, cut)
+            self.assertEqual(out, b"".join(frames), "read size %d" % cut)
+
+    def test_a_split_size_line_is_carried_not_dropped(self):
+        wire = self.chunked(b"data: one\n\n", b"data: two\n\n")
+        cut = wire.index(b"\r\n", 5) + 3  # inside the second chunk's size line
+        st = {"mode": "size", "size": 0}
+        a, pend = self.dechunk(wire[:cut], st)
+        self.assertEqual(a, b"data: one\n\n")
+        self.assertTrue(pend, "the partial size line must come back as the pending tail")
+        b, _ = self.dechunk(pend + wire[cut:], st)
+        self.assertEqual(b, b"data: two\n\n")
+
 if __name__ == "__main__":
     unittest.main(verbosity=2)