def test_sse_stream_first_events(self, api, db, db_engine):
        """SSE contract end-to-end. starlette's TestClient runs the whole ASGI app
        to completion before returning the response, so an endless stream would
        deadlock the client. A background thread marks the campaign completed
        mid-stream: the generator then emits connected -> progress -> keepalive(s)
        -> progress -> done and terminates on its own."""
        from app.models import Campaign
        from sqlalchemy.orm import Session as SyncSess
        import threading, time as _time

        f = _seed_campaign_fixture(db, user_id_of(api), n_recipients=2)
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="SSE")).json()["id"]
        r = api.post(f"/campaigns/{cid}/start", headers=H(api))
        assert r.status_code == 200, r.text

        def flip_to_completed_after(seconds: float):
            _time.sleep(seconds)
            with SyncSess(db_engine, expire_on_commit=False) as s:
                c = s.get(Campaign, cid)
                if c is not None:
                    c.status = "completed"
                    s.commit()

        th = threading.Thread(target=flip_to_completed_after, args=(3.0,), daemon=True)
        th.start()

        frames = []
        t0 = _time.monotonic()
        with api.stream("GET", f"/campaigns/{cid}/stream") as resp:
            assert resp.status_code == 200
            assert resp.headers["content-type"].startswith("text/event-stream")
            for line in resp.iter_lines():
                frames.append(line)
                assert _time.monotonic() - t0 < 30, frames  # safety net
        th.join(timeout=5)
        assert any("event: connected" in fr for fr in frames), frames
        assert any("event: progress" in fr for fr in frames), frames
        assert any("event: done" in fr for fr in frames), frames
        # the done frame carries the live counter payload
        done_idx = next(i for i, fr in enumerate(frames) if "event: done" in fr)
        done_data = frames[done_idx + 1]
        assert done_data.startswith("data: "), done_data
        import json as _json
        payload = _json.loads(done_data[len("data: "):])
        assert payload["status"] == "completed"
        assert payload["total"] == 2

    def test_campaign_isolation(self, api, db, second_user):
        f = _seed_campaign_fixture(db, user_id_of(api))
        cid = api.post("/campaigns/create", headers=H(api),
                       json=_cIn(f, name="Mine")).json()["id"]
        assert second_user.get(f"/campaigns/{cid}").status_code == 404
        r = second_user.post(f"/campaigns/{cid}/start", headers=H(second_user))
        assert r.status_code == 404