feat: #835 Live Composition Increment 2 — enable-able follow-ups (still dark)

qa/lib_beat_driver.sh

@@ -725,6 +725,15 @@ worldos_stream_flag_arg() {
 # 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
+#
+# #835 Increment 2 FIX B: the DM turn (dm_turn) runs inside a $(...) command-substitution SUBSHELL,
+# so WORLDOS_STREAM_TAILER_PID set here CANNOT escape to the parent shell where the EXIT/INT/TERM
+# cleanup trap runs (same subshell-isolation that forced .dm_last_result to be a FILE). On a clean
+# beat end worldos_stream_tailer_stop (called in the same subshell) kills it fine; but a SIGINT/
+# SIGTERM mid-beat skips that stop and would orphan the tailer with no PID visible to the parent.
+# So we ALSO persist the PID to $STATE_DIR/stream/tailer.pid, which the wrapper's signal trap reads
+# (worldos_stream_tailer_kill_pidfile) — a file survives the subshell boundary. The tailer's own
+# self-bounding caps (WORLDOS_STREAM_TAILER_MAX_S / idle) are the final backstop if even that is missed.
 worldos_stream_tailer_start() {
   WORLDOS_STREAM_TAILER_PID=""
   [ "$(worldos_env STREAM_BEATS 0)" = "1" ] || return 0
@@ -735,16 +744,40 @@ worldos_stream_tailer_start() {
   command -v python3 >/dev/null 2>&1 || return 0
   python3 "$script" "$out" "$state_dir/stream" >/dev/null 2>&1 &
   WORLDOS_STREAM_TAILER_PID="$!"
+  # Persist the PID across the subshell boundary for the signal-trap reaper (FIX B). Best-effort.
+  mkdir -p "$state_dir/stream" 2>/dev/null || true
+  printf '%s\n' "$WORLDOS_STREAM_TAILER_PID" > "$state_dir/stream/tailer.pid" 2>/dev/null || true
   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.
+# best-effort — a tailer that already exited is a benign no-op. Clears the PID (and the pidfile, so
+# a later signal-trap reaper doesn't kill a recycled PID) after.
 worldos_stream_tailer_stop() {
   local pid="${WORLDOS_STREAM_TAILER_PID:-}"
-  [ -n "$pid" ] || return 0
-  kill "$pid" >/dev/null 2>&1 || true
+  if [ -n "$pid" ]; then
+    kill "$pid" >/dev/null 2>&1 || true
+  fi
   WORLDOS_STREAM_TAILER_PID=""
+  # Clear the pidfile too (best-effort; $STATE_DIR is ambient in the harnesses but guard anyway).
+  [ -n "${STATE_DIR:-}" ] && rm -f "$STATE_DIR/stream/tailer.pid" 2>/dev/null || true
+  return 0
+}
+
+# Signal-trap reaper for the orphan-on-signal case (FIX B): kill whatever tailer PID was persisted
+# to $STATE_DIR/stream/tailer.pid by a (possibly subshell-isolated) worldos_stream_tailer_start,
+# then remove the pidfile. Called from the wrappers' EXIT/INT/TERM cleanup traps (which run in the
+# PARENT shell, where WORLDOS_STREAM_TAILER_PID is empty because dm_turn ran in a $(...) subshell).
+# No-op when streaming was OFF (no pidfile) or the tailer already exited. $1 = $STATE_DIR.
+worldos_stream_tailer_kill_pidfile() {
+  local state_dir="${1:-${STATE_DIR:-}}"
+  [ -n "$state_dir" ] || return 0
+  local pidfile="$state_dir/stream/tailer.pid" pid=""
+  [ -f "$pidfile" ] || return 0
+  pid="$(cat "$pidfile" 2>/dev/null)"
+  case "$pid" in ''|*[!0-9]*) pid="" ;; esac   # only a clean numeric PID is killable
+  [ -n "$pid" ] && kill "$pid" >/dev/null 2>&1 || true
+  rm -f "$pidfile" 2>/dev/null || true
   return 0
 }
 

qa/test_stream_tailer.py

@@ -24,6 +24,7 @@
 
 import json
 import sys
+import time
 from pathlib import Path
 
 SCRIPTS_DIR = Path(__file__).resolve().parent.parent / "scripts"
@@ -234,6 +235,82 @@ def test_persist_beat_recognized_but_not_streamed_increment1():
     assert _decode(rows) == ""
 
 
+# ---------------------------------------------------------------------------------------------
+# FIX D — text aliases: the engine's log_event accepts message/content/note as aliases for `text`
+# (normalized `text = text or message or content or note`), so a beat where the DM reaches for an
+# alias must stream just like the canonical `text`. A non-alias key (e.g. meta) is NOT prose.
+# ---------------------------------------------------------------------------------------------
+
+def test_alias_message_streams_like_text():
+    scene = "Rain hammers the slate roofs as you duck into the alley."
+    arg = json.dumps({"kind": "narration", "message": 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_alias_content_streams_like_text():
+    scene = "The chandelier sways; dust sifts down through a shaft of grey light."
+    arg = json.dumps({"kind": "dialogue", "content": scene})
+    rows = [block_start(1, "log_event")] + [block_delta(1, c) for c in _chunks_of(arg, 9)] + [block_stop(1)]
+    assert _decode(rows) == scene
+
+
+def test_alias_note_streams_like_text():
+    scene = "A low bell tolls somewhere beyond the fog-bound wharf."
+    arg = json.dumps({"kind": "narration", "note": scene})
+    rows = [block_start(1, "log_event")] + [block_delta(1, c) for c in _chunks_of(arg, 5)] + [block_stop(1)]
+    assert _decode(rows) == scene
+
+
+def test_alias_value_before_kind_buffers_until_known():
+    """An alias appearing BEFORE kind must buffer-until-kind exactly like `text` does."""
+    scene = "Lanternlight gutters across the wet cobbles."
+    arg = json.dumps({"message": scene, "kind": "narration"})  # alias first, then kind
+    rows = [block_start(1, "log_event")] + [block_delta(1, c) for c in _chunks_of(arg, 4)] + [block_stop(1)]
+    assert _decode(rows) == scene
+
+
+def test_alias_with_nonprose_kind_never_streams():
+    """An alias carrying a non-prose kind (e.g. system) is never streamed — same gate as text."""
+    arg = json.dumps({"kind": "system", "note": "state grounded; closing turn."})
+    rows = [block_start(1, "log_event")] + [block_delta(1, c) for c in _chunks_of(arg, 6)] + [block_stop(1)]
+    assert _decode(rows) == ""
+
+
+def test_text_wins_over_alias_when_both_present():
+    """When both a literal `text` and an alias appear, `text` WINS (mirrors the engine's
+    `text or message or content or note` normalization). The SCANNER resolves its final captured
+    prose to the canonical `text` regardless of an earlier alias — asserted at the scanner level
+    because the streaming decoder can't retract chunks it already flushed for the lower-precedence
+    alias (a non-issue in practice: the DM uses exactly one prose key per call)."""
+    canonical = "The canonical scene the player should see."
+    arg = json.dumps({"kind": "narration", "note": "ignore me", "text": canonical})
+    sc = stream_tailer._PartialJsonScanner()
+    for c in _chunks_of(arg, 5):
+        sc.feed(c)
+    assert sc.text == canonical
+    assert sc.kind == "narration"
+    # And the alias-before-text precedence holds whole-fed too (no chunk artifacts).
+    sc2 = stream_tailer._PartialJsonScanner()
+    sc2.feed(arg)
+    assert sc2.text == canonical
+
+
+def test_non_alias_key_meta_is_not_captured():
+    """A non-alias top-level key (meta) must NEVER be captured as prose, even with a prose kind."""
+    arg = json.dumps({"kind": "narration", "meta": "internal bookkeeping value"})
+    rows = [block_start(1, "log_event")] + [block_delta(1, c) for c in _chunks_of(arg, 6)] + [block_stop(1)]
+    assert _decode(rows) == ""
+
+
+def test_nested_alias_does_not_masquerade_as_flat_prose():
+    """A nested message/content/note (depth>1, e.g. inside persist_beat's events) must not leak
+    as flat narration text — depth-1-only capture, same as the nested-`text` guard."""
+    arg = json.dumps({"events": [{"kind": "narration", "message": "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.
 # ---------------------------------------------------------------------------------------------
@@ -287,6 +364,71 @@ def stop():
     assert decoded == scene
 
 
+def test_tail_stream_self_bounds_on_idle(tmp_path):
+    """FIX B self-bounding: with no new $out growth, the idle cap terminates the tailer even
+    when no stop() is provided — so a missed stop signal can't pin the sidecar open forever."""
+    out_path = tmp_path / "dm.jsonl"
+    out_path.write_text("", encoding="utf-8")  # exists, never grows
+    stream_path = tmp_path / "stream" / "current.jsonl"
+    t0 = time.time()
+    # Tiny idle cap + no lifetime cap: must return promptly via the idle path (no stop() at all).
+    written = stream_tailer.tail_stream(
+        str(out_path), str(stream_path),
+        poll_interval=0.001, max_idle_s=0.05, max_lifetime_s=None,
+    )
+    assert written == 0
+    assert time.time() - t0 < 5.0  # it self-terminated, did not hang
+
+
+def test_tail_stream_self_bounds_on_lifetime(tmp_path):
+    """FIX B self-bounding: the hard lifetime cap terminates the tailer even while $out keeps
+    GROWING (so a stuck-but-active stream can't pin it open past the ceiling)."""
+    out_path = tmp_path / "dm.jsonl"
+    # A line that does NOT decode to any prose (a non-target tool) but keeps the file growing so
+    # the idle cap never fires — only the lifetime cap can stop it.
+    out_path.write_text(json.dumps(block_start(0, "roll_dice")) + "\n", encoding="utf-8")
+    stream_path = tmp_path / "stream" / "current.jsonl"
+
+    appended = {"n": 0}
+
+    def stop():
+        # Side-effect: grow the file every pass so `progressed` stays True (idle cap can't fire).
+        appended["n"] += 1
+        with open(out_path, "a", encoding="utf-8") as fh:
+            fh.write(json.dumps(block_delta(0, "x")) + "\n")
+        return False  # never ask to stop via stop() — the lifetime cap must do it
+
+    t0 = time.time()
+    stream_tailer.tail_stream(
+        str(out_path), str(stream_path),
+        poll_interval=0.001, stop=stop, max_idle_s=None, max_lifetime_s=0.05,
+    )
+    assert time.time() - t0 < 5.0  # the lifetime cap stopped a still-growing stream
+
+
+def test_env_bound_parsing():
+    """FIX B: a positive override wins; 0/negative/garbage DISABLES (None); unset → default."""
+    import os as _os
+    name = "WORLDOS_STREAM_TAILER_MAX_S"
+    saved = _os.environ.get(name)
+    try:
+        _os.environ.pop(name, None)
+        assert stream_tailer._env_bound(name, 1800.0) == 1800.0   # unset → default
+        _os.environ[name] = "42"
+        assert stream_tailer._env_bound(name, 1800.0) == 42.0     # positive override
+        _os.environ[name] = "0"
+        assert stream_tailer._env_bound(name, 1800.0) is None     # 0 disables
+        _os.environ[name] = "-5"
+        assert stream_tailer._env_bound(name, 1800.0) is None     # negative disables
+        _os.environ[name] = "nonsense"
+        assert stream_tailer._env_bound(name, 1800.0) is None     # garbage disables
+    finally:
+        if saved is None:
+            _os.environ.pop(name, None)
+        else:
+            _os.environ[name] = saved
+
+
 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"

qa/test_stream_tailer_lifecycle.sh

@@ -0,0 +1,79 @@
+#!/usr/bin/env bash
+# BEHAVIORAL TEST (no model call): proves the #835 Increment-2 FIX-B stream-tailer lifecycle
+# helpers in qa/lib_beat_driver.sh:
+#   (1) worldos_stream_tailer_start is a NO-OP when streaming is OFF (default WORLDOS_STREAM_BEATS=0)
+#       — no pidfile, no PID — so the feature stays dark by default;
+#   (2) when ON, start persists the bg PID to $STATE_DIR/stream/tailer.pid (the subshell-survivable
+#       handle the signal trap reads) and worldos_stream_tailer_stop removes it;
+#   (3) worldos_stream_tailer_kill_pidfile (the trap reaper) kills the persisted PID and removes the
+#       pidfile — even when WORLDOS_STREAM_TAILER_PID is empty in the calling shell (the orphan-on-
+#       signal case where the tailer was launched inside dm_turn's $(...) subshell);
+#   (4) the reaper is a benign no-op with no pidfile.
+#
+# It sources the REAL qa/lib_beat_driver.sh and uses a stub long-lived bg process as the "tailer"
+# via WORLDOS_STREAM_TAILER pointing at a sleeper script. Self-contained under mktemp; macOS-safe.
+set -uo pipefail
+ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+. "$ROOT/qa/lib_beat_driver.sh"
+
+TMP="$(mktemp -d)"; trap 'rm -rf "$TMP"' EXIT
+STATE_DIR="$TMP/state"; mkdir -p "$STATE_DIR"
+OUT="$TMP/dm.jsonl"; : > "$OUT"
+
+# A stub "tailer": a long-lived PYTHON sleeper (the helper launches it as `python3 "$script"`, so the
+# stub must be valid python, not bash). Stands in for scripts/stream_tailer.py so the test never
+# depends on real DM output or python decode timing — only on the launch/persist/kill lifecycle.
+STUB="$TMP/fake_tailer.py"
+cat > "$STUB" <<'PY'
+import sys, time
+time.sleep(300)
+PY
+export WORLDOS_STREAM_TAILER="$STUB"
+
+fail=0
+chk() { if eval "$2"; then echo "PASS: $1"; else echo "FAIL: $1"; fail=1; fi; }
+alive() { kill -0 "$1" 2>/dev/null; }
+
+PIDFILE="$STATE_DIR/stream/tailer.pid"
+
+# (1) OFF by default: start is a no-op (no pidfile, empty PID).
+unset WORLDOS_STREAM_BEATS
+worldos_stream_tailer_start "$OUT" "$STATE_DIR"
+chk "OFF default: no PID set"        '[ -z "${WORLDOS_STREAM_TAILER_PID:-}" ]'
+chk "OFF default: no pidfile"        '[ ! -f "$PIDFILE" ]'
+
+# (2) ON: start launches the stub, sets the PID, and persists the pidfile.
+export WORLDOS_STREAM_BEATS=1
+worldos_stream_tailer_start "$OUT" "$STATE_DIR"
+sleep 0.2  # let python3 actually start so the liveness probe isn't racing the fork
+START_PID="${WORLDOS_STREAM_TAILER_PID:-}"
+chk "ON: PID is set"                 '[ -n "$START_PID" ]'
+chk "ON: process is alive"           'alive "$START_PID"'
+chk "ON: pidfile written"            '[ -f "$PIDFILE" ]'
+chk "ON: pidfile matches PID"        '[ "$(cat "$PIDFILE")" = "$START_PID" ]'
+
+# worldos_stream_tailer_stop kills the proc and removes the pidfile.
+worldos_stream_tailer_stop
+sleep 0.2
+chk "stop: process killed"           '! alive "$START_PID"'
+chk "stop: PID cleared"              '[ -z "${WORLDOS_STREAM_TAILER_PID:-}" ]'
+chk "stop: pidfile removed"          '[ ! -f "$PIDFILE" ]'
+
+# (3) Orphan-on-signal: start again, then simulate the subshell boundary by CLEARING the global
+# PID (as it would be in the parent shell) and reaping via the pidfile only.
+worldos_stream_tailer_start "$OUT" "$STATE_DIR"
+sleep 0.2
+ORPHAN_PID="${WORLDOS_STREAM_TAILER_PID:-}"
+chk "orphan: tailer alive pre-reap"  'alive "$ORPHAN_PID"'
+WORLDOS_STREAM_TAILER_PID=""          # the parent shell can't see the subshell's global
+worldos_stream_tailer_kill_pidfile "$STATE_DIR"
+sleep 0.2
+chk "orphan: reaper killed by pidfile" '! alive "$ORPHAN_PID"'
+chk "orphan: pidfile removed"          '[ ! -f "$PIDFILE" ]'
+
+# (4) Reaper is a benign no-op with no pidfile.
+worldos_stream_tailer_kill_pidfile "$STATE_DIR"
+chk "reaper no-op when no pidfile"   'true'
+
+if [ "$fail" -eq 0 ]; then echo "ALL PASS"; else echo "FAILURES"; fi
+exit "$fail"

scripts/play.sh

@@ -510,6 +510,13 @@ _play_cleanup() {
   declare -F worldos_release_launch_lock >/dev/null 2>&1 && worldos_release_launch_lock "$ROOT"
   kill "$SUP" 2>/dev/null
   [ -f "$VPID_FILE" ] && kill "$(cat "$VPID_FILE" 2>/dev/null)" 2>/dev/null
+  # #835 Increment 2 FIX B: a SIGINT/SIGTERM mid-beat (Ctrl-C, or a deadline killing the runner)
+  # bypasses the per-beat worldos_stream_tailer_stop, orphaning the live-composition tailer. The
+  # tailer was launched inside dm_turn's $(...) subshell, so WORLDOS_STREAM_TAILER_PID is empty in
+  # THIS parent shell — reap it from the pidfile the start helper persisted. No-op when streaming is
+  # OFF (no pidfile) or the tailer already exited (its own self-bounding caps cover a missed signal).
+  declare -F worldos_stream_tailer_kill_pidfile >/dev/null 2>&1 \
+    && worldos_stream_tailer_kill_pidfile "$STATE_DIR"
 }
 trap _play_cleanup EXIT
 trap '_play_cleanup; exit 130' INT TERM

scripts/play_party.sh

@@ -542,6 +542,13 @@ _party_cleanup() {
   declare -F worldos_release_launch_lock >/dev/null 2>&1 && worldos_release_launch_lock "$ROOT"
   kill "$SUP" 2>/dev/null
   [ -f "$VPID_FILE" ] && kill "$(cat "$VPID_FILE" 2>/dev/null)" 2>/dev/null
+  # #835 Increment 2 FIX B: a SIGINT/SIGTERM mid-beat (Ctrl-C, or a deadline killing the runner)
+  # bypasses the per-beat worldos_stream_tailer_stop, orphaning the live-composition tailer. The
+  # tailer was launched inside dm_turn's $(...) subshell, so WORLDOS_STREAM_TAILER_PID is empty in
+  # THIS parent shell — reap it from the pidfile the start helper persisted. No-op when streaming is
+  # OFF (no pidfile) or the tailer already exited (its own self-bounding caps cover a missed signal).
+  declare -F worldos_stream_tailer_kill_pidfile >/dev/null 2>&1 \
+    && worldos_stream_tailer_kill_pidfile "$STATE_DIR"
 }
 trap _party_cleanup EXIT
 trap '_party_cleanup; exit 130' INT TERM