Journal an initializing event during model load

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()

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)

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,

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=<seq>``, 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"]
 
 

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),

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()

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)

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),