diff --git a/qa/lib_beat_driver.sh b/qa/lib_beat_driver.sh
index 648218e3..ef33493b 100644
--- a/qa/lib_beat_driver.sh
+++ b/qa/lib_beat_driver.sh
@@ -696,6 +696,58 @@ worldos_dm_timeout() {
   fi
 }
 
+# LIVE COMPOSITION (#835 Increment 1) — the ONE shared implementation of the stream-beats lever,
+# so the three DM wrappers (scripts/play.sh, scripts/play_party.sh, qa/run_duo.sh) can't drift.
+#
+# GATE: everything here is behind WORLDOS_STREAM_BEATS (worldos_env, default 0 = OFF). When OFF,
+# the flag-arg array stays EMPTY and the launcher/killer are no-ops, so the live DM `claude -p`
+# invocation is BYTE-IDENTICAL to today (the `${WORLDOS_STREAM_FLAG[@]+...}` splice expands to
+# nothing). The owner flips WORLDOS_STREAM_BEATS=1 after validating this dark PR.
+#
+# When ON: the wrapper adds `--include-partial-messages` to the DM stream-json call (so the
+# per-attempt $out jsonl carries the raw API stream events as the model generates), and launches
+# scripts/stream_tailer.py against that $out BEFORE the call — the tailer decodes the player-facing
+# scene out of the DM's streaming log_event tool-arg and writes chunks to $STATE_DIR/stream/
+# current.jsonl, which the viewer polls (/beat-stream). The tailer is a SIDECAR: if it crashes the
+# beat is unaffected (the canonical /events + /chat paths still resolve it); the wrapper kills it on
+# beat end. The flag is read once into the array so every call site uses the SAME splice form.
+
+# Build WORLDOS_STREAM_FLAG: (--include-partial-messages) when streaming is ON, else empty. Spliced
+# into the DM argv via ${WORLDOS_STREAM_FLAG[@]+"${WORLDOS_STREAM_FLAG[@]}"} (set -u safe; empty
+# array expands to nothing → today's exact argv when OFF).
+worldos_stream_flag_arg() {
+  WORLDOS_STREAM_FLAG=()
+  [ "$(worldos_env STREAM_BEATS 0)" = "1" ] && WORLDOS_STREAM_FLAG=(--include-partial-messages)
+}
+
+# Launch the per-attempt stream tailer against the DM $out file (no-op when streaming is OFF). The
+# tailer is started in the BACKGROUND; its PID is captured in WORLDOS_STREAM_TAILER_PID for the
+# killer. Best-effort: a launch failure (missing python3 / missing script) never fails the beat —
+# the live stream simply doesn't appear and the canonical paths resolve normally.
+#   $1 = the DM $out stream-json path   $2 = $STATE_DIR
+worldos_stream_tailer_start() {
+  WORLDOS_STREAM_TAILER_PID=""
+  [ "$(worldos_env STREAM_BEATS 0)" = "1" ] || return 0
+  local out="$1" state_dir="$2"
+  [ -n "$out" ] && [ -n "$state_dir" ] || return 0
+  local script="${WORLDOS_STREAM_TAILER:-$ROOT/scripts/stream_tailer.py}"
+  [ -f "$script" ] || return 0
+  command -v python3 >/dev/null 2>&1 || return 0
+  python3 "$script" "$out" "$state_dir/stream" >/dev/null 2>&1 &
+  WORLDOS_STREAM_TAILER_PID="$!"
+  return 0
+}
+
+# Kill the tailer launched by worldos_stream_tailer_start (no-op when none ran). Idempotent and
+# best-effort — a tailer that already exited is a benign no-op. Clears the PID after.
+worldos_stream_tailer_stop() {
+  local pid="${WORLDOS_STREAM_TAILER_PID:-}"
+  [ -n "$pid" ] || return 0
+  kill "$pid" >/dev/null 2>&1 || true
+  WORLDOS_STREAM_TAILER_PID=""
+  return 0
+}
+
 # RETRY DEADLINE (F12-1's second half): both dm_turn paths captured `beat_timeout` ONCE and the
 # ONE retry re-invoked with the SAME deadline verbatim — so a healthy-but-long beat that tripped
 # the routine deadline was killed AGAIN at the same mark (two kills, zero narration). Attempt 2
