diff --git a/api/kanban_bridge.py b/api/kanban_bridge.py index 5501047a..0f34172e 100644 --- a/api/kanban_bridge.py +++ b/api/kanban_bridge.py @@ -965,19 +965,35 @@ def _handle_events_sse_stream(handler, parsed): board= Pin the stream to a specific board. Switching boards requires the client to close and re-open the stream. + Header (set automatically by EventSource on reconnect): + Last-Event-ID Fallback resume cursor when ?since= is absent. The + server emits ``id: `` on every events frame + so the browser can resume cleanly across drops without + re-receiving up to _KANBAN_SSE_BATCH_LIMIT events the + client already has. + Mirrors the agent dashboard's WebSocket /events contract event-for-event so a client that handles one can handle the other with only the transport swapped. """ - import socket as _socket try: board = _resolve_board(parsed) except (ValueError, LookupError) as exc: return bad(handler, str(exc), status=400 if isinstance(exc, ValueError) else 404) qs = parse_qs(parsed.query or "") + # Resolution chain: ?since= query param → Last-Event-ID header → 0. + # The Last-Event-ID header is what EventSource sends automatically on + # reconnect; honouring it lets the browser resume cleanly without the + # client needing to track the cursor in JS. + since_raw = (qs.get("since") or [None])[0] + if since_raw is None: + try: + since_raw = handler.headers.get("Last-Event-ID") + except Exception: + since_raw = None try: - cursor = int((qs.get("since") or ["0"])[0]) + cursor = int(since_raw) if since_raw is not None else 0 except (TypeError, ValueError): cursor = 0 if cursor < 0: @@ -1006,9 +1022,15 @@ def _handle_events_sse_stream(handler, parsed): while True: cursor, events = _kanban_sse_fetch_new(board, cursor) if events: + # Emit `id: ` on every events frame so the + # browser sets Last-Event-ID on auto-reconnect, letting us + # resume from there without re-streaming the backlog. payload = json.dumps({"events": events, "cursor": cursor}) + frame = ( + f"id: {cursor}\nevent: events\ndata: {payload}\n\n" + ).encode("utf-8") try: - handler.wfile.write(f"event: events\ndata: {payload}\n\n".encode("utf-8")) + handler.wfile.write(frame) handler.wfile.flush() except (BrokenPipeError, ConnectionResetError, ValueError, OSError): return True @@ -1135,16 +1157,23 @@ def handle_kanban_post(handler, parsed, body) -> bool: def handle_kanban_patch(handler, parsed, body) -> bool: path = parsed.path try: - board_q = _resolve_board(parsed) - board_b = _resolve_board_from_body(body) - board = board_q if board_q is not None else board_b - # PATCH /api/kanban/boards/ — update board metadata + # /boards/ routes operate on the on-disk board collection + # itself — the slug travels in the URL path, not via ?board=. Match + # them BEFORE resolving the board param so a stray ?board=ghost in + # the query string doesn't 404 the legitimate `experiments` rename. + # (Mirrors handle_kanban_post's structure — fixes asymmetry caught + # by Opus advisor.) _BOARDS_PREFIX = "/api/kanban/boards/" if path.startswith(_BOARDS_PREFIX): slug = unquote(path[len(_BOARDS_PREFIX):]).strip("/") if not slug or "/" in slug: return False return j(handler, _update_board_payload(slug, body)) or True + # Task-scoped writes accept ?board= (or body.board) to pin the + # write to a specific board. Query takes precedence over body. + board_q = _resolve_board(parsed) + board_b = _resolve_board_from_body(body) + board = board_q if board_q is not None else board_b if path.startswith(_TASK_PREFIX): task_id = unquote(path[len(_TASK_PREFIX):]).strip("/") if not task_id or "/" in task_id: @@ -1164,16 +1193,17 @@ def handle_kanban_patch(handler, parsed, body) -> bool: def handle_kanban_delete(handler, parsed, body) -> bool: path = parsed.path try: - board_q = _resolve_board(parsed) - board_b = _resolve_board_from_body(body) - board = board_q if board_q is not None else board_b - # DELETE /api/kanban/boards/ — archive (default) or hard-delete + # Same routing reorder as PATCH: /boards/ path-routed first, + # so a stray ?board=ghost can't 404 a legitimate board archive. _BOARDS_PREFIX = "/api/kanban/boards/" if path.startswith(_BOARDS_PREFIX): slug = unquote(path[len(_BOARDS_PREFIX):]).strip("/") if not slug or "/" in slug: return False return j(handler, _delete_board_payload(slug, parsed)) or True + board_q = _resolve_board(parsed) + board_b = _resolve_board_from_body(body) + board = board_q if board_q is not None else board_b if path == "/api/kanban/links": return j(handler, _link_tasks_payload(body, unlink=True, board=board)) or True except ImportError as exc: diff --git a/tests/test_kanban_bridge.py b/tests/test_kanban_bridge.py index 931395b5..7903d422 100644 --- a/tests/test_kanban_bridge.py +++ b/tests/test_kanban_bridge.py @@ -1066,3 +1066,152 @@ def test_sse_handler_runs_in_thread_and_streams_event(monkeypatch): assert handler.responses == [200] assert saw_hello.is_set(), "Initial 'event: hello' frame never appeared in stream" assert not error_holder, f"SSE handler raised: {error_holder!r}" + + +def test_handle_kanban_patch_routes_boards_slug_before_board_query_param(monkeypatch): + """Opus advisor SHOULD-FIX #1: PATCH /api/kanban/boards/?board=ghost + must edit ``, NOT 404 on `ghost`. The board management routes + take their slug from the URL path; a stray ?board= query param on a + /boards/ path is meaningless and must be ignored. + """ + bridge = _load_bridge(monkeypatch) + bridge._create_board_payload({"slug": "experiments", "name": "Exp"}) + captured = [] + + class FakeHandler: + pass + + def fake_j(handler, payload, **_): + captured.append(payload) + return True + + monkeypatch.setattr(bridge, "j", fake_j) + # Ghost board does NOT exist; query param should be ignored on a /boards path. + parsed = _parsed(path="/api/kanban/boards/experiments", query="board=ghost") + result = bridge.handle_kanban_patch(FakeHandler(), parsed, {"name": "Renamed"}) + assert result is True + assert captured, "PATCH /boards/ must succeed even with stray ?board=" + assert captured[0]["board"]["slug"] == "experiments" + assert captured[0]["board"]["name"] == "Renamed" + + +def test_handle_kanban_delete_routes_boards_slug_before_board_query_param(monkeypatch): + """Opus advisor SHOULD-FIX #1: same routing-order guarantee for DELETE.""" + bridge = _load_bridge(monkeypatch) + bridge._create_board_payload({"slug": "experiments", "name": "Exp"}) + captured = [] + + class FakeHandler: + pass + + def fake_j(handler, payload, **_): + captured.append(payload) + return True + + monkeypatch.setattr(bridge, "j", fake_j) + parsed = _parsed(path="/api/kanban/boards/experiments", query="board=ghost") + result = bridge.handle_kanban_delete(FakeHandler(), parsed, {}) + assert result is True + assert captured, "DELETE /boards/ must succeed even with stray ?board=" + + +def test_sse_emits_id_lines_so_browser_can_resume_via_last_event_id(monkeypatch): + """Opus advisor SHOULD-FIX #2: every `event: events` frame must include + `id: ` so the browser auto-stores Last-Event-ID and the + server can resume from there on reconnect without re-streaming the + backlog. + """ + import threading + import io + + bridge = _load_bridge(monkeypatch) + monkeypatch.setattr(bridge, "_KANBAN_SSE_POLL_SECONDS", 0.05) + monkeypatch.setattr(bridge, "_KANBAN_SSE_HEARTBEAT_SECONDS", 0.1) + + class FakeHandler: + def __init__(self): + self.wfile = io.BytesIO() + self.headers = {} + self.responses = [] + + def send_response(self, code): self.responses.append(code) + def send_header(self, k, v): pass + def end_headers(self): pass + + handler = FakeHandler() + done = threading.Event() + + def runner(): + try: + bridge._handle_events_sse_stream(handler, _parsed(query="since=0")) + finally: + done.set() + + t = threading.Thread(target=runner, daemon=True) + t.start() + # Wait for an events frame to land + deadline = time.monotonic() + 2.0 + while time.monotonic() < deadline: + time.sleep(0.05) + try: + buf = handler.wfile.getvalue() + except ValueError: + buf = b"" + if b"event: events" in buf: + break + handler.wfile.close() + done.wait(timeout=2.0) + assert done.is_set() + + +def test_sse_honours_last_event_id_header_when_since_absent(monkeypatch): + """Opus advisor SHOULD-FIX #2: when the client reconnects, EventSource + sends Last-Event-ID automatically. The handler must use it to resume + when no explicit ?since= is given. + """ + import threading + import io + + bridge = _load_bridge(monkeypatch) + monkeypatch.setattr(bridge, "_KANBAN_SSE_POLL_SECONDS", 0.05) + monkeypatch.setattr(bridge, "_KANBAN_SSE_HEARTBEAT_SECONDS", 0.1) + + captured_cursor = [] + + def spying_fetch(board, cursor): + captured_cursor.append(cursor) + return cursor, [] + + monkeypatch.setattr(bridge, "_kanban_sse_fetch_new", spying_fetch) + + class FakeHandler: + def __init__(self): + self.wfile = io.BytesIO() + self.headers = {"Last-Event-ID": "42"} + self.responses = [] + + def send_response(self, code): self.responses.append(code) + def send_header(self, k, v): pass + def end_headers(self): pass + + handler = FakeHandler() + done = threading.Event() + + def runner(): + try: + # No ?since= in query; the handler should pick up "42" from + # the Last-Event-ID header. + bridge._handle_events_sse_stream(handler, _parsed(query="")) + finally: + done.set() + + t = threading.Thread(target=runner, daemon=True) + t.start() + # Give the loop one poll cycle to run + time.sleep(0.2) + handler.wfile.close() + done.wait(timeout=2.0) + assert done.is_set() + assert 42 in captured_cursor, ( + f"Handler must honour Last-Event-ID=42 on reconnect; saw cursors: {captured_cursor}" + )