[REA-6854] Apply a session's starting input, sent by a system client

[Dere-Wah]

Why

A caller that starts a session without a client still needs the model set up: the prompt, the seed, an image to start from. Those arrive as starting_input on the start request, and they should go through the same path a client's commands take, so validation, upload resolution, and the journal behave the same and the model needs no change.

Two things stood in the way. A handler that takes client had no client to receive. And a model only generates while someone is connected: the default step loop waits for a client, and hand-written run() loops gate on self.connected. Both are solved by giving such a session a client the runtime owns.

That client must not change what the session state means. streaming is the runtime's occupancy signal: it says a client is being served, and anything that reads /events uses the state pair on each connection move to tell when a session gained or lost its clients. A live session is not served while the runtime applies its starting input, so the runtime's own client must not make it look occupied.

What Changed

The starting input. StartingInput.as_commands() turns each state key into the set_<key> command the model's state generates, followed by commands in order. Right after the session starts, the runner submits them one at a time through _submit_command. Each command's arguments were already checked against the contract when the session started (#232), so what can still fail here is what only running a command shows, such as an upload whose bytes never arrive: that command is journalled as an error and the rest still run. A command that names an upload waits up to the orphan timeout for its bytes, since the caller can only seed them once the session exists. Client commands wait until the list is submitted, the list runs once per session, and it stops if the session ends. The descriptor reports "starting_input": {"applied": N}, the number of commands the list expanded to.

The system client. runner/system_client.py adds SystemConnection, a connection with no wire on connection id 0, which no transport mints. It advertises no media, so the media fan-out skips it and a model is never held to a playout rate nobody is watching. Anything sent to it is dropped. The runner registers it for a session that has a starting input or steps, right after the start, so the model sees SessionStarted, then ClientConnected(system=True), then the starting commands, which it sends. It leaves once the list is submitted, unless the session has steps; then it stays until teardown. Because it is a real connection, self.connected is set while it is there, so existing models generate with no change.

The system client occupies only a session with steps. In a session with steps it is the session's only client, so it counts like any other connection: the session streams, and the orphan timeout cannot close it before its steps are done. In a session without steps it only sends the starting commands, so it does not count. SessionStateMachine gains a second entry point for that case. It applies a connection open or close as a self-loop on the current state and leaves the live count alone:

sm.send(SessionEvent.CONNECTION_OPENED, conn_id=1002)                      # counts: waiting -> streaming
sm.send_without_occupancy(SessionEvent.CONNECTION_OPENED, conn_id=0)       # does not: waiting -> waiting

ConnectionManager.register(conn, system=True, occupies=False) records the connection as not occupying the session, so its later drop() closes it the same way it opened. A client that joins while the system client is still applying the list opens the session as usual (waiting -> streaming), the system client's close is then a streaming -> streaming self-loop, and the client's own close leaves the session orphaned. Because a session without steps stays waiting, its orphan timer keeps running from the start, as it does for any session no client has joined yet.

The journal for a live session with a starting input:

start_session      ready -> waiting
connection_opened  waiting -> waiting     {"conn_id": 0, "system": true}
command            waiting                {"name": "set_spin_speed", "args": {"spin_speed": 2.0}, "conn_id": 0}
connection_closed  waiting -> waiting     {"conn_id": 0, "system": true}
connection_opened  waiting -> streaming   {"conn_id": 1002}    (the first client of its own)

And for a session with steps, where the system client holds the session until it ends:

start_session      ready -> waiting
connection_opened  waiting -> streaming   {"conn_id": 0, "system": true}
command            streaming              {"name": "set_spin_speed", "args": {"spin_speed": 2.0}, "conn_id": 0}

A session with steps answers /start_session with "state": "streaming", because the system client has already connected. A session with only a starting input, and an older body with neither key, answer "waiting".

The first step waits for the starting input. Without a viewer the system client paces nothing, so the default loop steps as fast as the model renders, and it began stepping the moment the system client connected, ahead of the starting commands. On examples/starter, a session with steps: 40 finished every step and closed in the 200 ms before the caller seeded the image its starting input named; the upload was then refused and the command rejected. So SessionStarted now says whether a starting input follows, and the runner posts an internal StartingInputApplied after the last starting command, on the same ordered queue, once any upload it names has resolved or timed out. Until it arrives the loop's gate stays shut, though self.connected is set and starting handlers still receive the system client as client. ReactorPipeline's loop waits on the same gate; a model with its own run() is unaffected. In a session without steps the system client leaves before the event, so a live session still takes its first step only once a viewer joins. SessionStarted.starting_input defaults to False, so anything else that posts it keeps today's behaviour.

Rerun on the starter with the fix: no step ran before the image arrived, every saved step showed the uploaded image, and a step's state carried the starting spin_speed rather than the default.

The system client's connection moves carry "system": true in the journal detail; other connections' detail is unchanged. The runtime's own metrics leave it out: it does not count in the connection series or in runtime_session_time_to_first_client_seconds. A model that needs to tell it apart reads the new ClientInfo.system, which ClientConnected carries.

Checked against examples/starter: a starting state of {spin_speed: 2.0, paused: true} and a set_static_interval of 4 left the model's live state at those values before any client connected, a 999 was rejected by the field's le=300 and journalled, and the next session started from the defaults.

[github-actions]

[codex-review] - [P1] src/reactor_runtime/runner/runner.py: steps is never enforced, so bounded sessions generate indefinitely.

Scope: full (d2fe770..3061225).

View workflow run.

[Dere-Wah]

• [REA-6855] Serve saved step folders over HTTP #239
• [REA-6855] Save each reported step as a folder of files #238
• [REA-6855] Write one step's output as a single MP4 file #237
• [REA-6855] Read the runtime.step_results block from the manifest #236
• [REA-6854] Close a session when it reaches its steps #235
• [REA-6854] Report each finished step to the runtime with complete_step() #234
• [REA-6854] Apply a session's starting input, sent by a system client #233 👈 (View in Graphite)
• [REA-6854] Read starting_input and steps from the start body #232
• [REA-6854] Deliver commands and reactor events to the model in posted order #231
• main

This stack of pull requests is managed by Graphite. Learn more about stacking.

[tempusfrangit]

Requesting changes because generation can start before the starting input is applied. With a delayed starting upload and steps: 1 in the full stack, I reproduced generation on default state followed by session closure before the upload command ran. Please gate generation until setup is applied and add a delayed-upload regression test.

[Dere-Wah]

@tempusfrangit thanks, reproduced on examples/starter too (a steps: 40 session finished before its image arrived). Fixed in 2bb381b: generation is gated until the starting input has landed, via an internal StartingInputApplied posted after the last starting command on the ordered queue; the default loop and ReactorPipeline wait for it, self.connected and own-run() models are unaffected. Delayed-upload regression tests are here (runner + model side) and end to end in #235 (8c05792). Details on the inline thread.

code was addressed

[Dere-Wah]

Merge activity

• Oct 5, 6:07 PM UTC: A user started a stack merge that includes this pull request via Graphite.
• Oct 5, 6:15 PM UTC: Graphite rebased this pull request as part of a merge.
• Oct 5, 6:16 PM UTC: @Dere-Wah merged this pull request with Graphite.