"""TDD — push must stream response bytes so Cloudflare doesn't 524-timeout. Root cause of the large-repo push failure (2026-04-28): _WireResponse buffered ALL wire_push_stream frames before sending a single HTTP response byte. CF's 120-second origin timeout fired during the silent commit-processing phase (827 commits × DB inserts ≈ 8+ minutes). Three tests codify the correct behaviour: T1 Static: _WireResponse must send http.response.start BEFORE it has consumed all frames from wire_push_stream. Sending headers early resets CF's 120-second "first byte" timer. T2 Integration: Push response body must be a sequence of msgpack objects: zero or more PROGRESS frames followed by one RESULT frame. (The old response was a single msgpack dict with no "t" field.) T3 Static: wire_push_stream must yield PROGRESS frames INSIDE the commit insertion loop — not just the single frame emitted before the entire loop. Without intra-loop heartbeats, CF still timeouts even after T1. """ from __future__ import annotations import inspect import textwrap import msgpack import pytest from httpx import AsyncClient from sqlalchemy.ext.asyncio import AsyncSession from muse.core.types import blob_id from muse.core.mpack import WIRE_CONTENT_TYPE, MuseWireFrameWriter from musehub.models.wire import ( SFRAME_COMMIT_PACK, SFRAME_END, SFRAME_ERROR, SFRAME_HEADER, SFRAME_OBJECT, SFRAME_PROGRESS, SFRAME_RESULT, ) from musehub.types.json_types import JSONObject, JSONValue, StrDict from tests.factories import create_repo _fw = MuseWireFrameWriter() # --------------------------------------------------------------------------- # Helpers # --------------------------------------------------------------------------- def _pack(obj: JSONValue) -> bytes: return msgpack.packb(obj, use_bin_type=True) def _wrap(ft: str, data: JSONValue) -> bytes: return _fw.wrap(frame_type=ft, payload=_pack(data)) def _oid(data: bytes) -> str: return blob_id(data) def _header_frame(n_objects: int = 0, n_commits: int = 1) -> bytes: return _wrap(SFRAME_HEADER, { "t": SFRAME_HEADER, "branch": "main", "force": False, "have": [], "head": _oid(b"head"), "n_objects": n_objects, "n_commits": n_commits, }) def _object_frame(raw: bytes, path: str = "file.py") -> tuple[str, bytes]: oid = _oid(raw) return oid, _wrap(SFRAME_OBJECT, { "t": SFRAME_OBJECT, "id": oid, "path": path, "enc": "raw", "content": raw, }) def _commit_pack_frame(commits: list[JSONObject], snapshots: list[JSONObject]) -> bytes: return _wrap(SFRAME_COMMIT_PACK, { "t": SFRAME_COMMIT_PACK, "commits": commits, "snapshots": snapshots, }) def _end_frame(n_objects: int = 0, n_commits: int = 1) -> bytes: return _wrap(SFRAME_END, {"t": SFRAME_END, "n_objects": n_objects, "n_commits": n_commits}) def _make_commit( snapshot_id: str, parent_id: str | None = None, suffix: str = "", ) -> JSONObject: return { "commit_id": _oid(f"commit-{snapshot_id}{suffix}".encode()), "parent_ids": [parent_id] if parent_id else [], "snapshot_id": snapshot_id, "branch": "main", "message": "timeout tdd commit", "author": "test-user-wire", "committed_at": "2026-04-28T00:00:00+00:00", "signature": "", "signer_key_id": "", "agent_id": "claude-code", "model_id": "claude-sonnet-4-6", "metadata": {}, } def _make_snapshot(snap_id: str, manifest: StrDict | None = None) -> JSONObject: return {"snapshot_id": snap_id, "manifest": manifest or {}} def _parse_response_frames(content: bytes) -> list[dict]: """Parse response body as a sequence of msgpack objects.""" unpacker = msgpack.Unpacker(raw=False) unpacker.feed(content) return list(unpacker) # --------------------------------------------------------------------------- # T1 — _WireResponse must send http.response.start BEFORE consuming all frames # --------------------------------------------------------------------------- def test_t1_wire_response_sends_headers_before_consuming_full_stream() -> None: """_WireResponse.__call__ must call send(http.response.start) before the async-for loop over wire_push_stream completes. The current broken pattern: async for frame in wire_push_stream(...): ... await send({"type": "http.response.start", ...}) # ← happens last The correct pattern: await send({"type": "http.response.start", ...}) # ← happens first async for frame in wire_push_stream(...): await send({"type": "http.response.body", "body": frame, "more_body": True}) await send({"type": "http.response.body", "body": b"", "more_body": False}) This test checks the source of push_stream for the correct ordering. """ from musehub.api.routes import wire source = inspect.getsource(wire.push_stream) # The response.start send must happen before any frame iteration. # In the correct implementation, http.response.start appears before the # async for loop; in the broken implementation it appears after. start_pos = source.find("http.response.start") loop_pos = source.find("async for frame_bytes in wire_push_stream") assert start_pos != -1, ( "push_stream source must contain 'http.response.start'" ) assert loop_pos != -1, ( "push_stream source must contain 'async for frame_bytes in wire_push_stream'" ) assert start_pos < loop_pos, ( "http.response.start must be sent BEFORE the async for loop over wire_push_stream. " "Currently the send happens after all frames are consumed — CF's 120-second timer " "fires during commit processing because no bytes reach CF until then. " "Fix: send http.response.start early, then stream P frames as body chunks." ) # --------------------------------------------------------------------------- # T2 — push response body is a stream of msgpack objects (P* then R) # --------------------------------------------------------------------------- @pytest.mark.asyncio async def test_t2_push_response_is_frame_stream( client: AsyncClient, db_session: AsyncSession, wire_headers: StrDict, ) -> None: """Push response body must be a sequence of msgpack objects ending in a RESULT frame, not a single opaque msgpack dict. The old response: one dict { "ok": True, "message": "...", ... } (no "t" key) The new response: N dicts with "t" == "P", then one dict with "t" == "R" This is what the protocol spec says ("The server responds with a matching frame stream: PROGRESS frames, ERROR frame or RESULT frame"). The streaming format lets CF see progress bytes and avoids the 524 timeout. """ repo = await create_repo(db_session, owner="test-user-wire", name="t2-frame-stream") snap_id = _oid(b"snap-t2") commit = _make_commit(snap_id) snap = _make_snapshot(snap_id) body = ( _header_frame(n_objects=0, n_commits=1) + _commit_pack_frame([commit], [snap]) + _end_frame(n_objects=0, n_commits=1) ) resp = await client.post( f"/{repo.owner}/{repo.slug}/push/stream", content=body, headers={**wire_headers, "Content-Type": WIRE_CONTENT_TYPE}, ) assert resp.status_code == 200, f"Unexpected status {resp.status_code}: {resp.text[:200]}" frames = _parse_response_frames(resp.content) assert frames, "Response body must contain at least one msgpack object" last = frames[-1] assert last.get("t") == SFRAME_RESULT, ( f"Last frame must be RESULT (t='R'). Got: {last}. " f"All frames: {frames}. " "Response must be a msgpack frame stream (P* then R), not a single opaque dict." ) assert last.get("ok") is True, f"RESULT frame must have ok=True. Got: {last}" for frame in frames[:-1]: t = frame.get("t") assert t == SFRAME_PROGRESS, ( f"All frames before the last must be PROGRESS (t='P'). Got t={t!r}: {frame}" ) # --------------------------------------------------------------------------- # T3 — wire_push_stream yields P frames INSIDE the commit loop # --------------------------------------------------------------------------- def test_t3_wire_push_stream_emits_progress_during_commit_loop() -> None: """wire_push_stream must yield PROGRESS frames INSIDE the commit insertion loop, not just the single frame emitted before the entire phase starts. Without intra-loop heartbeats, CF still times out on large repos even if T1 is fixed: the single P frame before 827 commits is sent immediately, then CF waits 120+ seconds for the next byte while all rows are inserted. This test checks the source of musehub_wire.wire_push_stream for a yield _prog(...) call that appears INSIDE the ordered_commits iteration. """ from musehub.services import musehub_wire source = inspect.getsource(musehub_wire.wire_push_stream) # Find the commit iteration loop (enumerate or plain for). loop_marker = "ordered_commits" loop_pos = -1 for candidate in ("for _i, wire_commit in enumerate(ordered_commits)", "for wire_commit in ordered_commits"): loop_pos = source.find(candidate) if loop_pos != -1: break assert loop_pos != -1, ( "wire_push_stream must iterate over ordered_commits. " "Expected to find 'for wire_commit in ordered_commits' or " "'for _i, wire_commit in enumerate(ordered_commits)'" ) # Find any yield _prog(...) that appears AFTER the loop start. tail = source[loop_pos:] assert "yield _prog(" in tail, ( "wire_push_stream must yield at least one PROGRESS frame INSIDE the " "'for wire_commit in ordered_commits' loop. " "Without intra-loop heartbeats, a push of 800+ commits is silent for " ">120 seconds and CF kills the connection with 524. " "Add: yield _prog(f'committing {i}/{n}…') every ~100 commits." )