From f13658e92ddc13bbb7383030d52db53c122d7450 Mon Sep 17 00:00:00 2001 From: John Sabath Date: Thu, 27 Aug 2026 15:39:06 -0400 Subject: [PATCH] Journal an initializing event during model load MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The runner emits nothing on /events between boot and initialization_success, so the model-load window reads as silence — a consumer replaying the journal cannot tell a loading runtime from a dead one. Add a SessionEvent.INITIALIZING self-loop, legal only in CREATED, sent once at boot before the (blocking) load. It changes no state and touches no side effect (the bridge is not built yet), so a consumer replaying /events observes the loading phase directly, ahead of initialization_success. Signed-off-by: John Sabath --- src/reactor_runtime/core/session.py | 7 +++++ src/reactor_runtime/runner/runner.py | 10 ++++-- src/reactor_runtime/runner/state_machine.py | 5 +++ tests/contract/test_events_sse.py | 32 +++++++++++++------- tests/contract/test_transition_vocabulary.py | 15 +++++++++ tests/unit/http/test_routes.py | 18 ++++++++--- tests/unit/runner/test_runner.py | 31 +++++++++++++++++++ tests/unit/runner/test_state_machine.py | 3 ++ 8 files changed, 104 insertions(+), 17 deletions(-) diff --git a/src/reactor_runtime/core/session.py b/src/reactor_runtime/core/session.py index b99b54a0..ff7542ad 100644 --- a/src/reactor_runtime/core/session.py +++ b/src/reactor_runtime/core/session.py @@ -69,6 +69,12 @@ class SessionEvent(Enum): ``CLOSING``, and carries an :class:`~reactor_runtime.core.model.EndReason` in ``detail.reason`` (and, for a crash, the error) so a consumer learns why. + ``INITIALIZING`` records that the runtime is still loading its weights. It is + a self-loop legal only in ``CREATED``, emitted once at boot before the load + blocks, so a consumer replaying the journal observes the loading phase — the + one phase otherwise silent, since the runner emits nothing until it leaves + ``CREATED`` on ``INITIALIZATION_SUCCESS``. + ``CHUNK_READY``, ``CLIP_READY``, ``COMMAND``, ``ERROR``, and ``METRIC`` are the journal-only events (:data:`JOURNAL_EVENTS`): facts recorded for an external consumer rather than moves of the lifecycle. Each is a pure @@ -81,6 +87,7 @@ class SessionEvent(Enum): INITIALIZATION_SUCCESS = auto() INITIALIZATION_FAIL = auto() + INITIALIZING = auto() START_SESSION = auto() STOP_SESSION = auto() TIMEOUT = auto() diff --git a/src/reactor_runtime/runner/runner.py b/src/reactor_runtime/runner/runner.py index 95cac16f..de2fde25 100644 --- a/src/reactor_runtime/runner/runner.py +++ b/src/reactor_runtime/runner/runner.py @@ -237,11 +237,17 @@ async def start(self) -> None: The model load runs off the event loop (it may block while it reads weights), so the HTTP surface — already up by the time this runs — stays - responsive throughout, and a client subscribed to ``/events`` observes - the ``initialization_success``/``initialization_fail`` transition live. + responsive throughout: a client subscribed to ``/events`` observes the + ``initializing`` self-loop journalled before the load, then the + ``initialization_success``/``initialization_fail`` transition when it ends. """ self._loop = asyncio.get_running_loop() logger.info("loading model", model=self._cfg.model_ref) + # Journal the loading phase before the (blocking) load, so a consumer + # replaying /events sees the runtime is initializing during the load + # window rather than nothing until READY. A self-loop on CREATED: no + # state change, no side effect (the bridge is not built yet). + self._sm.send(SessionEvent.INITIALIZING) started_at = time.monotonic() try: model_cls = import_model_class(self._cfg.model_ref) diff --git a/src/reactor_runtime/runner/state_machine.py b/src/reactor_runtime/runner/state_machine.py index cacc4489..8e4f5df1 100644 --- a/src/reactor_runtime/runner/state_machine.py +++ b/src/reactor_runtime/runner/state_machine.py @@ -42,6 +42,11 @@ _TRANSITIONS: dict[SessionEvent, dict[SessionState, SessionState]] = { SessionEvent.INITIALIZATION_SUCCESS: {SessionState.CREATED: SessionState.READY}, SessionEvent.INITIALIZATION_FAIL: {SessionState.CREATED: SessionState.TERMINATED}, + # INITIALIZING is the runtime's "still loading" fact: a self-loop legal only + # in CREATED, emitted once at boot so a journal consumer sees the loading + # phase that is otherwise silent until INITIALIZATION_SUCCESS leaves CREATED. + # It changes no state and leaves the connection count alone (see _update_count). + SessionEvent.INITIALIZING: {SessionState.CREATED: SessionState.CREATED}, SessionEvent.START_SESSION: {SessionState.READY: SessionState.WAITING}, SessionEvent.STOP_SESSION: { SessionState.STREAMING: SessionState.CLOSING, diff --git a/tests/contract/test_events_sse.py b/tests/contract/test_events_sse.py index 3e03e57c..fffd0573 100644 --- a/tests/contract/test_events_sse.py +++ b/tests/contract/test_events_sse.py @@ -3,7 +3,8 @@ A consumer holds one long-lived subscription to ``GET /events``, parses only the ``id:`` and ``data:`` SSE fields, resumes with ``?since=``, and opens at ``since=0`` on a cold start expecting the retained backlog — including the -``initialization_success`` journalled before it connected — to replay in order. +``initializing`` and ``initialization_success`` journalled before it connected — +to replay in order. """ from __future__ import annotations @@ -23,15 +24,22 @@ async def _run_full_session(harness: Harness) -> None: async def test_a_cold_consumer_replays_initialization_from_since_zero(harness: Harness) -> None: - frames = await read_sse(harness.app, "/events?since=0", count=1) + frames = await read_sse(harness.app, "/events?since=0", count=2) assert frames[0].seq == 1 - payload = frames[0].payload - assert payload["type"] == "transition" - assert payload["event"] == "initialization_success" - assert payload["from"] == "created" - assert payload["to"] == "ready" - assert payload["detail"] == {} + boot = frames[0].payload + assert boot["type"] == "transition" + assert boot["event"] == "initializing" + assert boot["from"] == "created" + assert boot["to"] == "created" + assert boot["detail"] == {} + + assert frames[1].seq == 2 + ready = frames[1].payload + assert ready["event"] == "initialization_success" + assert ready["from"] == "created" + assert ready["to"] == "ready" + assert ready["detail"] == {} async def test_sequence_ids_are_contiguous_from_one(harness: Harness) -> None: @@ -63,15 +71,17 @@ async def test_every_envelope_is_a_transition_with_the_locked_keys(harness: Harn async def test_the_full_lifecycle_replays_in_order(harness: Harness) -> None: await _run_full_session(harness) - frames = await read_sse(harness.app, "/events?since=0", count=4) + frames = await read_sse(harness.app, "/events?since=0", count=5) assert [frame.payload["event"] for frame in frames] == [ + "initializing", "initialization_success", "start_session", "stop_session", "cleanup_complete", ] assert [(frame.payload["from"], frame.payload["to"]) for frame in frames] == [ + ("created", "created"), ("created", "ready"), ("ready", "waiting"), ("waiting", "closing"), @@ -82,9 +92,9 @@ async def test_the_full_lifecycle_replays_in_order(harness: Harness) -> None: async def test_since_resumes_strictly_after_the_given_sequence(harness: Harness) -> None: await _run_full_session(harness) - frames = await read_sse(harness.app, "/events?since=2", count=2) + frames = await read_sse(harness.app, "/events?since=3", count=2) - assert [frame.seq for frame in frames] == [3, 4] + assert [frame.seq for frame in frames] == [4, 5] assert [frame.payload["event"] for frame in frames] == ["stop_session", "cleanup_complete"] diff --git a/tests/contract/test_transition_vocabulary.py b/tests/contract/test_transition_vocabulary.py index 7f21d928..373bba15 100644 --- a/tests/contract/test_transition_vocabulary.py +++ b/tests/contract/test_transition_vocabulary.py @@ -31,6 +31,7 @@ { "initialization_success", "initialization_fail", + "initializing", "start_session", "stop_session", "timeout", @@ -118,6 +119,19 @@ def test_boot_edges() -> None: ) +def test_initializing_self_loops_in_created() -> None: + # The "still loading" fact: a self-loop journalled once at boot before the + # load blocks, so a cold consumer sees the loading phase ahead of the + # initialization_success that leaves created. + payload = _apply(_machine("created"), SessionEvent.INITIALIZING) + assert payload is not None + assert (payload["event"], payload["from"], payload["to"]) == ( + "initializing", + "created", + "created", + ) + + def test_session_open_edge() -> None: payload = _apply(_machine("ready"), SessionEvent.START_SESSION) assert payload is not None @@ -235,6 +249,7 @@ def test_journal_facts_self_loop_in_every_state(state: str, fact: SessionEvent) ("state", "event"), [ ("ready", SessionEvent.INITIALIZATION_SUCCESS), + ("ready", SessionEvent.INITIALIZING), ("waiting", SessionEvent.START_SESSION), ("ready", SessionEvent.STOP_SESSION), ("ready", SessionEvent.TIMEOUT), diff --git a/tests/unit/http/test_routes.py b/tests/unit/http/test_routes.py index c36a0db2..4d0ce11f 100644 --- a/tests/unit/http/test_routes.py +++ b/tests/unit/http/test_routes.py @@ -547,14 +547,24 @@ async def test_events_replays_the_backlog_as_sse( client: tuple[httpx.AsyncClient, Runner], ) -> None: # The egress stream is unbounded, so drive the generator directly and read - # the first replayed message rather than consuming an endless HTTP body. + # the replayed messages rather than consuming an endless HTTP body. _, runner = client stream = _stream_events(runner, 0) try: - message = await asyncio.wait_for(anext(stream), timeout=1.0) - assert message.startswith("id: 1\n") - body = json.loads(message.split("data: ", 1)[1].strip()) + # The loading self-loop is the first journalled fact — emitted on CREATED + # before the model finishes loading — so a consumer sees the pod is + # initializing during the load window. + first = await asyncio.wait_for(anext(stream), timeout=1.0) + assert first.startswith("id: 1\n") + body = json.loads(first.split("data: ", 1)[1].strip()) assert body["type"] == "transition" + assert body["event"] == "initializing" + assert body["to"] == "created" + + # The readiness transition follows once the load completes. + second = await asyncio.wait_for(anext(stream), timeout=1.0) + assert second.startswith("id: 2\n") + body = json.loads(second.split("data: ", 1)[1].strip()) assert body["to"] == "ready" finally: await stream.aclose() diff --git a/tests/unit/runner/test_runner.py b/tests/unit/runner/test_runner.py index 411df2de..1632b305 100644 --- a/tests/unit/runner/test_runner.py +++ b/tests/unit/runner/test_runner.py @@ -215,6 +215,37 @@ async def test_start_resolves_loads_and_readies(monkeypatch: pytest.MonkeyPatch) await runner.stop() +async def test_start_journals_initializing_before_ready(monkeypatch: pytest.MonkeyPatch) -> None: + # The loading phase is otherwise silent: a consumer subscribed from the + # start of the journal must see an INITIALIZING self-loop on CREATED before + # the INITIALIZATION_SUCCESS that leaves CREATED, so it can report a booting + # pod during the load window rather than nothing until READY. + created_models.clear() + monkeypatch.setattr("reactor_runtime.runner.runner.import_model_class", lambda ref: FakeModel) + runner = _runner() + stream = runner._events.subscribe(since=0) + + await runner.start() + try: + + async def first(n: int) -> list[Transition]: + out: list[Transition] = [] + async for _seq, event in stream: + out.append(event.transition) + if len(out) >= n: + break + return out + + boot = await asyncio.wait_for(first(2), timeout=2) + assert boot[0].event is SessionEvent.INITIALIZING + assert boot[0].from_state is SessionState.CREATED + assert boot[0].to_state is SessionState.CREATED + assert boot[1].event is SessionEvent.INITIALIZATION_SUCCESS + assert boot[1].to_state is SessionState.READY + finally: + await runner.stop() + + async def test_start_binds_outbound_before_spawn(monkeypatch: pytest.MonkeyPatch) -> None: created_models.clear() monkeypatch.setattr("reactor_runtime.runner.runner.import_model_class", lambda ref: FakeModel) diff --git a/tests/unit/runner/test_state_machine.py b/tests/unit/runner/test_state_machine.py index 1050fc14..cb8c8081 100644 --- a/tests/unit/runner/test_state_machine.py +++ b/tests/unit/runner/test_state_machine.py @@ -11,6 +11,9 @@ LEGAL_EDGES: list[tuple[SessionState, SessionEvent, SessionState]] = [ (SessionState.CREATED, SessionEvent.INITIALIZATION_SUCCESS, SessionState.READY), (SessionState.CREATED, SessionEvent.INITIALIZATION_FAIL, SessionState.TERMINATED), + # INITIALIZING is the loading self-loop: legal only in CREATED, changing no + # state, and rejected everywhere else. + (SessionState.CREATED, SessionEvent.INITIALIZING, SessionState.CREATED), (SessionState.READY, SessionEvent.START_SESSION, SessionState.WAITING), # Eviction is terminal from every live state — an idle eviction or a crash. (SessionState.CREATED, SessionEvent.EVICTION, SessionState.TERMINATED),