[REA-6854] Report each finished step to the runtime with complete_step()

[Dere-Wah]

Why

The runtime has no idea when a model finishes a unit of work. It needs that to count a session's steps, which the next PR closes the session on, and later to keep each step's result. The default step loop knows where a step ends; a model with its own run() knows where its unit of work ends, such as a built clip. Both need one way to say so.

What Changed

A model reports a finished step with complete_step():

from reactor_runtime import ReactorApp, StepCompleted

class ClipModel(ReactorApp):
    async def run(self) -> None:
        while True:
            await self.connected.wait()
            clip = await self.build_next_clip()
            await self.emit(clip.output)
            await self.complete_step(StepCompleted(
                output=clip.output,
                files={"last_frame.png": clip.last_frame_png},
                elapsed=clip.seconds,
            ))

StepCompleted carries the step's output, extra files to keep with it, an error when it failed, and how long it took. File names must be one plain name, result.json and output.mp4 are reserved for the runtime, and contents must be bytes. complete_step() sits next to send() on ReactorCore.

The default step loop reports every step that runs, after it emits the step's media. That includes a step that produced nothing (output=None) and a step whose error process_output() recovered from (reported without an error, since the application handled it). A refused step and a crash report nothing. ReactorPipeline's generator driver reports nothing.

A ReactorApp that wants to write the report itself, to keep extra files with a step or to mark a step it recovered from, returns a StepCompleted from process_output() instead of the Output. The loop emits its output and reports it as given, filling in elapsed with the measured generate() time when it is unset:

async def process_output(self, outcome: StepOutcome) -> WaypointOutput | StepCompleted | None:
    if isinstance(outcome.error, NoReference):
        await self.send(CommandError(command="generate", reason="No reference image."))
        return StepCompleted(error="no reference image")
    if outcome.error is not None:
        raise outcome.error
    return StepCompleted(
        output=WaypointOutput(main_video=outcome.result.frames),
        files={"prompt.txt": self.state.prompt.encode()},
    )

Calling complete_step() from inside process_output() would report the step twice, so the return value is the one way to replace the loop's report.

The report crosses the bridge as a neutral CompletedStep, with the output turned into a media bundle, through a step sink bound beside the media sink, so the runner never sees a model class. The runner numbers each step within its session and journals a new step_completed fact, {"step": 1, "saved": false, "error": null}. saved stays false until step results land. The event-vocabulary tests that pin the journal event list gain it on purpose.

[github-actions]

[codex-review] - [P1] src/reactor_runtime/runner/runner.py: Step limits are counted but never stop the session.

Scope: full (3061225..9af1f4c).

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 👈 (View in Graphite)
• [REA-6854] Apply a session's starting input, sent by a system client #233
• [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]

Reviewed step reporting through the bridge and the default-loop coverage for refused, empty, recovered, and failed steps. The reporting contract looks sound.

[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:18 PM UTC: Graphite rebased this pull request as part of a merge.
• Oct 5, 6:19 PM UTC: @Dere-Wah merged this pull request with Graphite.