feat: #835 Live Composition Increment 1 — stream the DM's scene (dark, WORLDOS_STREAM_BEATS=0) - #1043
Merged
Merged
Conversation
…t's written (dark, WORLDOS_STREAM_BEATS default OFF) Build the missing DATA PIPELINE for #835: the viewer's streaming/latestStreamed UI is already wired (screen-table.jsx DmNarratingBeat); this adds the pipe that feeds it. Everything is gated behind WORLDOS_STREAM_BEATS (default 0 = OFF), so this is safe to merge dark — when off, the live DM `claude -p` invocation is byte-identical to today (proven: the flag-array is empty and the splice expands to nothing; the tailer launcher returns early). Layer A — wrapper flag (the ONE shared lever in qa/lib_beat_driver.sh, so the three DM wrappers can't drift): * worldos_stream_flag_arg → WORLDOS_STREAM_FLAG=(--include-partial-messages) only when streaming is on, else empty. * worldos_stream_tailer_start/stop → launch/kill the per-attempt sidecar tailer against the DM $out (no-op when off; best-effort — a crash never affects a beat). * Wired into _dm_invoke in scripts/play.sh + scripts/play_party.sh, and the inline DM turn in qa/run_duo.sh (203-region only). Each attempt (incl. the retry, which re-mints $out) gets its own tailer. Layer B — scripts/stream_tailer.py (NEW, pure stdlib): tails the DM stream-json $out as it's written; on a content_block_start whose tool name suffix is log_event/persist_beat it marks a CANDIDATE block, accumulates input_json_delta partial_json, and incrementally decodes the `text` arg of narration/dialogue log_event calls into $STATE_DIR/stream/current.jsonl ({"seq","text","ts"} rows, truncated per beat). Handles: kind BEFORE or AFTER text (buffer-until-kind); partial_json split across chunk boundaries incl. mid-`\"` and mid-`\uXXXX`; non-prose kinds never streamed; MCP-prefixed names; nested vs flat row shapes. persist_beat is recognized but its nested-events text is deferred (Increment 1). Layer C — viewer/server.py: GET /beat-stream?since=N reads the wrapper-owned sidecar with a line cursor (modeled on _read_events), returns {"chunks","next","complete"}. Read-only — the engine's sole-writer invariant is untouched (the sidecar is not campaign state). Layer D — viewer/openworlds/app.jsx: useLiveSession gains a /beat-stream poll (~500ms, active only while a turn is pending) that accumulates chunks into ONE transient `composing:true` narration row and calls notePendingProgress(). The composing row is flagged (never dedup-keyed), so the canonical seq-keyed /events narration row REPLACES it on beat completion (the /events poll strips composing rows) and clearPending drops it on resolution — zero duplication. Tests — qa/test_stream_tailer.py (NEW, 18 tests, single-process -p no:xdist): the riskiest part (partial-JSON parsing) is property-tested over every chunk- boundary size + adversarial escape splits; non-prose suppression; ordering; MCP names; row shapes; the file-tailing driver. All green. Regression: existing JSX harness suites pass unchanged — test_live_narration_stream (38), test_chronicle_dedup_opening + test_dm_beat_wait_alive (25).
📝 WalkthroughWalkthroughAdds an end-to-end live composition streaming pipeline for WorldOS ChangesLive Composition Streaming Pipeline
Sequence Diagram(s)sequenceDiagram
participant Shell as Shell Driver<br>(play.sh / run_duo.sh)
participant Tailer as stream_tailer.py<br>(background sidecar)
participant Claude as claude -p<br>(DM invocation)
participant Server as viewer/server.py<br>/beat-stream
participant Viewer as app.jsx<br>useLiveSession
Shell->>Tailer: worldos_stream_tailer_start (background)
Shell->>Claude: worldos_timeout claude -p --include-partial-messages --output-format stream-json > out.jsonl
Claude-->>Tailer: stream-json rows appended to out.jsonl
Tailer->>Tailer: _iter_new_lines → StreamDecoder.feed_row()
Tailer->>Tailer: _PartialJsonScanner decodes log_event narration/dialogue deltas
Tailer->>Server: appends {seq,text,ts} to $STATE_DIR/stream/current.jsonl
Viewer->>Server: GET /beat-stream?since=N (every ~500ms while pending)
Server-->>Viewer: {chunks, next, complete}
Viewer->>Viewer: accumulate composingText, render composing:true chatBeats row
Claude-->>Shell: process exits
Shell->>Tailer: worldos_stream_tailer_stop (kill PID)
Viewer->>Server: GET /events (canonical narration arrives)
Server-->>Viewer: seq-keyed narration beats
Viewer->>Viewer: remove composing:true row, insert canonical beat
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Possibly related issues
Possibly related PRs
Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches📝 Generate docstrings
Comment |
![github-code-quality[bot]](https://avatars.githubusercontent.com/u/9919?s=60&v=4){kind=small}
| // /events narration row carries the finished prose, so the partial composing preview must vanish to | ||
| // avoid duplication. Filtering by the `composing` flag (not by text) keeps this independent of the | ||
| // dedup sets — the composing row was never claimed in seenSeq/seenText, so removing it is clean. | ||
| const clearComposing = React.useCallback(() => { |
| finally: | ||
| try: | ||
| sink.close() | ||
| except OSError: |
| poll = float(os.environ.get("WORLDOS_STREAM_POLL_S", "0.25")) | ||
| try: | ||
| tail_stream(out_path, stream_path, poll_interval=poll) | ||
| except KeyboardInterrupt: |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🤖 Prompt for all review comments with AI agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
Inline comments:
In `@scripts/stream_tailer.py`:
- Around line 389-390: In the function that opens the sink file at lines
389-390, `os.path.dirname(stream_path)` returns an empty string when stream_path
is a direct filename, which causes `os.makedirs("")` to raise an error. Before
calling os.makedirs, check if the directory path is not empty, and only create
the directories if there is an actual directory component to create. This
ensures direct filenames are handled correctly without attempting to create an
empty directory path.
In `@viewer/openworlds/app.jsx`:
- Around line 601-602: The beatStreamCursor.current is only being reset per run
(at lines 601-602) but needs to also be reset per turn/attempt in the polling
loop. Since the tailer truncates current.jsonl per beat/attempt, the stale
cursor value can skip the start of a new stream if the file regrows before the
next poll. In addition to the existing reset at the run level, reset
beatStreamCursor.current in the polling loop logic (around line 832) or wherever
turn/attempt transitions occur to ensure the cursor doesn't carry over stale
values across different attempts.
In `@viewer/server.py`:
- Around line 8473-8475: The `since` parameter parsing in the `/beat-stream`
endpoint lacks error handling when converting user input to an integer using
`int((qs.get("since") or ["0"])[0])`. Wrap this integer conversion in a
try-except block to catch ValueError exceptions that occur when the user
provides a non-numeric value like `since=abc`. When the conversion fails,
default to 0 (the same as the current fallback behavior) so the request degrades
safely instead of raising an unhandled exception. The default value of 0 is
already provided in the fallback logic, so ensure the error handling maintains
this same safe default behavior.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: 71e9207c-708b-4607-8478-153e137e405b
📒 Files selected for processing (8)
qa/lib_beat_driver.shqa/run_duo.shqa/test_stream_tailer.pyscripts/play.shscripts/play_party.shscripts/stream_tailer.pyviewer/openworlds/app.jsxviewer/server.py
Comment on lines
+389
to
+390
| os.makedirs(os.path.dirname(stream_path), exist_ok=True) | ||
| return open(stream_path, "w", encoding="utf-8") |
There was a problem hiding this comment.
Handle direct-file sink paths without crashing.
When stream_path is a direct filename (allowed by the CLI contract), os.path.dirname(stream_path) is empty. On Line 389, os.makedirs("") can raise and kill the sidecar before streaming starts.
🩹 Proposed fix
def _open_sink(stream_path: str):
"""Open (and TRUNCATE) the stream sink at beat start, returning the file handle. Truncating
here implements 'reset on each new beat/attempt': the wrapper launches a fresh tailer per
attempt, so opening with "w" clears any prior attempt's chunks."""
- os.makedirs(os.path.dirname(stream_path), exist_ok=True)
+ parent = os.path.dirname(stream_path)
+ if parent:
+ os.makedirs(parent, exist_ok=True)
return open(stream_path, "w", encoding="utf-8")📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| os.makedirs(os.path.dirname(stream_path), exist_ok=True) | |
| return open(stream_path, "w", encoding="utf-8") | |
| def _open_sink(stream_path: str): | |
| """Open (and TRUNCATE) the stream sink at beat start, returning the file handle. Truncating | |
| here implements 'reset on each new beat/attempt': the wrapper launches a fresh tailer per | |
| attempt, so opening with "w" clears any prior attempt's chunks.""" | |
| parent = os.path.dirname(stream_path) | |
| if parent: | |
| os.makedirs(parent, exist_ok=True) | |
| return open(stream_path, "w", encoding="utf-8") |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@scripts/stream_tailer.py` around lines 389 - 390, In the function that opens
the sink file at lines 389-390, `os.path.dirname(stream_path)` returns an empty
string when stream_path is a direct filename, which causes `os.makedirs("")` to
raise an error. Before calling os.makedirs, check if the directory path is not
empty, and only create the directories if there is an actual directory component
to create. This ensures direct filenames are handled correctly without
attempting to create an empty directory path.
Comment on lines
+601
to
+602
| beatStreamCursor.current = 0; // #835: reset the live-composition /beat-stream tail per run | ||
| composingText.current = ""; // #835: …and its per-beat prose accumulator |
There was a problem hiding this comment.
Reset /beat-stream cursor per turn/attempt, not only per run.
beatStreamCursor.current is reset on run changes (Lines 601-602) but reused across turns in the polling loop (Line 832). Because the tailer truncates current.jsonl per beat/attempt, a stale high cursor can skip the start of a new stream if the file regrows quickly before the next poll.
🩹 Proposed fix
const armPending = React.useCallback((text) => {
clearTimers();
+ // New beat/attempt starts with a freshly truncated stream sidecar.
+ beatStreamCursor.current = 0;
+ composingText.current = "";
// `#406`: "first beat?" = no turn has RESOLVED on /chat yet (resolvedTurnsRef), NOT "no paragraph
// has streamed" (dmBeatCountRef). So a retried cold-open — one paragraph streamed, then "Try
// again" before the turn resolved — still gets the generous PENDING_RECOVERY_FIRST_MS window.
const firstBeat = resolvedTurnsRef.current === 0;Also applies to: 808-833
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@viewer/openworlds/app.jsx` around lines 601 - 602, The
beatStreamCursor.current is only being reset per run (at lines 601-602) but
needs to also be reset per turn/attempt in the polling loop. Since the tailer
truncates current.jsonl per beat/attempt, the stale cursor value can skip the
start of a new stream if the file regrows before the next poll. In addition to
the existing reset at the run level, reset beatStreamCursor.current in the
polling loop logic (around line 832) or wherever turn/attempt transitions occur
to ensure the cursor doesn't carry over stale values across different attempts.
Comment on lines
+8473
to
+8475
| qs = parse_qs(parsed.query) | ||
| since = int((qs.get("since") or ["0"])[0]) | ||
| chunks, nxt, complete = _read_beat_stream(since) |
There was a problem hiding this comment.
Guard since parsing on /beat-stream to avoid request-time exceptions.
Line 8474 uses int(...) directly on user input. A malformed value (e.g. since=abc) can raise and fail this request path instead of degrading safely.
🩹 Proposed fix
elif route == "/beat-stream":
# `#835` Live Composition Increment 1 — poll the wrapper-owned live-stream sidecar
# ($STATE_DIR/stream/current.jsonl) with a line cursor. Read-only (the tailer is the
# sole writer of this sidecar; the engine's sole-writership of campaign state is
# untouched). Returns {"chunks":[{seq,text,ts}...], "next":N, "complete":bool}. The
# `campaign` query arg is accepted for symmetry with /events but the stream is a single
# per-run sidecar (one in-flight DM beat at a time), so it isn't campaign-scoped here.
qs = parse_qs(parsed.query)
- since = int((qs.get("since") or ["0"])[0])
+ try:
+ since = max(0, int((qs.get("since") or ["0"])[0]))
+ except (TypeError, ValueError):
+ since = 0
chunks, nxt, complete = _read_beat_stream(since)
self._json({"chunks": chunks, "next": nxt, "complete": complete})📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| qs = parse_qs(parsed.query) | |
| since = int((qs.get("since") or ["0"])[0]) | |
| chunks, nxt, complete = _read_beat_stream(since) | |
| qs = parse_qs(parsed.query) | |
| try: | |
| since = max(0, int((qs.get("since") or ["0"])[0])) | |
| except (TypeError, ValueError): | |
| since = 0 | |
| chunks, nxt, complete = _read_beat_stream(since) |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@viewer/server.py` around lines 8473 - 8475, The `since` parameter parsing in
the `/beat-stream` endpoint lacks error handling when converting user input to
an integer using `int((qs.get("since") or ["0"])[0])`. Wrap this integer
conversion in a try-except block to catch ValueError exceptions that occur when
the user provides a non-numeric value like `since=abc`. When the conversion
fails, default to 0 (the same as the current fallback behavior) so the request
degrades safely instead of raising an unhandled exception. The default value of
0 is already provided in the fallback logic, so ensure the error handling
maintains this same safe default behavior.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
KEYSTONE #835 Increment 1, default OFF (
WORLDOS_STREAM_BEATS=0→ byte-identical to today). The viewer'sstreaming/latestStreamedUI was already wired; this adds the missing data pipeline:--include-partial-messageson the DM call (sharedlib_beat_driver.shhelper) →scripts/stream_tailer.py(a robust partial-JSON lexer that extractslog_eventnarration deltas, handling chunk boundaries + kind-before/after-text + unicode escapes) →GET /beat-stream→ a React poll feedingcomposing:truerows that the existing dedup collapses on beat completion. 18 tailer unit tests + existing JSX tests green. persist_beat is recognized-but-not-streamed (Increment 1 scope).Summary by CodeRabbit
/beat-streamendpoint providing streamed content chunks with cursor-based pagination.