diff --git a/qa/run_duo.sh b/qa/run_duo.sh
index c0b9acef..02a7faf7 100755
--- a/qa/run_duo.sh
+++ b/qa/run_duo.sh
@@ -199,11 +199,20 @@ turn() {
     # → no result event → worldos_dm_final_text echoes empty → turn_retry's empty-output retry fires.
     # Player turn stays unbounded (it is a fast facade turn and was never the hang source).
     local beat_timeout; beat_timeout="$(worldos_dm_timeout "$first")"
+    # #835 Increment 1 — Live Composition flag (default OFF behind WORLDOS_STREAM_BEATS). Shared
+    # helpers (qa/lib_beat_driver.sh) build WORLDOS_STREAM_FLAG = (--include-partial-messages) ONLY
+    # when streaming is on; off → empty array → the splice expands to nothing → byte-identical to
+    # today. The stream tailer is launched against $out before the call and killed after (no-op when
+    # off); it is a sidecar (a crash never affects the beat).
+    worldos_stream_flag_arg
+    worldos_stream_tailer_start "$out" "$STATE_DIR"
     worldos_timeout "$beat_timeout" \
       claude -p "$msg" ${resume[@]+"${resume[@]}"} ${extra[@]+"${extra[@]}"} --plugin-dir "$ROOT" --mcp-config "$DM_CFG" --strict-mcp-config \
         --model "$WORLDOS_DM_MODEL" ${WORLDOS_DM_EFFORT[@]+"${WORLDOS_DM_EFFORT[@]}"} --permission-mode bypassPermissions --max-budget-usd "$BUDGET" \
+        ${WORLDOS_STREAM_FLAG[@]+"${WORLDOS_STREAM_FLAG[@]}"} \
         --output-format stream-json --verbose > "$out" 2>> "$T/$RUN.dm.err"
     rc=$?
+    worldos_stream_tailer_stop
     cat "$out" >> "$COMBINED"
     # F12-11: surface the REAL failure cause on a nonzero rc with NO error-class result (a timeout
     # rc=124 writes no result event; a CLI crash; a rate-limit exit) — these were MASKED because
diff --git a/qa/test_stream_tailer.py b/qa/test_stream_tailer.py
new file mode 100644
index 00000000..b79a4731
--- /dev/null
+++ b/qa/test_stream_tailer.py
@@ -0,0 +1,303 @@
+#!/usr/bin/env python3
+"""Tests for scripts/stream_tailer.py — WorldOS #835 Live Composition Increment 1.
+
+The tailer's partial-JSON parsing is the riskiest part of the feature, so these tests feed
+SYNTHETIC stream-json sequences (content_block_start log_event + input_json_delta chunks)
+through the pure StreamDecoder and assert:
+  * the decoded prose equals the intended scene text;
+  * `kind` BEFORE *and* AFTER `text` both decode correctly (buffer-until-kind);
+  * partial_json split across arbitrary chunk boundaries (incl. mid `\\"` and mid `\\uXXXX`)
+    decodes correctly;
+  * a non-narration kind (system/roll) is NEVER streamed;
+  * multiple sequential log_event calls in one beat stream in order;
+  * MCP-prefixed tool names (mcp__clawdnd-engine__log_event) are recognized;
+  * both the nested-`event` and flat stream-json row shapes are tolerated;
+  * the file-tailing driver decodes a real growing-file stream.
+
+Pure-stdlib; SINGLE-PROCESS (no xdist). Run with:
+    python3 -m pytest qa/test_stream_tailer.py -q -p no:xdist
+or:
+    uv run --directory servers/engine python -m pytest qa/test_stream_tailer.py -q -p no:xdist
+"""
+
+from __future__ import annotations
+
+import json
+import sys
+from pathlib import Path
+
+SCRIPTS_DIR = Path(__file__).resolve().parent.parent / "scripts"
+sys.path.insert(0, str(SCRIPTS_DIR))
+
+import stream_tailer  # noqa: E402
+
+
+# ---------------------------------------------------------------------------------------------
+# Helpers: build synthetic stream-json rows mirroring `claude -p --include-partial-messages`.
+# ---------------------------------------------------------------------------------------------
+
+def _row(event: dict, *, nested: bool = True) -> dict:
+    """Wrap a raw API stream `event` as the CLI emits it. `nested=True` -> the canonical
+    `{"type":"stream_event","event":{...}}`; `nested=False` -> the flattened shape (the event
+    fields at top level) — the tailer must tolerate both."""
+    if nested:
+        return {"type": "stream_event", "event": event, "session_id": "s", "uuid": "u"}
+    return dict(event)
+
+
+def block_start(index: int, name: str, *, nested: bool = True) -> dict:
+    return _row({
+        "type": "content_block_start",
+        "index": index,
+        "content_block": {"type": "tool_use", "id": f"toolu_{index}", "name": name, "input": {}},
+    }, nested=nested)
+
+
+def block_delta(index: int, partial: str, *, nested: bool = True) -> dict:
+    return _row({
+        "type": "content_block_delta",
+        "index": index,
+        "delta": {"type": "input_json_delta", "partial_json": partial},
+    }, nested=nested)
+
+
+def block_stop(index: int, *, nested: bool = True) -> dict:
+    return _row({"type": "content_block_stop", "index": index}, nested=nested)
+
+
+def _decode(rows, **kw):
+    """Run rows through StreamDecoder, returning the concatenated emitted prose."""
+    out = []
+    dec = stream_tailer.StreamDecoder(out.append)
+    for r in rows:
+        dec.feed_row(r)
+    return "".join(out)
+
+
+def _chunks_of(s: str, size: int):
+    return [s[i:i + size] for i in range(0, len(s), size)]
+
+
+# ---------------------------------------------------------------------------------------------
+# Core decoding: kind ordering.
+# ---------------------------------------------------------------------------------------------
+
+def test_kind_before_text():
+    """The canonical order: {"kind":"narration","text":"<scene>"} — streams the full scene."""
+    scene = "The weathered lantern flickers above the tavern door as dusk bleeds purple."
+    arg = json.dumps({"kind": "narration", "text": scene})
+    rows = [block_start(1, "log_event")] + [block_delta(1, c) for c in _chunks_of(arg, 7)] + [block_stop(1)]
+    assert _decode(rows) == scene
+
+
+def test_kind_after_text_buffers_until_known():
+    """When `text` appears BEFORE `kind`, the decoded prose must be buffered and flushed
+    retroactively once `kind` resolves to narration — not dropped, not leaked early."""
+    scene = "Lantern light smears the rain-slicked streets into silver rivers."
+    arg = json.dumps({"text": scene, "kind": "dialogue"})  # text first, then kind
+    rows = [block_start(2, "log_event")] + [block_delta(2, c) for c in _chunks_of(arg, 9)] + [block_stop(2)]
+    # dialogue is a prose kind → the full text streams.
+    assert _decode(rows) == scene
+
+
+def test_kind_after_text_nonprose_never_streams():
+    """text-before-kind where kind resolves to a NON-prose kind → nothing is ever emitted
+    (the buffered text is discarded once kind is known to be non-prose)."""
+    arg = json.dumps({"text": "rolled 17 vs AC 14 — hit", "kind": "roll"})
+    rows = [block_start(3, "log_event")] + [block_delta(3, c) for c in _chunks_of(arg, 5)] + [block_stop(3)]
+    assert _decode(rows) == ""
+
+
+# ---------------------------------------------------------------------------------------------
+# Core decoding: non-prose kinds are never streamed (kind-first too).
+# ---------------------------------------------------------------------------------------------
+
+def test_nonprose_kind_first_never_streams():
+    arg = json.dumps({"kind": "system", "text": "State grounded; closing turn."})
+    rows = [block_start(1, "log_event")] + [block_delta(1, c) for c in _chunks_of(arg, 8)] + [block_stop(1)]
+    assert _decode(rows) == ""
+
+
+def test_combat_kind_never_streams():
+    arg = json.dumps({"kind": "combat", "text": "Goblin takes 6 slashing."})
+    rows = [block_start(1, "log_event"), block_delta(1, arg), block_stop(1)]
+    assert _decode(rows) == ""
+
+
+# ---------------------------------------------------------------------------------------------
+# Chunk-boundary robustness: split the SAME scene at every possible boundary + adversarial sizes.
+# ---------------------------------------------------------------------------------------------
+
+def test_chunk_boundaries_every_split_size():
+    scene = 'A "weathered" sign creaks; the harbor wind carries salt and woodsmoke.'
+    arg = json.dumps({"kind": "narration", "text": scene})
+    for size in range(1, len(arg) + 1):
+        rows = [block_start(1, "log_event")] + [block_delta(1, c) for c in _chunks_of(arg, size)] + [block_stop(1)]
+        assert _decode(rows) == scene, f"failed at chunk size {size}"
+
+
+def test_escaped_quote_split_across_chunks():
+    """A JSON `\\"` escape split BETWEEN two deltas (backslash in chunk N, quote in chunk N+1)
+    must decode to a single `"` — not terminate the string early."""
+    scene = 'She said, "hold the line," and drew her blade.'
+    arg = json.dumps({"kind": "narration", "text": scene})
+    # Find a `\"` and split exactly between the backslash and the quote.
+    idx = arg.find('\\"')
+    assert idx != -1
+    chunks = [arg[:idx + 1], arg[idx + 1:]]  # backslash ends chunk 0; quote starts chunk 1
+    rows = [block_start(1, "log_event")] + [block_delta(1, c) for c in chunks] + [block_stop(1)]
+    assert _decode(rows) == scene
+
+
+def test_unicode_escape_split_mid_sequence():
+    """A `\\uXXXX` escape split mid-sequence (across one or more chunk boundaries) must decode
+    to the correct single character."""
+    scene = "dusk bleeds purple — a long dash é accent."  # em-dash + e-acute
+    arg = json.dumps({"kind": "narration", "text": scene}, ensure_ascii=True)  # forces \uXXXX
+    assert "\\u" in arg
+    # Split at size 3 so a \uXXXX (6 chars: \,u,X,X,X,X) is guaranteed cut mid-sequence somewhere.
+    rows = [block_start(1, "log_event")] + [block_delta(1, c) for c in _chunks_of(arg, 3)] + [block_stop(1)]
+    assert _decode(rows) == scene
+
+
+def test_newline_and_tab_escapes():
+    scene = "Line one.\nLine two.\tTabbed."
+    arg = json.dumps({"kind": "narration", "text": scene})
+    rows = [block_start(1, "log_event")] + [block_delta(1, c) for c in _chunks_of(arg, 4)] + [block_stop(1)]
+    assert _decode(rows) == scene
+
+
+# ---------------------------------------------------------------------------------------------
+# Multiple sequential log_event calls in one beat → ordered concatenation.
+# ---------------------------------------------------------------------------------------------
+
+def test_multiple_log_events_stream_in_order():
+    a = "First, the door groans open."
+    b = "Then, a figure steps from the dark."
+    arg_a = json.dumps({"kind": "narration", "text": a})
+    arg_b = json.dumps({"kind": "dialogue", "text": b})
+    rows = (
+        [block_start(1, "log_event")] + [block_delta(1, c) for c in _chunks_of(arg_a, 6)] + [block_stop(1)]
+        + [block_start(2, "log_event")] + [block_delta(2, c) for c in _chunks_of(arg_b, 6)] + [block_stop(2)]
+    )
+    assert _decode(rows) == a + b
+
+
+def test_interleaved_nonprose_between_prose_calls():
+    """A roll log_event between two narration calls must not pollute the streamed prose."""
+    a = "The blade sings free."
+    b = "Blood beads on the cobbles."
+    roll = json.dumps({"kind": "roll", "text": "d20=18"})
+    arg_a = json.dumps({"kind": "narration", "text": a})
+    arg_b = json.dumps({"kind": "narration", "text": b})
+    rows = (
+        [block_start(1, "log_event")] + [block_delta(1, arg_a)] + [block_stop(1)]
+        + [block_start(2, "log_event")] + [block_delta(2, roll)] + [block_stop(2)]
+        + [block_start(3, "log_event")] + [block_delta(3, arg_b)] + [block_stop(3)]
+    )
+    assert _decode(rows) == a + b
+
+
+# ---------------------------------------------------------------------------------------------
+# Tool-name handling: MCP prefixes + non-target tools.
+# ---------------------------------------------------------------------------------------------
+
+def test_mcp_prefixed_log_event_name_recognized():
+    scene = "The market square wakes under a bruised sky."
+    arg = json.dumps({"kind": "narration", "text": scene})
+    rows = [block_start(1, "mcp__clawdnd-engine__log_event"), block_delta(1, arg), block_stop(1)]
+    assert _decode(rows) == scene
+
+
+def test_non_target_tool_ignored():
+    """A non-prose tool (e.g. roll_dice / attack) is NOT a candidate — its args never stream."""
+    arg = json.dumps({"expression": "1d20+5", "text": "should not stream"})
+    rows = [block_start(1, "mcp__clawdnd-engine__roll_dice"), block_delta(1, arg), block_stop(1)]
+    assert _decode(rows) == ""
+
+
+def test_assistant_text_delta_never_streams():
+    """A `text_delta` (assistant prose, NOT a tool arg) must never stream in Increment 1."""
+    rows = [_row({
+        "type": "content_block_delta",
+        "index": 0,
+        "delta": {"type": "text_delta", "text": "internal planning scaffolding..."},
+    })]
+    assert _decode(rows) == ""
+
+
+def test_persist_beat_recognized_but_not_streamed_increment1():
+    """persist_beat is a recognized candidate but its nested-events `text` is deferred — it must
+    NOT leak its top-level args, and (Increment 1) streams nothing."""
+    arg = json.dumps({"events": [{"kind": "narration", "text": "nested scene"}]})
+    rows = [block_start(1, "persist_beat")] + [block_delta(1, c) for c in _chunks_of(arg, 5)] + [block_stop(1)]
+    assert _decode(rows) == ""
+
+
+# ---------------------------------------------------------------------------------------------
+# Row-shape tolerance: flat (non-nested) stream-json rows.
+# ---------------------------------------------------------------------------------------------
+
+def test_flat_row_shape_tolerated():
+    scene = "A flat-shaped event stream still decodes."
+    arg = json.dumps({"kind": "narration", "text": scene})
+    rows = (
+        [block_start(1, "log_event", nested=False)]
+        + [block_delta(1, c, nested=False) for c in _chunks_of(arg, 5)]
+        + [block_stop(1, nested=False)]
+    )
+    assert _decode(rows) == scene
+
+
+# ---------------------------------------------------------------------------------------------
+# File-tailing driver: a real growing file → decoded chunks in current.jsonl.
+# ---------------------------------------------------------------------------------------------
+
+def test_tail_stream_driver_decodes_growing_file(tmp_path):
+    scene = "The campfire gutters; shadows lengthen across the ruined keep."
+    arg = json.dumps({"kind": "narration", "text": scene})
+    lines = (
+        [block_start(1, "log_event")]
+        + [block_delta(1, c) for c in _chunks_of(arg, 11)]
+        + [block_stop(1)]
+    )
+    out_path = tmp_path / "dm.jsonl"
+    # Write the whole stream up front (the driver reads from offset 0 forward), include a
+    # trailing partial line to prove the tail-carry never yields it.
+    body = "\n".join(json.dumps(r) for r in lines) + "\n" + '{"type":"stream_eve'  # half line
+    out_path.write_text(body, encoding="utf-8")
+
+    stream_path = tmp_path / "stream" / "current.jsonl"
+    # stop() after one pass: the file is already fully written, so one read decodes everything.
+    passes = {"n": 0}
+
+    def stop():
+        passes["n"] += 1
+        return passes["n"] > 2  # let it read + idle a couple of cycles
+
+    written = stream_tailer.tail_stream(
+        str(out_path), str(stream_path), poll_interval=0.001, stop=stop
+    )
+    assert written >= 1
+    decoded = "".join(
+        json.loads(ln)["text"]
+        for ln in stream_path.read_text(encoding="utf-8").splitlines()
+        if ln.strip()
+    )
+    assert decoded == scene
+
+
+def test_tail_stream_truncates_sink_on_start(tmp_path):
+    """The sink (current.jsonl) is truncated at beat start (reset on each new attempt)."""
+    stream_path = tmp_path / "stream" / "current.jsonl"
+    stream_path.parent.mkdir(parents=True)
+    stream_path.write_text('{"seq":99,"text":"STALE FROM PRIOR BEAT","ts":0}\n', encoding="utf-8")
+    out_path = tmp_path / "dm.jsonl"
+    out_path.write_text("", encoding="utf-8")
+    stream_tailer.tail_stream(str(out_path), str(stream_path), poll_interval=0.001, stop=lambda: True)
+    # Opened with "w" → the stale row is gone.
+    assert "STALE FROM PRIOR BEAT" not in stream_path.read_text(encoding="utf-8")
+
+
+if __name__ == "__main__":
+    raise SystemExit(__import__("pytest").main([__file__, "-q", "-p", "no:xdist"]))
diff --git a/scripts/play.sh b/scripts/play.sh
index 1ac28808..a6c6f94f 100755
--- a/scripts/play.sh
+++ b/scripts/play.sh
@@ -384,14 +384,26 @@ dm_turn() {
   # WORLDOS_BEAT_TIMEOUT (default 360s). Keyed off the SAME `first` signal as the effort tier above.
   beat_timeout="$(worldos_dm_timeout "$first")"
   out="$DM_LOG.$(date +%s%N).jsonl"
+  # #835 Increment 1 — Live Composition flag (default OFF behind WORLDOS_STREAM_BEATS). The shared
+  # helpers (qa/lib_beat_driver.sh) build WORLDOS_STREAM_FLAG = (--include-partial-messages) ONLY
+  # when streaming is on; when off the array is empty and the splice below expands to nothing, so the
+  # invocation is byte-identical to today. _dm_invoke launches the per-attempt stream tailer against
+  # $out before the call and kills it after (no-op when off) — the tailer is a sidecar (a crash never
+  # affects the beat). Both attempt 1 and the retry (which re-mint $out) get their own tailer.
+  worldos_stream_flag_arg
   # F12-8: worldos_timeout (qa/lib_beat_driver.sh) — timeout(1) when present, else a python3
   # fallback with the same rc=124 semantics. A bare `timeout` died rc=127 on stock (non-coreutils)
   # Macs, killing every beat in <1s with the failure masked.
   _dm_invoke() {
+    worldos_stream_tailer_start "$out" "$STATE_DIR"
     worldos_timeout "$beat_timeout" \
       claude -p "$msg" ${resume[@]+"${resume[@]}"} ${extra[@]+"${extra[@]}"} --plugin-dir "$ROOT" --mcp-config "$DM_CFG" --strict-mcp-config \
         --model "$WORLDOS_DM_MODEL" ${WORLDOS_DM_EFFORT[@]+"${WORLDOS_DM_EFFORT[@]}"} --permission-mode bypassPermissions --max-budget-usd "$BUDGET" \
+        ${WORLDOS_STREAM_FLAG[@]+"${WORLDOS_STREAM_FLAG[@]}"} \
         --output-format stream-json --verbose > "$out" 2>> "$DM_LOG.err"
+    local _rc=$?
+    worldos_stream_tailer_stop
+    return $_rc
   }
   _dm_invoke; rc=$?
   if [ "$rc" -ne 0 ]; then
diff --git a/scripts/play_party.sh b/scripts/play_party.sh
index 704333b9..c1dd5d3f 100755
--- a/scripts/play_party.sh
+++ b/scripts/play_party.sh
@@ -378,14 +378,25 @@ turn() {
     # DM turn with ONE retry (parity with scripts/play.sh dm_turn — play_party is the native app's
     # entry point and previously had NO DM retry, so a transient cold-open failure was permanent).
     local rc
+    # #835 Increment 1 — Live Composition flag (default OFF behind WORLDOS_STREAM_BEATS). Shared
+    # helpers (qa/lib_beat_driver.sh) build WORLDOS_STREAM_FLAG = (--include-partial-messages) ONLY
+    # when streaming is on; off → empty array → the splice expands to nothing → byte-identical to
+    # today. _dm_invoke launches the per-attempt stream tailer against $out before the call and kills
+    # it after (no-op when off); the tailer is a sidecar (a crash never affects the beat).
+    worldos_stream_flag_arg
     # F12-8: worldos_timeout (qa/lib_beat_driver.sh) — timeout(1) when present, else a python3
     # fallback with the same rc=124 semantics. A bare `timeout` died rc=127 on stock
     # (non-coreutils) Macs, killing every beat in <1s with the failure masked.
     _dm_invoke() {
+      worldos_stream_tailer_start "$out" "$STATE_DIR"
       worldos_timeout "$beat_timeout" \
         claude -p "$msg" ${resume[@]+"${resume[@]}"} ${extra[@]+"${extra[@]}"} --plugin-dir "$ROOT" --mcp-config "$DM_CFG" --strict-mcp-config \
           --model "$WORLDOS_DM_MODEL" ${WORLDOS_DM_EFFORT[@]+"${WORLDOS_DM_EFFORT[@]}"} --permission-mode bypassPermissions --max-budget-usd "$BUDGET" \
+          ${WORLDOS_STREAM_FLAG[@]+"${WORLDOS_STREAM_FLAG[@]}"} \
           --output-format stream-json --verbose > "$out" 2>> "$DM_LOG.err"
+      local _rc=$?
+      worldos_stream_tailer_stop
+      return $_rc
     }
     _dm_invoke; rc=$?
     if [ "$rc" -ne 0 ]; then
diff --git a/scripts/stream_tailer.py b/scripts/stream_tailer.py
new file mode 100644
index 00000000..58a699b1
--- /dev/null
+++ b/scripts/stream_tailer.py
@@ -0,0 +1,468 @@
+#!/usr/bin/env python3
+"""WorldOS #835 Live Composition — Increment 1: the stream tailer (Layer B).
+
+Tails the DM's per-attempt stream-json `$out` file AS IT IS WRITTEN (the wrapper runs
+`claude -p ... --output-format stream-json --include-partial-messages` only when
+WORLDOS_STREAM_BEATS=1) and decodes the player-facing scene prose out of the streaming
+TOOL ARGUMENT of the DM's `log_event(kind="narration"|"dialogue", text=...)` calls,
+writing decoded chunks to `$STATE_DIR/stream/current.jsonl` for the viewer to poll.
+
+Why the TOOL ARG and not assistant text (the keystone, from #835):
+  A DM beat is multi-turn (tool calls + internal planning + the scene). Naively streaming
+  assistant `text_delta` leaks scaffolding (the #732 class). But the DM writes its
+  player-facing scene THROUGH `log_event(kind="narration", text=<scene>)`. With
+  `--include-partial-messages` the CLI emits `content_block_start` carrying the tool NAME
+  *before* its args stream, and the args then arrive as `input_json_delta.partial_json`
+  chunks. So the prose is deterministically player-safe BY CONSTRUCTION — identified
+  before its first word arrives — and we stream only the `text` value of narration/dialogue
+  log_event calls.
+
+This module is PURE-STDLIB and split into a testable parsing core (`StreamDecoder`) and a
+thin file-tailing driver (`tail_stream` / `main`). The parser is the riskiest part and is
+unit-tested in qa/test_stream_tailer.py. The tailer is a SIDECAR: if it crashes the DM beat
+is unaffected (the wrapper kills it on beat end; the viewer simply stops seeing new chunks
+and the canonical /events + /chat paths resolve the beat normally).
+
+Increment 1 scope:
+  * `log_event` (flat `text` arg) is the PRIMARY target — fully handled.
+  * `persist_beat` (nested events list) is best-effort/skippable for Increment 1: its block
+    is recognized as a candidate but its nested-`text` extraction is intentionally not wired
+    (a later increment). It never streams scaffolding — worst case it streams nothing.
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import sys
+import time
+from typing import Callable, Iterable, Optional
+
+# Tool-name suffixes whose tool-arg `text` carries player-facing prose. Matched on the bare
+# name (after splitting off any MCP prefix, e.g. `mcp__clawdnd-engine__log_event` -> `log_event`),
+# matching story_readout / latency_rollup's `.split("__")[-1]` convention. `persist_beat` is a
+# candidate for recognition but its nested-events `text` extraction is deferred (Increment 1).
+PROSE_TOOL_SUFFIXES = ("log_event", "persist_beat")
+# Only these `kind` values are player-safe prose. A log_event with any other kind (system, roll,
+# combat, ...) is NEVER streamed (its text is internal bookkeeping, not scene prose).
+PROSE_KINDS = frozenset({"narration", "dialogue"})
+
+
+def _short_tool(name: object) -> str:
+    """Bare engine tool name from a possibly MCP-prefixed tool_use name.
+    ``mcp__clawdnd-engine__log_event`` -> ``log_event`` (matches latency_rollup._short_tool)."""
+    return str(name or "").split("__")[-1]
+
+
+def _unwrap_event(row: dict) -> dict:
+    """Return the raw Anthropic stream event from a stream-json row.
+
+    The CLI's `--include-partial-messages` wraps each raw API SSE event. Depending on CLI
+    version the event is either nested under an `event` key
+    (`{"type":"stream_event","event":{"type":"content_block_start",...}}`) or the row IS the
+    event itself (`{"type":"content_block_start",...}`). Tolerate BOTH so the tailer survives
+    a CLI shape change."""
+    ev = row.get("event")
+    if isinstance(ev, dict):
+        return ev
+    return row
+
+
+class _PartialJsonScanner:
+    """Incrementally extract the `text` (and `kind`) string values from a JSON object whose
+    serialization arrives in arbitrary chunks (the tool-arg `partial_json` deltas).
+
+    We do NOT wait for a complete, parseable JSON object — the whole point is to stream the
+    `text` value as it grows, mid-string. This is a small hand-rolled JSON-string state machine
+    that:
+      * tracks whether we are inside a string, and (when inside) whether that string is a KEY
+        or a VALUE, and which key the current value belongs to;
+      * decodes JSON string escapes (`\\"`, `\\n`, `\\uXXXX`, ...) INCREMENTALLY, carrying
+        backslash / \\uXXXX state ACROSS chunk boundaries so a `\\"` or a `\\uXXXX` split
+        between two deltas decodes correctly;
+      * emits only fully-decoded characters of the `text` value (never a half-formed escape).
+
+    It captures the decoded value of `kind` (small, completes fast) and the running decoded
+    prefix of `text`. The driver decides, using `kind`, whether to surface `text`.
+
+    Robustness: the scanner only ever READS the buffer forward; it never needs the object to be
+    well-formed past the `text` value, and a malformed tail simply stops producing new decoded
+    text (the beat still resolves via the canonical paths). Only top-level keys are tracked
+    (depth-aware), so a nested object's `text` (e.g. inside persist_beat's events list) does not
+    masquerade as the flat narration text.
+    """
+
+    def __init__(self) -> None:
+        # Decoded values captured so far.
+        self.kind: Optional[str] = None       # decoded `kind` value once its string closes
+        self.text: str = ""                    # running decoded prefix of the `text` value
+        self._text_closed = False              # the `text` string has fully closed
+
+        # Lexer state.
+        self._in_string = False
+        self._is_key = False                   # the current string is an object KEY
+        self._cur_key: Optional[str] = None    # decoded chars of the in-progress key
+        self._cur_val_key: Optional[str] = None  # which key the in-progress VALUE belongs to
+        self._expect_value = False             # we just saw a `:` — the next string is a value
+        self._depth = 0                        # object/array nesting; we track keys only at depth 1
+        self._after_key = False                # we just closed a key string; awaiting `:`
+
+        # Escape state, carried across chunk boundaries.
+        self._escape = False                   # previous char was an unescaped backslash
+        self._u_remaining = 0                  # \uXXXX hex digits still expected
+        self._u_acc = ""                       # accumulated hex digits of the current \u escape
+
+    @property
+    def text_complete(self) -> bool:
+        return self._text_closed
+
+    def feed(self, chunk: str) -> None:
+        """Consume a `partial_json` chunk, advancing the lexer + decoders."""
+        for ch in chunk:
+            self._consume(ch)
+
+    # -- internals -----------------------------------------------------------------
+
+    def _emit_decoded(self, decoded: str) -> None:
+        """Route a fully-decoded character into the value it belongs to."""
+        if self._is_key:
+            self._cur_key = (self._cur_key or "") + decoded
+            return
+        # A VALUE character.
+        if self._cur_val_key == "text" and not self._text_closed:
+            self.text += decoded
+        # `kind` is captured on string close (it's short); no need to stream it char-by-char.
+        elif self._cur_val_key == "kind":
+            # Buffer kind chars in self.text? No — keep them separate.
+            self._kind_acc = getattr(self, "_kind_acc", "") + decoded
+
+    def _consume(self, ch: str) -> None:
+        if self._in_string:
+            self._consume_in_string(ch)
+            return
+        self._consume_structural(ch)
+
+    def _consume_in_string(self, ch: str) -> None:
+        # Mid \uXXXX escape: accumulate exactly 4 hex digits, then emit one decoded char.
+        if self._u_remaining > 0:
+            self._u_acc += ch
+            self._u_remaining -= 1
+            if self._u_remaining == 0:
+                try:
+                    self._emit_decoded(chr(int(self._u_acc, 16)))
+                except ValueError:
+                    pass  # malformed \u — drop it rather than throw (sidecar must never crash)
+                self._u_acc = ""
+            return
+        if self._escape:
+            self._escape = False
+            if ch == "u":
+                self._u_remaining = 4
+                self._u_acc = ""
+                return
+            self._emit_decoded(_JSON_ESCAPES.get(ch, ch))
+            return
+        if ch == "\\":
+            self._escape = True
+            return
+        if ch == '"':
+            # String closes.
+            self._in_string = False
+            if self._is_key:
+                self._after_key = True
+            else:
+                # A value string just closed — finalize captured values.
+                if self._cur_val_key == "kind":
+                    self.kind = getattr(self, "_kind_acc", "")
+                    self._kind_acc = ""
+                elif self._cur_val_key == "text":
+                    self._text_closed = True
+                self._cur_val_key = None
+            self._is_key = False
+            return
+        # An ordinary in-string character.
+        self._emit_decoded(ch)
+
+    def _consume_structural(self, ch: str) -> None:
+        if ch == '"':
+            self._in_string = True
+            # At object depth 1, a string is a KEY when we're NOT expecting a value; otherwise a VALUE.
+            if self._depth == 1 and not self._expect_value:
+                self._is_key = True
+                self._cur_key = ""
+            else:
+                self._is_key = False
+                if self._expect_value and self._depth == 1:
+                    self._cur_val_key = self._pending_key
+                self._expect_value = False
+            return
+        if ch == ":":
+            if self._after_key:
+                self._pending_key = self._cur_key
+                self._after_key = False
+                self._expect_value = True
+            return
+        if ch in "{[":
+            self._depth += 1
+            self._expect_value = False
+            self._after_key = False
+            return
+        if ch in "}]":
+            self._depth -= 1
+            self._expect_value = False
+            self._after_key = False
+            return
+        if ch == ",":
+            self._expect_value = False
+            self._after_key = False
+            return
+        # whitespace / digits / other scalars between structure — ignored for our purpose.
+
+
+# JSON single-char escape map (excluding \u which is handled separately).
+_JSON_ESCAPES = {
+    '"': '"',
+    "\\": "\\",
+    "/": "/",
+    "b": "\b",
+    "f": "\f",
+    "n": "\n",
+    "r": "\r",
+    "t": "\t",
+}
+
+
+class StreamDecoder:
+    """Stateful decoder over a sequence of stream-json ROWS (dicts). Drives one
+    `_PartialJsonScanner` per CANDIDATE tool_use block (a log_event/persist_beat call), and
+    emits decoded prose chunks for narration/dialogue ones.
+
+    The emit callback is invoked with the NEWLY-DECODED text delta (the increment since the
+    last emit for that block), already gated on a KNOWN player-safe `kind`. Key-ORDER is
+    handled: if `kind` is not yet known when `text` starts decoding, the decoded text is
+    BUFFERED and only flushed once `kind` resolves to a prose kind (and discarded if it
+    resolves to a non-prose kind). This handles `kind` appearing BEFORE or AFTER `text`.
+
+    Pure: feed rows, get callbacks. The file-tailing driver below adapts a growing file to this.
+    """
+
+    def __init__(self, emit: Callable[[str], None]) -> None:
+        self._emit = emit
+        # Per-block-index state. A block is a CANDIDATE if its tool name suffix is a prose tool.
+        # state: {"scanner", "tool", "emitted": int(chars already emitted), "buffered": bool}
+        self._blocks: dict[int, dict] = {}
+
+    def feed_row(self, row: dict) -> None:
+        if not isinstance(row, dict):
+            return
+        ev = _unwrap_event(row)
+        if not isinstance(ev, dict):
+            return
+        etype = ev.get("type")
+        if etype == "content_block_start":
+            self._on_block_start(ev)
+        elif etype == "content_block_delta":
+            self._on_block_delta(ev)
+        elif etype == "content_block_stop":
+            self._on_block_stop(ev)
+        # message_start/message_delta/message_stop/result and assistant text_delta are ignored:
+        # we NEVER stream assistant text in Increment 1 (scaffolding risk, #732 class).
+
+    def _on_block_start(self, ev: dict) -> None:
+        idx = ev.get("index")
+        block = ev.get("content_block") or {}
+        if not isinstance(block, dict):
+            return
+        if block.get("type") != "tool_use":
+            return
+        tool = _short_tool(block.get("name"))
+        if tool not in PROSE_TOOL_SUFFIXES:
+            return
+        # Mark this block index as a CANDIDATE. `persist_beat` is recognized but its nested-`text`
+        # is not extracted in Increment 1 (the scanner only flushes a depth-1 flat `text`).
+        if isinstance(idx, int):
+            self._blocks[idx] = {
+                "scanner": _PartialJsonScanner(),
+                "tool": tool,
+                "emitted": 0,
+            }
+
+    def _on_block_delta(self, ev: dict) -> None:
+        idx = ev.get("index")
+        st = self._blocks.get(idx) if isinstance(idx, int) else None
+        if st is None:
+            return
+        delta = ev.get("delta") or {}
+        if not isinstance(delta, dict) or delta.get("type") != "input_json_delta":
+            return
+        partial = delta.get("partial_json")
+        if not isinstance(partial, str) or not partial:
+            return
+        st["scanner"].feed(partial)
+        self._maybe_emit(st)
+
+    def _on_block_stop(self, ev: dict) -> None:
+        idx = ev.get("index")
+        st = self._blocks.pop(idx, None) if isinstance(idx, int) else None
+        if st is None:
+            return
+        # Final flush: kind may have resolved exactly at close; surface any remaining buffered text.
+        self._maybe_emit(st)
+
+    def _maybe_emit(self, st: dict) -> None:
+        """Emit the newly-decoded prose for a candidate block, gated on a known prose `kind`.
+
+        Buffer-until-kind: while `kind` is unknown the decoded `text` accumulates in the scanner
+        but is NOT emitted. Once `kind` is known:
+          * prose kind  -> flush everything decoded-so-far (catches the BUFFERED prefix) and keep
+            streaming the tail;
+          * non-prose   -> never emit (mark the block so we stop checking).
+        For `persist_beat` we never emit in Increment 1 (its flat depth-1 `text` is absent — the
+        prose lives in a nested events list — so the scanner's `text` stays empty; this is a
+        no-op, never a leak).
+        """
+        if st.get("suppressed"):
+            return
+        scanner = st["scanner"]
+        kind = scanner.kind
+        if kind is None:
+            # Kind not known yet — buffer (do nothing). The decoded text is retained in the
+            # scanner and will be flushed retroactively once kind resolves.
+            return
+        if kind not in PROSE_KINDS:
+            # A non-narration/dialogue log_event (system/roll/combat) — NEVER stream it.
+            st["suppressed"] = True
+            return
+        if st["tool"] != "log_event":
+            # persist_beat (nested events) — deferred for Increment 1; nothing to flush.
+            return
+        full = scanner.text
+        already = st["emitted"]
+        if len(full) > already:
+            delta = full[already:]
+            st["emitted"] = len(full)
+            if delta:
+                self._emit(delta)
+
+
+# ------------------------------------------------------------------------------------------
+# File-tailing driver (the side-effectful half; the parsing core above is what tests exercise).
+# ------------------------------------------------------------------------------------------
+
+
+def _iter_new_lines(path: str, state: dict) -> Iterable[str]:
+    """Yield any COMPLETE new lines appended to `path` since the last call.
+
+    Carries a byte offset + a partial-line tail across calls so a line split across two
+    appends (writes can exceed PIPE_BUF) is only yielded once fully terminated by a newline.
+    Tolerates the file not existing yet (the attempt may not have created `$out` at launch)."""
+    try:
+        size = os.path.getsize(path)
+    except OSError:
+        return
+    offset = state.get("offset", 0)
+    if size < offset:
+        # The file shrank/rotated (a retry minted a fresh $out at the same path is not our case —
+        # the wrapper hands us the newest path — but be defensive): restart from the top.
+        offset = 0
+        state["tail"] = ""
+    if size == offset:
+        return
+    try:
+        with open(path, "r", encoding="utf-8", errors="replace") as f:
+            f.seek(offset)
+            data = f.read()
+            state["offset"] = f.tell()
+    except OSError:
+        return
+    buf = state.get("tail", "") + data
+    lines = buf.split("\n")
+    state["tail"] = lines.pop()  # trailing partial (or "" if data ended on a newline)
+    for line in lines:
+        yield line
+
+
+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)
+    return open(stream_path, "w", encoding="utf-8")
+
+
+def tail_stream(
+    out_path: str,
+    stream_path: str,
+    *,
+    poll_interval: float = 0.25,
+    stop: Optional[Callable[[], bool]] = None,
+    max_idle_s: Optional[float] = None,
+) -> int:
+    """Tail `out_path` (the DM stream-json file) and write decoded prose chunks to
+    `stream_path` ($STATE_DIR/stream/current.jsonl), one JSON line per chunk. Returns the
+    number of chunks written. Runs until `stop()` returns True (when provided) or the process
+    is killed by the wrapper at beat end. `max_idle_s` is a safety valve for tests/standalone
+    runs (stop after that long with no new file growth)."""
+    state: dict = {"offset": 0, "tail": ""}
+    seq = 0
+    sink = _open_sink(stream_path)
+
+    def emit(delta: str) -> None:
+        nonlocal seq
+        row = {"seq": seq, "text": delta, "ts": time.time()}
+        sink.write(json.dumps(row, ensure_ascii=False) + "\n")
+        sink.flush()
+        seq += 1
+
+    decoder = StreamDecoder(emit)
+    last_progress = time.time()
+    try:
+        while True:
+            if stop is not None and stop():
+                break
+            progressed = False
+            for line in _iter_new_lines(out_path, state):
+                progressed = True
+                line = line.strip()
+                if not line:
+                    continue
+                try:
+                    row = json.loads(line)
+                except json.JSONDecodeError:
+                    continue  # half-written or non-JSON wrapper line — skip
+                decoder.feed_row(row)
+            if progressed:
+                last_progress = time.time()
+            elif max_idle_s is not None and (time.time() - last_progress) > max_idle_s:
+                break
+            time.sleep(poll_interval)
+    finally:
+        try:
+            sink.close()
+        except OSError:
+            pass
+    return seq
+
+
+def main(argv: Optional[list] = None) -> int:
+    argv = list(sys.argv[1:] if argv is None else argv)
+    if len(argv) < 2:
+        sys.stderr.write("usage: stream_tailer.py <dm_out.jsonl> <stream_dir_or_current.jsonl>\n")
+        return 2
+    out_path = argv[0]
+    target = argv[1]
+    # Accept either the stream DIR ($STATE_DIR/stream) or the file path directly.
+    if os.path.isdir(target) or not target.endswith(".jsonl"):
+        stream_path = os.path.join(target, "current.jsonl")
+    else:
+        stream_path = target
+    poll = float(os.environ.get("WORLDOS_STREAM_POLL_S", "0.25"))
+    try:
+        tail_stream(out_path, stream_path, poll_interval=poll)
+    except KeyboardInterrupt:
+        pass
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
diff --git a/viewer/openworlds/app.jsx b/viewer/openworlds/app.jsx
index a3f1f998..67af68eb 100644
--- a/viewer/openworlds/app.jsx
+++ b/viewer/openworlds/app.jsx
@@ -327,6 +327,16 @@ function useLiveSession(state) {
   // object without it, so each new turn re-derives `streaming` from its own /events arrivals.
   const chatCursor = React.useRef(0);
   const eventsCursor = React.useRef(0);                // #393: per-file cursor for the live /events tail
+  // #835 Increment 1 — the /beat-stream live-composition tail. `beatStreamCursor` is the per-file
+  // line cursor into the wrapper's $STATE_DIR/stream/current.jsonl sidecar; `composingText` is the
+  // running concatenation of the decoded prose chunks for the IN-FLIGHT beat. The chronicle renders
+  // ONE transient "composing" narration row (composing:true) carrying composingText while a turn is
+  // pending; it is dropped the moment the canonical /events seq-keyed narration row for the same beat
+  // lands (eventsStreamedThisTurnRef) or the turn resolves on /chat — so the live preview is replaced
+  // by the canonical row with ZERO duplication (it never goes through the seq/text dedup sets, so it
+  // cannot poison them). composingText resets per beat (the tailer truncates current.jsonl per beat).
+  const beatStreamCursor = React.useRef(0);
+  const composingText = React.useRef("");
   const dmBeatCountRef = React.useRef(0);
   // #406: the count of turns that have RESOLVED on /chat (the turn-END signal), bumped ONLY in the
   // /chat poll — NOT by streamed /events paragraphs. `firstBeat` (the generous cold-open recovery
@@ -444,8 +454,23 @@ function useLiveSession(state) {
     }
     clearTimers();
     setPendingState(null);
+    // #835 Increment 1: the turn is over → drop the transient composing preview (the canonical
+    // /events narration row already carries the finished beat). Inlined (not via clearComposing) to
+    // keep clearPending's dependency set unchanged — composing rows are flagged, never dedup-keyed.
+    composingText.current = "";
+    setChatBeats((prev) => (prev.some((b) => b && b.composing) ? prev.filter((b) => !(b && b.composing)) : prev));
   }, [clearTimers, setPendingState]);
 
+  // #835 Increment 1 — drop the transient live-composition "composing" row + reset its accumulator.
+  // Called when a turn resolves (clearPending) or a new run starts: by then the canonical, seq-keyed
+  // /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(() => {
+    composingText.current = "";
+    setChatBeats((prev) => (prev.some((b) => b && b.composing) ? prev.filter((b) => !(b && b.composing)) : prev));
+  }, []);
+
   // #826: authoritatively ROLL BACK the optimistic in-flight arm when the /move POST itself is
   // REJECTED by the server (the move never started, so there is no DM turn to wait for). postMove now
   // arms the narrating gate the INSTANT the player commits — BEFORE the network round-trip — so the
@@ -573,6 +598,8 @@ function useLiveSession(state) {
   React.useEffect(() => {
     chatCursor.current = 0;
     eventsCursor.current = 0;          // #393: reset the live /events tail per run
+    beatStreamCursor.current = 0;      // #835: reset the live-composition /beat-stream tail per run
+    composingText.current = "";        // #835: …and its per-beat prose accumulator
     dmBeatCountRef.current = 0;
     resolvedTurnsRef.current = 0;       // #406: a fresh run has resolved no turns yet (cold-open window)
     seenSeq.current = new Set();        // #405: a fresh run shares no seq dedup keys with the last
@@ -746,7 +773,12 @@ function useLiveSession(state) {
             })
             .filter(Boolean);
           if (beats.length) {
-            setChatBeats((prev) => boundTail([...prev, ...beats], MAX_LIVE_BEATS));  // #402: cap the live tail
+            // #835 Increment 1: the canonical, seq-keyed /events narration for this beat just landed
+            // → strip any transient live-composition "composing" preview row in the SAME update, so
+            // the canonical row REPLACES the partial preview with zero duplication (the composing row
+            // is flagged, never dedup-keyed, so this is a clean swap). composingText resets too.
+            composingText.current = "";
+            setChatBeats((prev) => boundTail([...prev.filter((b) => !(b && b.composing)), ...beats], MAX_LIVE_BEATS));  // #402: cap the live tail
             // The scene is visibly building → the turn is plainly alive. Count the streamed prose as
             // real DM beats (so the NEXT turn isn't mis-treated as a cold-open 'firstBeat') and reset
             // the stall clock so a long-but-healthy streaming turn is never falsely declared 'stuck'.
@@ -773,6 +805,72 @@ function useLiveSession(state) {
     return () => { cancelled = true; stop(); document.removeEventListener("visibilitychange", onVisibility); };
   }, [campaignId, source, runId, notePendingProgress, claimNarration]);
 
+  // #835 Increment 1 — the LIVE COMPOSITION tail. While a turn is pending, poll /beat-stream (the
+  // wrapper's stream_tailer sidecar) ~every 500ms, accumulate the decoded prose chunks, and render
+  // them as ONE transient "composing" narration row so the scene UNFOLDS word-by-word as the DM
+  // writes it (instead of a static spinner). This is strictly ADDITIVE to the #393 /events tail:
+  //   • When the feature is OFF (WORLDOS_STREAM_BEATS=0) the wrapper never launches the tailer, so
+  //     current.jsonl is absent → /beat-stream returns empty → this loop is an inert no-op. Today's
+  //     behavior is byte-identical.
+  //   • The composing row is transient + flagged (composing:true), never claimed in the seq/text
+  //     dedup sets, so it cannot poison dedup. The canonical seq-keyed /events narration row for the
+  //     same beat REPLACES it (the /events poll strips composing rows when it lands), and clearPending
+  //     drops it on turn resolution — zero duplication either way. notePendingProgress() flips the
+  //     pending turn to `streaming` so the narrating affordance + latestStreamed preview light up.
+  // Gated on `pendingRef`: it only does work mid-turn, so it never churns the chronicle at rest.
+  React.useEffect(() => {
+    if (!campaignId) return undefined;
+    let cancelled = false;
+    let timer = null;
+    const pollOnce = async () => {
+      if (cancelled) return;
+      // Only stream while a turn is actually in flight — at rest there is no composing beat.
+      if (!pendingRef.current) return;
+      try {
+        const params = new URLSearchParams();
+        params.set("campaign", campaignId);
+        params.set("since", String(beatStreamCursor.current));
+        const resp = await fetch(`/beat-stream?${params.toString()}`, { cache: "no-store" });
+        if (!resp.ok) return;
+        const payload = await resp.json();
+        const chunks = Array.isArray(payload.chunks) ? payload.chunks : [];
+        if (!cancelled && chunks.length) {
+          // Concatenate the decoded chunks in order onto the in-flight beat's running prose. The
+          // tailer already gated these to narration/dialogue text (player-safe by construction), but
+          // run them through the shared sanitizer too (#740 parity with /events) before display.
+          let added = "";
+          for (const c of chunks) { if (c && typeof c.text === "string") added += c.text; }
+          if (added) {
+            composingText.current += added;
+            const preview = sanitize(composingText.current);
+            if (preview) {
+              // Render/replace the single composing row (keyed by the composing flag). nextLogSeq()
+              // keeps it after the latest chronicle entry so latestStreamed (which scans back-to-front
+              // for narration) surfaces it inline at the spinner.
+              const row = { kind: "narration", text: preview, at: nextLogSeq(), composing: true };
+              setChatBeats((prev) => {
+                const base = prev.filter((b) => !(b && b.composing));
+                return boundTail([...base, row], MAX_LIVE_BEATS);
+              });
+              notePendingProgress();  // the scene is arriving → flip `streaming`, keep the turn alive
+            }
+          }
+        }
+        if (!cancelled && typeof payload.next === "number") beatStreamCursor.current = payload.next;
+      } catch (_e) { /* the live-composition tail is non-critical; /events + /chat are the backstop */ }
+    };
+    const stop = () => { if (timer !== null) { window.clearInterval(timer); timer = null; } };
+    // ~500ms cadence: responsive typewriter feel without hammering the stdlib server (the tailer
+    // flushes per decoded chunk, so the sidecar grows faster than this; we coalesce on each poll).
+    const start = () => { if (timer === null) timer = window.setInterval(pollOnce, 500); };
+    const onVisibility = () => {
+      if (document.visibilityState === "visible") { pollOnce(); start(); } else { stop(); }
+    };
+    document.addEventListener("visibilitychange", onVisibility);
+    onVisibility();
+    return () => { cancelled = true; stop(); document.removeEventListener("visibilitychange", onVisibility); };
+  }, [campaignId, source, runId, notePendingProgress]);
+
   // #745: expose notePendingProgress so the live-progress signal is part of the hook's public surface
   // (consistent with armPending/clearPending; the /events poll calls the same ref). Purely additive —
   // existing consumers destructure named fields, so nothing breaks; it also makes the mid-stream stall
diff --git a/viewer/server.py b/viewer/server.py
index ac21be5b..cdedc5a5 100644
--- a/viewer/server.py
+++ b/viewer/server.py
@@ -7127,6 +7127,51 @@ def _read_events(campaign_id: str, since: int) -> tuple[list[dict], int]:
     return out, consumed
 
 
+def _read_beat_stream(since: int) -> tuple[list[dict], int, bool]:
+    """#835 Live Composition Increment 1 — return (new prose chunks after line `since`, new
+    line count, complete?) from the wrapper-owned live-stream sidecar
+    ``$STATE_DIR/stream/current.jsonl``.
+
+    The sidecar is written by scripts/stream_tailer.py (a wrapper-launched tailer that decodes
+    the DM's player-facing scene out of its streaming log_event tool-arg) — it is NOT campaign
+    state, so reading it here keeps the engine's sole-writer invariant untouched (mirrors the
+    read-only /events tail). Each row is ``{"seq":N,"text":"...","ts":...}``. The tailer truncates
+    the file at the start of each beat/attempt; the client resets its cursor when ``since`` runs
+    past the current line count (the same session-rotation guard /events uses).
+
+    Defensive like _read_events: a half-written trailing line is dropped (NOT advanced past) so a
+    torn append re-reads cleanly next poll. ``complete`` is advisory only (Increment 1 has no
+    per-beat terminal marker yet — it always reports False; the canonical /events row resolving the
+    beat is what collapses the composing block, via the existing claimNarration dedup)."""
+    path = _state_dir() / "stream" / "current.jsonl"
+    if not path.exists():
+        return [], since, False
+    try:
+        lines = path.read_text(encoding="utf-8").splitlines()
+    except OSError:
+        return [], since, False
+    # Truncation reset: the tailer rewrites current.jsonl per beat/attempt, so a stale high cursor
+    # would strand the feed (lines[since:] empty forever). Re-read from the top when the cursor is
+    # past the end of the current file.
+    if since > len(lines):
+        since = 0
+    out: list[dict] = []
+    consumed = since
+    for raw in lines[since:]:
+        stripped = raw.strip()
+        if not stripped:
+            consumed += 1
+            continue
+        try:
+            row = json.loads(stripped)
+        except json.JSONDecodeError:
+            break  # half-written trailing chunk — don't advance; re-read next poll
+        if isinstance(row, dict):
+            out.append(row)
+        consumed += 1
+    return out, consumed, False
+
+
 # Roll-result detection (#35): the dice tool's *result* (total / nat-d20 / crit)
 # lives in the tool_RESULT, not the tool_use input — so to headline a real number
 # with crit/miss coloring we mine results too. A roll result is a JSON dict that
@@ -8418,6 +8463,17 @@ def do_GET(self) -> None:  # noqa: N802
             # session rotation (cold-open + DM-turn-retry re-mint), suppressing the new session's
             # post-move narration (seq 0,1,2 already claimed by the prior session's cold-open).
             self._json({"entries": entries, "next": nxt, "sid": _active_session_id(view_cid)})
+        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])
+            chunks, nxt, complete = _read_beat_stream(since)
+            self._json({"chunks": chunks, "next": nxt, "complete": complete})
         elif route == "/activity":
             qs = parse_qs(parsed.query)
             since = int((qs.get("since") or ["0"])[0])