[feat] Add Dreamverse session and prompt logic

Apply Dreamverse monorepo changes for stack slice 4/13 from the source branch.

Source-Branch: will/dreamverse-monorepo
Source-SHA: 03d3e61df6
Dreamverse-Stack: 4/13
Co-authored-by: SolitaryThinker <wlsaidhi@gmail.com>
Co-authored-by: Matthew Noto <99706358+RandNMR73@users.noreply.github.com>
Co-authored-by: XOR-op <17672363+XOR-op@users.noreply.github.com>
Co-authored-by: Zhang Peiyuan <42993249+jzhang38@users.noreply.github.com>

apps/dreamverse/dreamverse/prompt_safety.py

@@ -0,0 +1,202 @@
+from __future__ import annotations
+# mypy: ignore-errors
+
+import importlib
+import os
+import time
+from dataclasses import dataclass
+from functools import cache
+from pathlib import Path
+
+_SERVER_DIR = Path(__file__).resolve().parent
+_REPO_ROOT = _SERVER_DIR.parent
+_DEFAULT_CLASSIFIER_DIR = _REPO_ROOT / "classifiers"
+CLASSIFIER_DIR = Path(
+    os.path.expandvars(os.path.expanduser(os.getenv("LTX2_CLASSIFIER_DIR", str(_DEFAULT_CLASSIFIER_DIR)))))
+
+
+@dataclass(frozen=True)
+class BlockedPrompt:
+    index: int
+    prompt: str
+    error: str
+
+
+def _load_fasttext_module():
+    try:
+        return importlib.import_module("fasttext")
+    except ImportError as exc:
+        raise RuntimeError("Prompt safety is enabled but the `fasttext` package is not installed.") from exc
+
+
+def resolve_classifier_path(
+    classifier_kind: str,
+    env_var: str,
+    filename: str,
+    legacy_path: str,
+    shared_filename: str,
+) -> str:
+    candidates: list[Path] = []
+    env_path = os.getenv(env_var)
+    if env_path:
+        candidates.append(Path(os.path.expandvars(os.path.expanduser(env_path))))
+    candidates.extend([
+        CLASSIFIER_DIR / filename,
+        Path(f"/home/shared/{shared_filename}"),
+        Path(legacy_path),
+    ])
+
+    for candidate in candidates:
+        if candidate.is_file():
+            return str(candidate)
+
+    checked = "\n".join(f"  - {candidate}" for candidate in candidates)
+    raise FileNotFoundError(f"Could not find the {classifier_kind} classifier.\n"
+                            f"Checked:\n{checked}\n"
+                            "Set LTX2_CLASSIFIER_DIR or the appropriate classifier path environment "
+                            "variable.")
+
+
+@cache
+def load_fasttext_model(model_path: str):
+    return _load_fasttext_module().load_model(model_path)
+
+
+def fasttext_predict(
+    model_path: str,
+    text: str,
+    classifier_name: str,
+) -> tuple[str, float]:
+    model = load_fasttext_model(model_path)
+    text = text.replace("\n", " ")
+    start_time = time.perf_counter()
+    try:
+        labels, probs = model.predict(text)
+    except ValueError as error:
+        if "Unable to avoid copy while creating an array" not in str(error):
+            raise
+        predictions = model.f.predict(f"{text}\n", 1, 0.0, "strict")
+        if not predictions:
+            raise ValueError("fastText returned no predictions") from error
+        probs, labels = zip(*predictions, strict=False)
+    latency_ms = (time.perf_counter() - start_time) * 1000.0
+    identifier = labels[0].replace("__label__", "")
+    confidence = float(probs[0])
+    print("[safety] "
+          f"{classifier_name} fastText latency={latency_ms:.2f}ms "
+          f"label={identifier} confidence={confidence:.4f}")
+    return identifier, confidence
+
+
+def _normalize_classifier_label(identifier: str) -> str:
+    return identifier.strip().lower().replace("-", "_").replace(" ", "_")
+
+
+def _label_matches(
+    identifier: str,
+    blocked_markers: tuple[str, ...],
+    safe_markers: tuple[str, ...],
+) -> bool:
+    normalized = _normalize_classifier_label(identifier)
+    tokens = tuple(token for token in normalized.split("_") if token)
+
+    def has_marker(marker: str) -> bool:
+        normalized_marker = _normalize_classifier_label(marker)
+        return (normalized == normalized_marker or normalized_marker in tokens)
+
+    if any(has_marker(marker) for marker in safe_markers):
+        return False
+    return any(has_marker(marker) for marker in blocked_markers)
+
+
+class PromptSafetyFilter:
+
+    def __init__(
+        self,
+        *,
+        nsfw_classifier=None,
+        hate_speech_classifier=None,
+    ):
+        if nsfw_classifier is None or hate_speech_classifier is None:
+            _load_fasttext_module()
+
+        if nsfw_classifier is None:
+            nsfw_model_path = resolve_classifier_path(
+                "NSFW",
+                "LTX2_NSFW_CLASSIFIER_PATH",
+                "jigsaw_fasttext_bigrams_nsfw_final.bin",
+                "/data/classifiers/dolma_fasttext_nsfw_jigsaw_model.bin",
+                "dolma-jigsaw-fasttext-bigrams-nsfw-final.bin",
+            )
+            self._nsfw_classifier = lambda text: fasttext_predict(
+                nsfw_model_path,
+                text,
+                "nsfw",
+            )
+        else:
+            self._nsfw_classifier = nsfw_classifier
+
+        if hate_speech_classifier is None:
+            hate_speech_model_path = resolve_classifier_path(
+                "hate speech",
+                "LTX2_HATESPEECH_CLASSIFIER_PATH",
+                "jigsaw_fasttext_bigrams_hatespeech_final.bin",
+                "/data/classifiers/dolma_fasttext_hatespeech_jigsaw_model.bin",
+                "dolma-jigsaw-fasttext-bigrams-hatespeech-final.bin",
+            )
+            self._hate_speech_classifier = lambda text: fasttext_predict(
+                hate_speech_model_path,
+                text,
+                "hate_speech",
+            )
+        else:
+            self._hate_speech_classifier = hate_speech_classifier
+
+    def get_prompt_safety_error(self, prompt: str) -> str | None:
+        normalized_prompt = prompt.strip()
+        if not normalized_prompt:
+            return None
+
+        nsfw_label, _ = self._nsfw_classifier(normalized_prompt)
+        if _label_matches(
+                nsfw_label,
+                blocked_markers=("nsfw", ),
+                safe_markers=("sfw", "safe"),
+        ):
+            return ("This prompt was flagged as NSFW. "
+                    "You can't generate a video with this specified prompt.")
+
+        hate_label, _ = self._hate_speech_classifier(normalized_prompt)
+        if _label_matches(
+                hate_label,
+                blocked_markers=("hatespeech", "hate", "toxic", "offensive", "abusive"),
+                safe_markers=(
+                    "non_hatespeech",
+                    "not_hatespeech",
+                    "non_toxic",
+                    "not_toxic",
+                    "safe",
+                    "clean",
+                ),
+        ):
+            return ("This prompt was flagged as hate speech. "
+                    "You can't generate a video with this specified prompt.")
+
+        return None
+
+    def get_first_blocked_prompt(
+        self,
+        prompts: list[str],
+    ) -> BlockedPrompt | None:
+        for index, prompt in enumerate(prompts):
+            normalized_prompt = str(prompt or "").strip()
+            if not normalized_prompt:
+                continue
+            error = self.get_prompt_safety_error(normalized_prompt)
+            if error is not None:
+                return BlockedPrompt(
+                    index=index,
+                    prompt=normalized_prompt,
+                    error=error,
+                )
+        return None

apps/dreamverse/dreamverse/prompts/README.md

@@ -0,0 +1,59 @@
+# Prompt Files
+
+This directory contains the editable system prompts used by the
+`PromptEnhancer` in `server/prompt_enhancer.py`.
+
+## `next_segment_system_prompt.md`
+
+Use this prompt for guided continuation.
+
+It is loaded as the "next-segment" system prompt and used by
+`PromptEnhancer.enhance_prompt()` for the normal continuation flow when the
+request includes:
+
+- locked prior segments
+- a new user conditioning prompt describing what should happen next
+
+In that path, the model is asked to write exactly one new segment prompt that
+continues from the locked history while satisfying the user's requested next
+beat.
+
+This is the prompt used for the live non-single-clip enhancement path in
+`server/main.py`.
+
+## `auto_extension_system_prompt.md`
+
+Use this prompt for autonomous prompt expansion.
+
+It is loaded as the "auto-extension" system prompt and used in two different
+paths:
+
+1. `PromptEnhancer.generate_auto_prompt()`
+
+This is the background auto-extension flow. The model only receives locked
+segment history and must infer the next narrative beat on its own.
+
+1. `PromptEnhancer.enhance_prompt()` in single-clip mode
+
+This is the single-clip expansion flow. A short user idea is expanded into one
+standalone detailed 5-second prompt for a single clip.
+
+## `rewrite_user_system_prompt.md`
+
+Use this prompt for new-rollout generation from a user's initial rollout
+instruction.
+
+This prompt is used for the rewrite path when there is no existing prompt
+window yet and the system needs to generate the initial 6 segment prompts from
+scratch.
+
+If this file is missing or empty, the server falls back to
+`rewrite_window_system_prompt.md` so startup does not break while the file is
+still being drafted.
+
+## Practical difference
+
+- `next_segment_system_prompt.md` is for "continue based on the user's new
+  instruction."
+- `auto_extension_system_prompt.md` is for "expand or infer the next clip
+  without that guided continuation input."

apps/dreamverse/dreamverse/prompts/auto_extension_system_prompt.md

@@ -0,0 +1,52 @@
+You are a prompt extender for LTX-2.3 video generation.
+
+Your job is to expand a short user idea into a detailed cinematic prompt for a single 5-second video clip.
+
+LTX-2.3 works best when prompts clearly describe:
+• the subject
+• the action
+• the environment
+• lighting
+• camera behavior
+• audio
+
+Write the scene as one continuous cinematic shot.
+
+Guidelines:
+- Preserve the user’s subject and intent.
+- Add concrete visual details (appearance, materials, setting).
+- Use cinematic language such as medium shot, close-up, slow push-in, pan, tracking shot.
+- Describe motion using clear verbs and visible actions.
+- Express emotion through physical cues rather than internal thoughts.
+- If dialogue is included, put spoken lines in quotation marks and keep them short.
+- Include simple audio when relevant (console beeps, footsteps, rain, room tone).
+- End the prompt with a stable visual frame.
+
+Prompt structure (single paragraph, ~4–8 sentences):
+1. Shot and subject
+2. Environment and lighting
+3. Main action
+4. Small reaction or follow-up beat
+5. Optional camera movement
+6. Audio elements
+7. Stable ending frame
+
+Avoid:
+- scene cuts
+- conflicting lighting
+- overloaded scenes
+- abstract emotional descriptions
+
+Return valid JSON only.
+
+Output exactly one JSON object with this schema:
+{"prompt":"<one detailed 5-second video prompt>"}
+
+Rules:
+- The top-level JSON object must contain exactly one field: "prompt".
+- "prompt" must be a single string.
+- Do not return markdown fences.
+- Do not return commentary, explanations, or any text before or after the JSON.
+- Do not return "next_prompt".
+- Do not return "segment_prompts".
+- Do not return an array.

apps/dreamverse/dreamverse/prompts/next_segment_system_prompt.md

@@ -0,0 +1,105 @@
+You are a prompt writer for LTX-2 video continuation. You write one new
+segment prompt that continues a video from where it left off, guided by
+a user's conditioning prompt.
+
+<context>
+LTX-2 generates video in 6 sequential segments of 5 seconds each
+(30 seconds total). Each segment is generated using the LAST FRAME of
+the previous segment as its starting image. The model has no memory of
+earlier segments - only that single frame. This means:
+- Characters, objects, and settings that are visible in the last frame
+  carry forward naturally.
+- Anything that left the frame (via a scene cut, hard transition, or
+  camera movement away) cannot be recreated - the model has never seen
+  it before.
+- If a segment ends mid-action or with sudden motion, the next segment
+  inherits a blurry or unstable starting frame, which degrades quality.
+</context>
+
+<task>
+You will receive locked segments (already generated or currently
+generating) and a conditioning prompt describing what should happen
+next. Write exactly one new next-segment prompt continuing naturally
+from the last locked segment.
+</task>
+
+<rules>
+<conditioning>
+- Place the user's requested event in this next segment.
+- Complete the requested event within this single 5-second segment.
+- Keep continuity from locked segments while satisfying the request.
+</conditioning>
+
+<writing_style>
+Write the segment as a single flowing paragraph in present tense using
+active language ("is walking", "reaches for", "speaks softly").
+
+Structure the segment with these layers:
+1. Establish the shot using cinematography terms (medium shot, close-up,
+   wide establishing shot).
+2. Set the scene: lighting, color palette, textures, atmosphere.
+3. Describe action as a chronological sequence using temporal connectors
+   ("as", "then", "while").
+4. Define characters through observable features: age, hairstyle,
+   clothing, distinguishing marks.
+5. Weave in an audio layer alongside the action - specific ambient
+   sounds ("the hum of fluorescent lights", "a clock ticking on the
+   wall"), effects, and speech integrated with the visual description.
+6. Place all spoken dialogue in quotation marks. Preserve any dialogue
+   from the user's conditioning prompt exactly as written.
+
+Express emotion through physical cues (clenched fists, trembling lip,
+wide eyes) rather than labels ("sad", "angry"). Describe only what is
+seen and heard - no smell, taste, or internal thoughts. Use restrained,
+natural phrasing. Start directly with scene description.
+
+Keep actions gradual. LTX-2 struggles with sudden, abrupt movements -
+they produce artifacts and blurry frames. Any camera movement within
+the segment should settle to a still frame by the end - the last moment
+should be a stable, static shot.
+</writing_style>
+
+<scene_continuity>
+Static shots held across multiple consecutive segments can cause visual
+artifacts to accumulate in video continuation. Changing the scene can
+refresh the image and reset quality.
+
+There are two types of segments:
+
+1. Continuation segments (most segments): The segment continues from
+   the last frame of the previous segment. Write it as an image-to-video
+   prompt - describe only what changes from the previous scene. Keep
+   characters, setting, and framing consistent with what was already
+   on screen.
+
+2. Cut segments: The segment is an entirely new scene. Write it as a
+   full text-to-video prompt - fully describe the new setting,
+   characters, lighting, and atmosphere from scratch, as if the model
+   has never seen any of it before. Only use a cut when the conditioning
+   prompt naturally requires a scene change.
+
+For late segments (5-6), use only subtle camera movements (slow zoom,
+gentle pan, slight drift) and keep the scene stable for a clean ending.
+</scene_continuity>
+
+<dialogue>
+When a segment lacks significant action or sound, fill the 5 seconds
+with spoken dialogue in quotation marks to keep the scene engaging.
+Characters should react to and reference the conditioning event in
+their dialogue. Weave dialogue throughout the segment alongside action.
+</dialogue>
+
+<length>
+Match the length of the locked segments. Count the sentences in the
+locked segments and write about the same number here. Typically
+3-5 sentences. If the locked segments are short, keep this one short.
+</length>
+</rules>
+
+<output_format>
+Return valid JSON only:
+{
+  "next_prompt": "prompt for the next segment"
+}
+One key only, no markdown fences, no extra keys.
+</output_format>

apps/dreamverse/dreamverse/prompts/rewrite_user_system_prompt.md

@@ -0,0 +1,207 @@
+You are a real-time prompt writer for ltx2, a video generation model for image-audio-to-video continuation.
+
+<inputs>
+You receive:
+1. A user prompt describing a video idea, scene, character moment, joke, action, or story beat
+</inputs>
+
+<context>
+Your job is to expand the user's prompt into a full rollout of 6 sequential segment prompts, each describing 5 seconds of video, for a total of 30 seconds.
+
+Each segment is generated independently but conditioned only on:
+- the last 9 video frames of the previous segment
+- the last 49 audio frames of the previous segment
+
+Important implications:
+- Subjects, props, and scene elements that remain visible in the previous segment's last frame carry forward best.
+- If a subject disappears from frame because of a hard transition or because the camera moves away, that subject should not reappear later unless the user explicitly wants a new reveal and that reveal is plausible from the current frame.
+- Stable end frames improve continuation.
+- The final sentence of each segment should land on a clean, readable visual state.
+- Quiet continuing ambience in the final sentence is fine.
+- Avoid ending a segment with a brand-new major action, a heavy new line of dialogue, or visual chaos.
+</context>
+
+<task>
+Return a complete original 6-segment rollout based on the user's prompt.
+
+You must:
+- Expand the user's idea into a coherent beginning-to-end 30-second rollout.
+- Preserve the user's core concept, tone, style, characters, setting, and requested actions.
+- If the user gives only a broad or simple idea, infer missing details conservatively and add clear staging, scene logic, and pacing.
+- If the user gives a detailed prompt, follow it closely.
+- Keep all 6 segments coherent as one continuous rollout.
+- Return all 6 segment prompts.
+</task>
+
+<instruction_handling>
+Handle the user's prompt in one of these ways:
+
+1. Very broad prompt
+- Expand into a clear, staged rollout with defined characters, setting, and progression.
+
+1. Moderately specific prompt
+- Preserve given details and fill only what is needed for a complete rollout.
+
+1. Highly detailed prompt
+- Follow closely while maintaining clarity, pacing, and continuity.
+</instruction_handling>
+
+<priority_rules>
+When rules conflict, resolve them in this order:
+1. The user's prompt
+2. Continuation plausibility from one segment to the next
+3. Clarity, staging, and visual quality
+4. Default stylistic preferences in this system prompt
+</priority_rules>
+
+<house_style>
+Default to a dialogue-forward, character-centered rollout with clear staging, one readable beat per segment, and strong visual continuity.
+
+Core defaults:
+- One main action + one reaction beat per segment
+- Compact, legible segments
+- One dominant speaker per segment
+- Stable ending frames
+
+Conversation bias:
+- When the prompt supports comedy, character interaction, or everyday scenarios, prioritize conversational beats over pure visual spectacle.
+- Prefer dialogue-driven progression rather than action-only sequences when both are plausible.
+- Use dialogue to reveal character, humor, tension, or situation changes.
+- Keep exchanges short and punchy rather than long or dense.
+- Let visual acting and timing complement the dialogue instead of replacing it.
+
+Exception:
+- If the user explicitly asks for cinematic spectacle, action-heavy sequences, or minimal dialogue, follow that instead.
+</house_style>
+
+<default_rollout_rhythm>
+Unless the user gives different timing, use this structure:
+- Segment 1: establish scene, subject, situation, first speaking beat
+- Segment 2: small escalation or reaction
+- Segment 3: continuation beat
+- Segment 4: pivot, reveal, pan, cut, or new subject focus
+- Segment 5: payoff or response
+- Segment 6: closing button, stable hold
+
+Segment 4 is the default pivot point unless specified otherwise.
+</default_rollout_rhythm>
+
+<segment_length_rules>
+Treat counts as soft targets. Do not pad unnaturally.
+
+- Segment 1: 4–5 sentences, ~95–140 words
+- Segment 2: 3–4 sentences, ~55–95 words
+- Segment 3: 3–4 sentences, ~55–95 words
+- Segment 4: 3–5 sentences, ~85–130 words
+- Segment 5: 3–4 sentences, ~55–95 words
+- Segment 6: 3–4 sentences, ~55–100 words
+
+Adjust if user intent requires.
+</segment_length_rules>
+
+<style_rules>
+- Respect user-specified style.
+- Use a "Style:" prefix only if clearly beneficial or already implied.
+- Keep style consistent across segments.
+</style_rules>
+
+<description_compression_rules>
+- Fully describe scene/subjects on first appearance.
+- Compress repeated details in later segments.
+- Maintain key identity anchors without redundancy.
+- Reintroduce full detail only when scene or subject changes.
+</description_compression_rules>
+
+<segment_prompt_rules>
+Each segment must:
+- Be present tense
+- Describe only visible/audible elements
+- Be one paragraph
+- Match detail to shot scale
+- End on a stable visual frame
+</segment_prompt_rules>
+
+<scene_rules>
+- Maintain coherent setting, layout, lighting
+- Use concrete visual details and textures
+- Avoid conflicting lighting or environment logic
+- Prefer one primary location with optional pivot at segment 4
+</scene_rules>
+
+<subject_rules>
+- Use consistent naming across segments
+- Maintain appearance and identity anchors
+- Introduce new characters clearly once
+- Prefer small number of recurring subjects
+</subject_rules>
+
+<camera_rules>
+- Keep camera language efficient
+- Prefer stable shots unless movement matters
+- Max one clear camera move per segment
+- Describe post-movement composition
+- Use segment 4 for major camera transitions by default
+</camera_rules>
+
+<action_rules>
+- One main action + one reaction beat
+- Keep motion readable and grounded
+- Favor simple, clear gestures
+- Avoid chaotic or overloaded motion
+</action_rules>
+
+<dialogue_rules>
+- Include dialogue in most segments when appropriate (typically 5–6 segments)
+- One dominant speaker per segment
+- 1 short line or 2 clipped mini-lines
+- Keep lines natural and brief
+- Place dialogue before final sentence when possible
+</dialogue_rules>
+
+<audio_rules>
+- Tie audio to visible action/environment
+- Use 1–2 concrete sound cues per segment
+- Maintain audio continuity across segments
+- Avoid introducing new dominant sounds at the end
+</audio_rules>
+
+<emotion_rules>
+- No internal thoughts
+- Use visible cues for emotion
+</emotion_rules>
+
+<continuity_rules>
+- Ensure smooth visual/audio continuity across segments
+- Do not reintroduce off-screen subjects unless justified
+- Stabilize new scenes immediately after transitions
+- End segments with stable compositions
+</continuity_rules>
+
+<text_and_logo_rules>
+- Avoid reliance on readable text unless explicitly requested
+</text_and_logo_rules>
+
+<id_and_label_rules>
+- Generate "id" in snake_case
+- Generate concise descriptive "label"
+</id_and_label_rules>
+
+<output_format>
+Return ONLY valid JSON using exactly this structure:
+
+{
+  "id": "...",
+  "label": "...",
+  "segment_prompts": [
+    "segment 1 text",
+    "segment 2 text",
+    "segment 3 text",
+    "segment 4 text",
+    "segment 5 text",
+    "segment 6 text"
+  ]
+}
+
+Do not include explanations, markdown, or additional fields.
+Only output the JSON object.
+</output_format>

apps/dreamverse/dreamverse/prompts/rewrite_window_system_prompt.md

@@ -0,0 +1,255 @@
+You are a real-time prompt editor for ltx2, a video generation model for image-audio-to-video continuation.
+
+<inputs>
+You receive:
+1. An existing rollout JSON with:
+   - "id"
+   - "label"
+   - "segment_prompts": an array of 6 sequential segment prompts
+2. The user's latest instruction about how to revise the rollout
+</inputs>
+
+<context>
+The rollout contains 6 sequential segment prompts, each describing 5 seconds of video, for a total of 30 seconds.
+
+Each segment is generated independently but conditioned only on:
+- the last 9 video frames of the previous segment
+- the last 49 audio frames of the previous segment
+
+Important implications:
+- Subjects, props, and scene elements that remain visible in the previous segment's last frame carry forward best.
+- If a subject disappears from frame because of a hard transition or because the camera moves away, that subject should not reappear later unless the user explicitly wants a new reveal and that reveal is plausible from the current frame.
+- Stable end frames improve continuation.
+- The final sentence of each segment should land on a clean, readable visual state.
+- Quiet continuing ambience in the final sentence is fine.
+- Avoid ending a segment with a brand-new major action, a heavy new line of dialogue, or visual chaos.
+</context>
+
+<task>
+Return a complete revised 6-segment rollout.
+
+You must:
+- Preserve as much of the existing rollout as possible unless the user asked to change it or it causes logical inconsistency, continuity problems, or common-sense failure.
+- Adjust any number of segments as needed so the full 6-segment rollout stays coherent.
+- Preserve unaffected details, pacing, camera logic, setting, props, and subject identity unless the user explicitly changes them or they become inconsistent.
+- Revise all dependent details when one attribute changes.
+- Return all 6 segment prompts, even if only one segment changes.
+</task>
+
+<instruction_handling>
+Handle the user's latest instruction in one of these ways:
+
+1. No actionable rollout instruction
+- If the user's latest instruction does not actually request a change to the rollout, return the existing rollout unchanged.
+
+1. General instruction
+- If the user's latest instruction is broad or high-level, expand it into a more detailed, coherent rollout while preserving existing details wherever possible.
+
+1. Detailed instruction
+- If the user's latest instruction is specific and detailed, follow it closely while preserving continuity and physical plausibility.
+</instruction_handling>
+
+<priority_rules>
+When rules conflict, resolve them in this order:
+1. The user's latest instruction
+2. Continuation plausibility from the previous segment's last visible frame and recent audio
+3. Preservation of the existing rollout
+4. Default stylistic preferences in this system prompt
+</priority_rules>
+
+<house_style>
+Default to a dialogue-forward, character-centered rollout with clear staging, one readable beat per segment, and strong visual continuity.
+
+Common defaults:
+- Use one main action plus one follow-up reaction beat per segment.
+- Keep most segments compact and legible.
+- Include dialogue in most segments unless the user explicitly wants a quiet, purely visual, or action-only rollout.
+- Usually keep one speaking subject dominant within a segment.
+- Favor clean, stable ending images over flashy exits.
+</house_style>
+
+<default_rollout_rhythm>
+Unless the user gives different timing, use this as the default 6-segment rhythm:
+- Segment 1: establish the setting, the main subject, the core situation, and the first speaking beat
+- Segment 2: small escalation, reaction, or new piece of information
+- Segment 3: continue the situation with one more beat of action or reaction
+- Segment 4: visual refresh, reveal, pivot, pan, cut, nearby location change, or new subject focus
+- Segment 5: payoff, response, or aftermath in the refreshed composition
+- Segment 6: closing button and stable held ending
+
+This is a default pattern, not a hard requirement. If the user specifies different timing, follow the user.
+</default_rollout_rhythm>
+
+<segment_length_rules>
+Treat sentence and word counts as soft pacing targets, not hard quotas. Do not pad or compress unnaturally just to hit counts.
+
+Default pacing:
+- Segment 1: usually 4 to 5 sentences, roughly 95 to 140 words
+- Segment 2: usually 3 to 4 sentences, roughly 55 to 95 words
+- Segment 3: usually 3 to 4 sentences, roughly 55 to 95 words
+- Segment 4: usually 3 to 5 sentences, roughly 85 to 130 words
+- Segment 5: usually 3 to 4 sentences, roughly 55 to 95 words
+- Segment 6: usually 3 to 4 sentences, roughly 55 to 100 words
+
+If the user requests a slower, denser, faster, quieter, or more cinematic rollout, adjust these ranges as needed while preserving clarity.
+</segment_length_rules>
+
+<style_rules>
+- Preserve the rollout's existing style-marker pattern whenever possible.
+- If the existing rollout consistently starts segments with a style prefix such as "Style: ...", keep that pattern.
+- If the existing rollout does not use a style prefix, do not add one unless the user explicitly asks for a style change or the rollout needs a new clear style cue.
+- Keep the visual style consistent across all 6 segments unless the user explicitly changes it.
+</style_rules>
+
+<description_compression_rules>
+- When a scene, subject, or important prop first appears, describe it clearly and specifically.
+- In later segments within the same scene, compress repeated details and restate only the key anchors needed for continuity, identity, and image quality.
+- Do not fully re-describe the same room, outfit, prop, or character in every segment unless the user explicitly wants that repetition.
+- When a new location appears, treat that segment as a fresh introduction for the new location.
+- When a new subject appears, describe that subject clearly on first appearance, then use stable shorthand afterward.
+</description_compression_rules>
+
+<segment_prompt_rules>
+Each segment prompt must:
+- Be written in present tense.
+- Describe only what is seen and heard.
+- Avoid internal thoughts, abstract emotions, or motivations unless shown through visible or audible cues.
+- Be a single flowing paragraph.
+- Match the amount of detail to the shot scale. Close shots should emphasize facial detail, hands, fabric, texture, and subtle motion. Wide shots should emphasize layout, blocking, and readable movement.
+- End in a stable, readable visual state.
+</segment_prompt_rules>
+
+<scene_rules>
+- Keep the setting, spatial layout, lighting logic, and important props coherent across segments unless the user changes them.
+- When useful, include materials and textures such as glossy plastic, worn fabric, tiled floor, brushed metal, wet pavement, fingerprint-textured clay, or soft fur.
+- Use concrete visual details that help the model stage the scene cleanly.
+- Do not introduce conflicting lighting logic within the same scene.
+</scene_rules>
+
+<subject_rules>
+- Use the same noun for the same subject across all segments.
+- Do not rename the same subject with synonyms in later segments.
+- Keep appearance and wardrobe anchors consistent unless the user changes them.
+- Include enough appearance detail to preserve identity, such as age, hairstyle, clothing, distinguishing features, body type, species, or surface detail when relevant.
+- If a subject speaks or sings, keep voice traits consistent unless the user changes them.
+- If there are multiple recurring subjects, distinguish them with stable identifiers.
+</subject_rules>
+
+<camera_rules>
+- Each segment should imply a clear framing or shot, but camera language should stay efficient.
+- Use explicit camera movement only when it matters.
+- Most non-pivot segments should use stable framing or gentle motion.
+- Prefer at most one deliberate camera move per segment.
+- Describe camera movement relative to the subject when useful.
+- After a pan, push, pull, tilt, whip-pan, or cut, describe what the camera now lands on.
+- If the existing rollout includes explicit camera-transition wording such as:
+  - "The camera whip-pans fast to the right..."
+  - "The camera slowly pans across..."
+  - "The camera pushes in..."
+  preserve that transition wording exactly and keep it in the same segment and same relative location unless the user explicitly asks to change it.
+- Unless the user specifies otherwise, segment 4 is the preferred place for a visual refresh such as a pan, whip-pan, cut, reveal, nearby location shift, or new subject focus.
+</camera_rules>
+
+<action_rules>
+- Keep motion readable and physically plausible.
+- Prefer one main visible action plus one follow-up reaction beat per segment.
+- Keep blocking simple and legible.
+- Favor small clear gestures such as a head tilt, raised eyebrow, folding arms, looking down, stepping forward, crouching, shifting weight, or setting an object down.
+- Avoid overloaded choreography, chaotic physics, or too many simultaneous actions unless the user explicitly wants that complexity.
+</action_rules>
+
+<dialogue_rules>
+- Dialogue is allowed and often useful.
+- For dialogue-forward rollouts, include at least one short spoken beat in most segments, usually 5 or 6 of the 6 segments, unless the user requests silence or a mostly nonverbal sequence.
+- Usually keep one speaker dominant within a segment.
+- Usually use 1 short quoted line or 2 clipped mini-lines by the same speaker.
+- Keep spoken lines short, natural, and easy to act.
+- Put all spoken dialogue in quotation marks.
+- Prefer visible acting around the line, such as a glance, pause, grin, sigh, shrug, or gesture.
+- Avoid long speeches and dense back-and-forth exchanges inside one segment.
+- When possible, place dialogue before the final sentence so the segment can land on a stable visual ending.
+</dialogue_rules>
+
+<audio_rules>
+- Include audio when it helps the scene.
+- Tie sound to visible action, speech, movement, or ongoing ambience.
+- Usually 1 or 2 specific sound cues is enough for a segment.
+- Favor concrete sounds such as hums, clicks, beeps, footsteps, water lapping, crickets, distant chatter, vent hiss, keyboard clacks, birds, sprinkler clicks, or soft music already present in the scene.
+- Keep audio scene-appropriate and continuation-safe.
+- Quiet ambient audio can continue into the final sentence, but avoid introducing a brand-new dominant sound at the very end.
+</audio_rules>
+
+<emotion_rules>
+- Do not describe internal thoughts.
+- Prefer visible and audible cues over abstract emotional labels.
+- Use posture, expression, gaze, timing, breathing, hand movement, and vocal delivery instead of unsupported inner-state narration.
+</emotion_rules>
+
+<continuity_rules>
+- Later segments must follow naturally from what is plausibly visible and audible from the previous segment's ending.
+- Do not reintroduce subjects that are no longer visible after a major transition unless the user explicitly requests it and the reintroduction is plausible.
+- If a major scene transition occurs, re-establish the new scene clearly in that same segment so later segments can continue from it.
+- Good ending images include a held pose, a settled camera, a quiet look, a character standing still, a character seated calmly, or a clean locked composition.
+- Avoid ending on blur, sudden subject exit, unresolved camera motion, or a fresh unresolved event.
+</continuity_rules>
+
+<text_and_logo_rules>
+- Avoid making readable text, signage, or logos the main point of the scene unless the user explicitly asks for it or the existing rollout already uses it successfully.
+- If preserving existing readable text details, keep them short and simple.
+</text_and_logo_rules>
+
+<rewrite_biases>
+- Prefer the smallest set of edits that fully satisfies the user's latest instruction.
+- Preserve the existing rollout's successful pacing, rhythm, and density unless the user explicitly asks to change them.
+- Preserve the rollout's existing structural asymmetry when it is already working well, such as a fuller segment 1, a pivot or reveal around segment 4, and shorter compressed later segments.
+- If a segment already contains a clear spoken beat and the user did not ask to remove dialogue, preserve dialogue in that segment or replace it with a similarly short spoken beat.
+- Preserve which subject is dominant in each segment unless the user explicitly changes the focus.
+- Preserve stable framing in non-pivot segments when possible.
+- Preserve exact camera-transition wording and keep it in the same segment and same relative location unless the user explicitly asks to change it.
+- Preserve shorthand description in later segments when earlier segments already established the scene, subject, and props clearly.
+- When a user change affects one segment, first patch that segment and its immediate neighbors before rewriting the whole rollout.
+- When one anchor changes, propagate only the downstream changes required for continuity, identity, and common sense.
+- Prefer edits that preserve the final held image of each segment whenever possible.
+- When the user asks for a stronger result such as funnier, sharper, warmer, or more dramatic, first strengthen dialogue, reaction beats, visible acting, and timing before adding new props, new characters, or larger scene changes.
+- Preserve the rollout's existing tonal temperature unless the user explicitly asks to change it.
+- Do not add new spectacle, extra dialogue, extra camera movement, or extra scene changes that the user did not request.
+</rewrite_biases>
+
+<avoid>
+Avoid:
+- Fully re-describing the same character or room in every segment
+- Internal emotional narration without visible cues
+- Conflicting lighting logic
+- Overloaded scenes with too many characters or actions
+- Unclear subject naming
+- Sudden unsupported reappearances
+- Long monologues
+- Ending on chaos, blur, or a fresh unresolved event
+</avoid>
+
+<id_and_label_rules>
+- Output an "id" in snake_case.
+- Output a short "label" that matches the current concept.
+- If editing an existing rollout, preserve the existing id and label unless the user's change makes them inaccurate.
+- If they become inaccurate, update them minimally.
+</id_and_label_rules>
+
+<output_format>
+Return ONLY valid JSON using exactly this structure:
+
+{
+  "id": "...",
+  "label": "...",
+  "segment_prompts": [
+    "segment 1 text",
+    "segment 2 text",
+    "segment 3 text",
+    "segment 4 text",
+    "segment 5 text",
+    "segment 6 text"
+  ]
+}
+
+Do not include explanations, markdown, or additional fields.
+Only output the JSON object.
+</output_format>

apps/dreamverse/dreamverse/rewrite_prompt_payload.py

@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+REWRITE_REQUEST_TEXT = ("Rewrite all segment prompts with improved continuity and cinematic detail. "
+                        "Keep count and ordering identical.")
+DEFAULT_REWRITE_SEGMENT_COUNT = 6
+REWRITE_MODE_NEW = "new_rollout"
+REWRITE_MODE_EDIT_EXISTING = "edit_existing_rollout"
+DEFAULT_REWRITE_ROLLOUT_ID = "current_rollout"
+DEFAULT_REWRITE_ROLLOUT_LABEL = "Current rollout"
+
+
+def normalize_prompt_window_prompts(values: list[Any] | None) -> list[str]:
+    if not isinstance(values, list):
+        return []
+    normalized: list[str] = []
+    for item in values:
+        if not isinstance(item, str):
+            continue
+        clean_item = item.strip()
+        if not clean_item:
+            continue
+        normalized.append(clean_item)
+    return normalized
+
+
+def build_rewrite_user_payload(
+    *,
+    prompt_window_prompts: list[str],
+    preset_id: str | None = None,
+    preset_label: str | None = None,
+    rewrite_instruction: str | None = None,
+) -> dict[str, Any]:
+    rollout_id = (preset_id or "").strip() or DEFAULT_REWRITE_ROLLOUT_ID
+    rollout_label = (preset_label or "").strip() or DEFAULT_REWRITE_ROLLOUT_LABEL
+    instruction = (rewrite_instruction.strip() if isinstance(rewrite_instruction, str) else "")
+    if len(prompt_window_prompts) == 0:
+        return {
+            "mode": REWRITE_MODE_NEW,
+            "request": REWRITE_REQUEST_TEXT,
+            "user_instruction": instruction,
+            "desired_segment_count": DEFAULT_REWRITE_SEGMENT_COUNT,
+            "rollout_id_hint": rollout_id,
+            "rollout_label_hint": rollout_label,
+        }
+    return {
+        "mode": REWRITE_MODE_EDIT_EXISTING,
+        "request": REWRITE_REQUEST_TEXT,
+        "user_instruction": instruction,
+        "current_rollout": {
+            "id": rollout_id,
+            "label": rollout_label,
+            "segment_prompts": list(prompt_window_prompts),
+        },
+    }
+
+
+def build_rewrite_request_body(
+    *,
+    system_prompt: str,
+    prompt_window_prompts: list[str],
+    preset_id: str | None,
+    preset_label: str | None,
+    rewrite_instruction: str | None,
+    model: str,
+    temperature: float,
+    max_completion_tokens: int,
+) -> dict[str, Any]:
+    user_payload = build_rewrite_user_payload(
+        prompt_window_prompts=prompt_window_prompts,
+        preset_id=preset_id,
+        preset_label=preset_label,
+        rewrite_instruction=rewrite_instruction,
+    )
+    return {
+        "model":
+        model,
+        "temperature":
+        temperature,
+        "max_completion_tokens":
+        max_completion_tokens,
+        "response_format": {
+            "type": "json_object"
+        },
+        "messages": [
+            {
+                "role": "system",
+                "content": system_prompt,
+            },
+            {
+                "role": "user",
+                "content": json.dumps(user_payload, ensure_ascii=False),
+            },
+        ],
+    }

apps/dreamverse/dreamverse/session/__init__.py

@@ -0,0 +1,5 @@
+"""Per-WebSocket-connection code: state, controller, and helpers.
+
+Nothing in this package outlives a single client session — lifetime matches
+one ``websocket.accept()`` to disconnect cycle.
+"""

apps/dreamverse/dreamverse/session/messages.py

@@ -0,0 +1,21 @@
+"""Dataclasses for the prompt-submission pipeline queues."""
+from __future__ import annotations
+
+from dataclasses import dataclass
+
+
+@dataclass
+class PromptSubmission:
+    prompt_id: str
+    raw_prompt: str
+    created_at_s: float
+
+
+@dataclass
+class ReadyPrompt:
+    prompt: str
+    source: str
+    prompt_id: str | None = None
+    fallback_used: bool = False
+    seed_prompt_index: int | None = None
+    loop_iteration: int | None = None

apps/dreamverse/dreamverse/session_init_image.py

@@ -0,0 +1,101 @@
+# pyright: reportMissingTypeArgument=false, reportArgumentType=false
+from __future__ import annotations
+
+import base64
+import io
+import re
+import shutil
+import tempfile
+from dataclasses import dataclass
+from pathlib import Path
+
+from PIL import Image
+
+MAX_SESSION_INIT_IMAGE_BYTES = 15 * 1024 * 1024
+SUPPORTED_SESSION_INIT_IMAGE_MIME_TYPES = {
+    "image/jpeg",
+    "image/png",
+    "image/webp",
+}
+
+_DATA_URL_RE = re.compile(r"^data:(?P<mime>[-\w.+/]+);base64,(?P<data>[A-Za-z0-9+/=\s]+)$")
+
+
+@dataclass(frozen=True)
+class SessionInitImage:
+    file_path: Path
+    temp_dir: Path
+    display_name: str
+    mime_type: str
+
+
+def _sanitize_display_name(raw_name: object) -> str:
+    text = str(raw_name or "").strip()
+    if not text:
+        return "uploaded-image"
+    return Path(text).name or "uploaded-image"
+
+
+def persist_session_init_image(
+    payload: object,
+    *,
+    temp_root: Path | None = None,
+) -> SessionInitImage | None:
+    if payload is None:
+        return None
+    if not isinstance(payload, dict):
+        raise ValueError("initial_image must be an object.")
+
+    data_url = str(payload.get("data_url") or "").strip()
+    if not data_url:
+        return None
+
+    data_url_match = _DATA_URL_RE.match(data_url)
+    if data_url_match is None:
+        raise ValueError("initial_image.data_url must be a base64 data URL.")
+
+    data_url_mime_type = data_url_match.group("mime").strip().lower()
+    mime_type = str(payload.get("mime_type") or data_url_mime_type).strip().lower()
+    if mime_type != data_url_mime_type:
+        raise ValueError("initial_image.mime_type must match the data URL mime type.")
+    if mime_type not in SUPPORTED_SESSION_INIT_IMAGE_MIME_TYPES:
+        raise ValueError("initial_image must be a PNG, JPEG, or WebP image.")
+
+    encoded_bytes = re.sub(r"\s+", "", data_url_match.group("data"))
+    try:
+        raw_bytes = base64.b64decode(encoded_bytes, validate=True)
+    except Exception as exc:
+        raise ValueError("initial_image.data_url is not valid base64 data.") from exc
+
+    if len(raw_bytes) == 0:
+        raise ValueError("initial_image.data_url did not contain image bytes.")
+    if len(raw_bytes) > MAX_SESSION_INIT_IMAGE_BYTES:
+        raise ValueError("initial_image must be 15 MB or smaller.")
+
+    try:
+        with Image.open(io.BytesIO(raw_bytes)) as image:
+            image.load()
+            normalized = image.convert("RGBA" if "A" in image.getbands() else "RGB")
+    except Exception as exc:
+        raise ValueError("initial_image must decode as a valid image.") from exc
+
+    temp_dir = Path(
+        tempfile.mkdtemp(
+            prefix="ltx2_session_init_",
+            dir=str(temp_root) if temp_root is not None else None,
+        ))
+    file_path = temp_dir / "initial_frame.png"
+    normalized.save(file_path, format="PNG")
+
+    return SessionInitImage(
+        file_path=file_path,
+        temp_dir=temp_dir,
+        display_name=_sanitize_display_name(payload.get("name")),
+        mime_type=mime_type,
+    )
+
+
+def cleanup_session_init_image(session_image: SessionInitImage | None) -> None:
+    if session_image is None:
+        return
+    shutil.rmtree(session_image.temp_dir, ignore_errors=True)

apps/dreamverse/dreamverse/session_logger.py

@@ -0,0 +1,46 @@
+# pyright: reportMissingTypeArgument=false
+from __future__ import annotations
+
+import asyncio
+from datetime import datetime, timezone
+import json
+import socket
+from pathlib import Path
+from typing import Any
+
+
+def _utc_now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+class SessionEventLogger:
+
+    def __init__(self, root_dir: Path):
+        self.hostname = socket.gethostname()
+        timestamp = datetime.now(timezone.utc).strftime("%y%m%d_%H%M%S")
+        self.directory = root_dir / self.hostname
+        self.path = self.directory / f"{timestamp}.jsonl"
+        self._lock = asyncio.Lock()
+
+        self.directory.mkdir(parents=True, exist_ok=True)
+        self.path.touch(exist_ok=False)
+
+    async def write_event(
+        self,
+        *,
+        event: str,
+        client_id: str,
+        payload: dict[str, Any] | None = None,
+    ) -> None:
+        entry = {
+            "ts": _utc_now_iso(),
+            "event": event,
+            "hostname": self.hostname,
+            "client_id": client_id,
+        }
+        if payload:
+            entry.update(payload)
+
+        async with self._lock:
+            with self.path.open("a", encoding="utf-8") as fp:
+                fp.write(json.dumps(entry, ensure_ascii=False) + "\n")

apps/dreamverse/dreamverse/tests/test_realtime_stress.py

@@ -0,0 +1,951 @@
+"""Multiprocess realtime stress test for LTX2 streaming websocket service."""
+# pyright: reportArgumentType=false, reportOptionalMemberAccess=false
+
+from __future__ import annotations
+
+import argparse
+import asyncio
+from collections import defaultdict
+from datetime import datetime, timezone
+import json
+import math
+import multiprocessing as mp
+import os
+from pathlib import Path
+import sys
+import time
+import traceback
+from typing import Any
+import uuid
+
+import pytest
+
+pytestmark = pytest.mark.gpu
+
+try:
+    import websockets
+except ModuleNotFoundError:
+    websockets = None  # type: ignore[assignment]
+
+
+DEFAULT_PRESET_FILE = (
+    Path(__file__).resolve().parents[2]
+    / "web"
+    / "prompts"
+    / "selected_ltx2_continuation_story_presets.json"
+)
+
+
+def utc_now_iso() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def iso_from_epoch(epoch_s: float) -> str:
+    return datetime.fromtimestamp(epoch_s, tz=timezone.utc).isoformat()
+
+
+def iso_to_epoch(iso_ts: str | None) -> float | None:
+    if not iso_ts:
+        return None
+    try:
+        return datetime.fromisoformat(iso_ts).timestamp()
+    except ValueError:
+        return None
+
+
+def safe_percentile(values: list[float], percentile: float) -> float | None:
+    if not values:
+        return None
+    sorted_values = sorted(values)
+    if len(sorted_values) == 1:
+        return sorted_values[0]
+    rank = (len(sorted_values) - 1) * (percentile / 100.0)
+    lower = math.floor(rank)
+    upper = math.ceil(rank)
+    if lower == upper:
+        return sorted_values[lower]
+    fraction = rank - lower
+    return (
+        sorted_values[lower]
+        + (sorted_values[upper] - sorted_values[lower]) * fraction
+    )
+
+
+def summarize_series(values: list[float]) -> dict[str, float | int | None]:
+    if not values:
+        return {
+            "count": 0,
+            "min": None,
+            "p50": None,
+            "p95": None,
+            "p99": None,
+            "max": None,
+            "avg": None,
+        }
+    return {
+        "count": len(values),
+        "min": min(values),
+        "p50": safe_percentile(values, 50),
+        "p95": safe_percentile(values, 95),
+        "p99": safe_percentile(values, 99),
+        "max": max(values),
+        "avg": sum(values) / len(values),
+    }
+
+
+def parse_int(value: object) -> int | None:
+    try:
+        parsed = int(value)
+    except (TypeError, ValueError):
+        return None
+    return parsed
+
+
+def format_num(value: float | int | None, digits: int = 2) -> str:
+    if value is None:
+        return "n/a"
+    if isinstance(value, int):
+        return str(value)
+    return f"{value:.{digits}f}"
+
+
+def load_curated_prompts(
+    preset_file: Path,
+    preset_id: str | None,
+    curated_limit: int,
+) -> tuple[str, list[str], int]:
+    if not preset_file.is_file():
+        raise ValueError(f"Preset file not found: {preset_file}")
+    try:
+        payload = json.loads(preset_file.read_text(encoding="utf-8"))
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"Invalid JSON in preset file: {preset_file}") from exc
+
+    if not isinstance(payload, list) or len(payload) == 0:
+        raise ValueError("Preset file must contain a non-empty JSON array.")
+
+    selected: dict[str, Any] | None = None
+    if preset_id:
+        for item in payload:
+            if not isinstance(item, dict):
+                continue
+            if str(item.get("id", "")).strip() == preset_id:
+                selected = item
+                break
+        if selected is None:
+            raise ValueError(f"Preset id not found: {preset_id}")
+    else:
+        for item in payload:
+            if isinstance(item, dict):
+                selected = item
+                break
+        if selected is None:
+            raise ValueError("No valid preset object found in preset file.")
+
+    selected_id = str(selected.get("id", "")).strip() or "unknown_preset"
+    raw_prompts = selected.get("segment_prompts", [])
+    if not isinstance(raw_prompts, list):
+        raise ValueError(
+            f"Preset {selected_id} has invalid segment_prompts (must be list)."
+        )
+
+    prompts = [
+        str(prompt).strip()
+        for prompt in raw_prompts
+        if isinstance(prompt, str) and str(prompt).strip()
+    ]
+    if not prompts:
+        raise ValueError(f"Preset {selected_id} has no non-empty prompts.")
+
+    limited = prompts[:curated_limit]
+    if not limited:
+        raise ValueError(
+            f"curated_limit={curated_limit} produced no prompts for preset "
+            f"{selected_id}."
+        )
+    return selected_id, limited, len(prompts)
+
+
+async def run_single_session(
+    *,
+    worker_id: int,
+    worker_session_idx: int,
+    config: dict[str, Any],
+) -> dict[str, Any]:
+    session_id = f"w{worker_id}_u{worker_session_idx}_{uuid.uuid4().hex[:8]}"
+    process_id = os.getpid()
+    url = str(config["url"])
+    session_timeout_s = float(config["session_timeout_s"])
+    post_complete_wait_s = float(config["post_complete_wait_s"])
+    connect_timeout_s = float(config["connect_timeout_s"])
+    curated_prompts = list(config["curated_prompts"])
+    preset_id = str(config["preset_id"])
+
+    session_start_monotonic = time.monotonic()
+    session_start_epoch = time.time()
+    session_data: dict[str, Any] = {
+        "session_id": session_id,
+        "process_id": process_id,
+        "worker_id": worker_id,
+        "status": "failed",
+        "error": None,
+        "preset_id": preset_id,
+        "curated_prompt_count": len(curated_prompts),
+        "connect_start_ts_utc": iso_from_epoch(session_start_epoch),
+        "connect_finish_ts_utc": None,
+        "session_init_sent_ts_utc": None,
+        "gpu_assigned_ts_utc": None,
+        "target_segment_complete_ts_utc": None,
+        "leave_sent_ts_utc": None,
+        "close_ts_utc": None,
+        "duration_ms": None,
+        "queue_wait_ms": None,
+        "initial_total_segments": None,
+        "segments_started": 0,
+        "segments_completed": 0,
+        "media_segments_completed": 0,
+        "total_chunks": 0,
+        "total_chunk_bytes": 0,
+        "first_chunk_finish_ts_utc": None,
+        "last_chunk_finish_ts_utc": None,
+        "first_media_segment_complete_ts_utc": None,
+        "first_chunk_before_first_media_complete": None,
+        "session_goodput_mbps": None,
+        "chunks": [],
+    }
+
+    connect_finish_monotonic: float | None = None
+    current_segment_idx: int | None = None
+    initial_total_segments: int | None = None
+    first_chunk_finish_epoch: float | None = None
+    first_media_segment_complete_epoch: float | None = None
+    last_chunk_finish_epoch: float | None = None
+    last_chunk_finish_monotonic: float | None = None
+
+    try:
+        async with websockets.connect(
+            url,
+            max_size=None,
+            ping_interval=None,
+            open_timeout=connect_timeout_s,
+            close_timeout=2.0,
+        ) as ws:
+            connect_finish_monotonic = time.monotonic()
+            session_data["connect_finish_ts_utc"] = utc_now_iso()
+
+            init_payload = {
+                "type": "session_init_v2",
+                "preset_id": preset_id,
+                "curated_prompts": curated_prompts,
+                "enhancement_enabled": False,
+                "auto_extension_enabled": False,
+                "loop_generation_enabled": False,
+            }
+            await ws.send(json.dumps(init_payload))
+            session_data["session_init_sent_ts_utc"] = utc_now_iso()
+
+            while True:
+                elapsed_s = time.monotonic() - session_start_monotonic
+                timeout_remaining = session_timeout_s - elapsed_s
+                if timeout_remaining <= 0:
+                    session_data["status"] = "timeout"
+                    session_data["error"] = (
+                        f"Session timed out after {session_timeout_s:.1f}s."
+                    )
+                    break
+
+                recv_start_epoch = time.time()
+                recv_start_monotonic = time.monotonic()
+                recv_start_iso = iso_from_epoch(recv_start_epoch)
+
+                try:
+                    message = await asyncio.wait_for(
+                        ws.recv(),
+                        timeout=timeout_remaining,
+                    )
+                except asyncio.TimeoutError:
+                    session_data["status"] = "timeout"
+                    session_data["error"] = (
+                        "Timed out waiting for websocket message."
+                    )
+                    break
+                except Exception as exc:
+                    session_data["status"] = "failed"
+                    session_data["error"] = f"WebSocket receive failed: {exc}"
+                    break
+
+                recv_finish_epoch = time.time()
+                recv_finish_monotonic = time.monotonic()
+                recv_finish_iso = iso_from_epoch(recv_finish_epoch)
+
+                if isinstance(message, bytes):
+                    session_data["total_chunks"] += 1
+                    session_data["total_chunk_bytes"] += len(message)
+
+                    if first_chunk_finish_epoch is None:
+                        first_chunk_finish_epoch = recv_finish_epoch
+                        session_data["first_chunk_finish_ts_utc"] = recv_finish_iso
+
+                    chunk_gap_ms: float | None = None
+                    if last_chunk_finish_monotonic is not None:
+                        chunk_gap_ms = (
+                            recv_finish_monotonic - last_chunk_finish_monotonic
+                        ) * 1000.0
+
+                    session_data["chunks"].append(
+                        {
+                            "segment_idx": current_segment_idx,
+                            "chunk_idx": session_data["total_chunks"],
+                            "size_bytes": len(message),
+                            "chunk_start_ts_utc": recv_start_iso,
+                            "chunk_finish_ts_utc": recv_finish_iso,
+                            "chunk_gap_ms": chunk_gap_ms,
+                        }
+                    )
+                    last_chunk_finish_monotonic = recv_finish_monotonic
+                    last_chunk_finish_epoch = recv_finish_epoch
+                    session_data["last_chunk_finish_ts_utc"] = recv_finish_iso
+                    continue
+
+                if not isinstance(message, str):
+                    continue
+
+                try:
+                    data = json.loads(message)
+                except json.JSONDecodeError as exc:
+                    session_data["status"] = "protocol_error"
+                    session_data["error"] = f"Invalid JSON message: {exc}"
+                    break
+
+                msg_type = data.get("type")
+                if msg_type == "gpu_assigned":
+                    session_data["gpu_assigned_ts_utc"] = recv_finish_iso
+                    if connect_finish_monotonic is not None:
+                        session_data["queue_wait_ms"] = (
+                            recv_finish_monotonic - connect_finish_monotonic
+                        ) * 1000.0
+                elif msg_type == "ltx2_stream_start":
+                    if initial_total_segments is None:
+                        parsed_total = parse_int(data.get("total_segments"))
+                        if parsed_total is not None and parsed_total > 0:
+                            initial_total_segments = parsed_total
+                            session_data["initial_total_segments"] = parsed_total
+                elif msg_type == "ltx2_segment_start":
+                    parsed_idx = parse_int(data.get("segment_idx"))
+                    current_segment_idx = parsed_idx
+                    session_data["segments_started"] += 1
+                elif msg_type == "media_segment_complete":
+                    session_data["media_segments_completed"] += 1
+                    if first_media_segment_complete_epoch is None:
+                        first_media_segment_complete_epoch = recv_finish_epoch
+                        session_data[
+                            "first_media_segment_complete_ts_utc"
+                        ] = recv_finish_iso
+                elif msg_type == "ltx2_segment_complete":
+                    session_data["segments_completed"] += 1
+                    seg_idx = parse_int(data.get("segment_idx"))
+                    if (
+                        initial_total_segments is not None
+                        and seg_idx is not None
+                        and seg_idx >= initial_total_segments
+                    ):
+                        session_data[
+                            "target_segment_complete_ts_utc"
+                        ] = recv_finish_iso
+                        await asyncio.sleep(post_complete_wait_s)
+                        session_data["leave_sent_ts_utc"] = utc_now_iso()
+                        try:
+                            await ws.send(json.dumps({"type": "leave"}))
+                        except Exception:
+                            pass
+                        session_data["status"] = "success"
+                        break
+                elif msg_type == "session_timeout":
+                    session_data["status"] = "timeout"
+                    session_data["error"] = str(
+                        data.get("message") or "Backend session timeout"
+                    )
+                    break
+                elif msg_type == "error":
+                    session_data["status"] = "failed"
+                    session_data["error"] = str(
+                        data.get("message") or "Backend error message"
+                    )
+                    break
+
+            if session_data["status"] == "failed" and session_data["error"] is None:
+                session_data["error"] = "Session ended without success."
+    except Exception as exc:
+        session_data["status"] = "failed"
+        session_data["error"] = f"WebSocket connect/run failed: {exc}"
+
+    if (
+        first_chunk_finish_epoch is not None
+        and last_chunk_finish_epoch is not None
+        and session_data["total_chunk_bytes"] > 0
+    ):
+        duration_s = last_chunk_finish_epoch - first_chunk_finish_epoch
+        if duration_s > 0:
+            session_data["session_goodput_mbps"] = (
+                session_data["total_chunk_bytes"] * 8.0 / duration_s / 1_000_000.0
+            )
+
+    if (
+        first_chunk_finish_epoch is not None
+        and first_media_segment_complete_epoch is not None
+    ):
+        session_data["first_chunk_before_first_media_complete"] = (
+            first_chunk_finish_epoch < first_media_segment_complete_epoch
+        )
+
+    session_data["close_ts_utc"] = utc_now_iso()
+    session_data["duration_ms"] = (
+        time.monotonic() - session_start_monotonic
+    ) * 1000.0
+    return session_data
+
+
+async def run_worker_sessions(
+    *,
+    worker_id: int,
+    session_count: int,
+    config: dict[str, Any],
+) -> list[dict[str, Any]]:
+    tasks = [
+        asyncio.create_task(
+            run_single_session(
+                worker_id=worker_id,
+                worker_session_idx=idx,
+                config=config,
+            )
+        )
+        for idx in range(session_count)
+    ]
+    if not tasks:
+        return []
+    return await asyncio.gather(*tasks)
+
+
+def worker_entry(
+    worker_id: int,
+    session_count: int,
+    config: dict[str, Any],
+    start_event: Any,
+    ready_queue: Any,
+    result_queue: Any,
+) -> None:
+    try:
+        ready_queue.put({"worker_id": worker_id, "status": "ready"})
+        start_event.wait()
+        sessions = asyncio.run(
+            run_worker_sessions(
+                worker_id=worker_id,
+                session_count=session_count,
+                config=config,
+            )
+        )
+        result_queue.put(
+            {
+                "worker_id": worker_id,
+                "status": "ok",
+                "sessions": sessions,
+            }
+        )
+    except Exception as exc:
+        result_queue.put(
+            {
+                "worker_id": worker_id,
+                "status": "error",
+                "error": str(exc),
+                "traceback": traceback.format_exc(),
+            }
+        )
+
+
+def build_summary(
+    *,
+    sessions: list[dict[str, Any]],
+    chunk_gap_threshold_ms: float,
+) -> dict[str, Any]:
+    status_counts: dict[str, int] = defaultdict(int)
+    chunk_gaps: list[float] = []
+    queue_waits: list[float] = []
+    session_goodputs: list[float] = []
+    progressive_eligible = 0
+    progressive_success = 0
+
+    all_chunk_finish_epochs: list[float] = []
+    bucket_bytes: dict[int, int] = defaultdict(int)
+    total_chunk_bytes = 0
+
+    for session in sessions:
+        status = str(session.get("status") or "unknown")
+        status_counts[status] += 1
+
+        queue_wait_ms = session.get("queue_wait_ms")
+        if isinstance(queue_wait_ms, (int, float)):
+            queue_waits.append(float(queue_wait_ms))
+
+        session_goodput = session.get("session_goodput_mbps")
+        if isinstance(session_goodput, (int, float)):
+            session_goodputs.append(float(session_goodput))
+
+        progressive_value = session.get("first_chunk_before_first_media_complete")
+        if isinstance(progressive_value, bool):
+            progressive_eligible += 1
+            if progressive_value:
+                progressive_success += 1
+
+        for chunk in session.get("chunks", []):
+            gap = chunk.get("chunk_gap_ms")
+            if isinstance(gap, (int, float)):
+                chunk_gaps.append(float(gap))
+
+            size_bytes = int(chunk.get("size_bytes") or 0)
+            finish_epoch = iso_to_epoch(chunk.get("chunk_finish_ts_utc"))
+            if finish_epoch is None or size_bytes <= 0:
+                continue
+            total_chunk_bytes += size_bytes
+            all_chunk_finish_epochs.append(finish_epoch)
+            bucket_bytes[int(finish_epoch)] += size_bytes
+
+    chunk_gap_stats = summarize_series(chunk_gaps)
+    queue_wait_stats = summarize_series(queue_waits)
+    session_goodput_stats = summarize_series(session_goodputs)
+
+    global_goodput_mbps: float | None = None
+    if len(all_chunk_finish_epochs) >= 2 and total_chunk_bytes > 0:
+        duration_s = max(all_chunk_finish_epochs) - min(all_chunk_finish_epochs)
+        if duration_s > 0:
+            global_goodput_mbps = (
+                total_chunk_bytes * 8.0 / duration_s / 1_000_000.0
+            )
+
+    bucket_throughputs_mbps = [
+        (bytes_count * 8.0) / 1_000_000.0
+        for _, bytes_count in sorted(bucket_bytes.items())
+    ]
+    bucket_stats = summarize_series(bucket_throughputs_mbps)
+
+    chunk_gap_threshold_breaches = [
+        value for value in chunk_gaps if value >= chunk_gap_threshold_ms
+    ]
+    non_success = len(sessions) - status_counts.get("success", 0)
+
+    fail_reasons: list[str] = []
+    if non_success > 0:
+        fail_reasons.append(
+            f"{non_success} session(s) did not complete successfully."
+        )
+    if not chunk_gaps:
+        fail_reasons.append("No chunk gap data collected.")
+    if chunk_gap_threshold_breaches:
+        fail_reasons.append(
+            f"{len(chunk_gap_threshold_breaches)} chunk gap(s) were >= "
+            f"{chunk_gap_threshold_ms:.0f}ms."
+        )
+
+    passed = len(fail_reasons) == 0
+    progressive_ratio = None
+    if progressive_eligible > 0:
+        progressive_ratio = progressive_success / progressive_eligible
+
+    return {
+        "passed": passed,
+        "fail_reasons": fail_reasons,
+        "sessions": {
+            "total": len(sessions),
+            "success": status_counts.get("success", 0),
+            "failed": status_counts.get("failed", 0),
+            "timeout": status_counts.get("timeout", 0),
+            "protocol_error": status_counts.get("protocol_error", 0),
+            "other": (
+                len(sessions)
+                - (
+                    status_counts.get("success", 0)
+                    + status_counts.get("failed", 0)
+                    + status_counts.get("timeout", 0)
+                    + status_counts.get("protocol_error", 0)
+                )
+            ),
+        },
+        "chunk_gap_ms": {
+            **chunk_gap_stats,
+            "threshold_ms": chunk_gap_threshold_ms,
+            "breach_count": len(chunk_gap_threshold_breaches),
+        },
+        "queue_wait_ms": queue_wait_stats,
+        "progressive_streaming": {
+            "eligible_sessions": progressive_eligible,
+            "success_sessions": progressive_success,
+            "ratio": progressive_ratio,
+        },
+        "bandwidth_mbps": {
+            "per_session": session_goodput_stats,
+            "global_goodput_mbps": global_goodput_mbps,
+            "bucketed_1s": {
+                "count": bucket_stats.get("count"),
+                "avg_mbps": bucket_stats.get("avg"),
+                "peak_mbps": bucket_stats.get("max"),
+            },
+        },
+    }
+
+
+def print_summary(
+    *,
+    run_info: dict[str, Any],
+    summary: dict[str, Any],
+) -> None:
+    sessions = summary["sessions"]
+    chunk_gap = summary["chunk_gap_ms"]
+    queue_wait = summary["queue_wait_ms"]
+    progressive = summary["progressive_streaming"]
+    bandwidth = summary["bandwidth_mbps"]
+    per_session_bw = bandwidth["per_session"]
+    bucket_bw = bandwidth["bucketed_1s"]
+
+    print("=== LTX2 Realtime Stress Test Summary ===")
+    print(
+        "Run: "
+        f"url={run_info['url']} clients={run_info['clients']} "
+        f"processes={run_info['processes']} "
+        f"preset={run_info['preset_id']} "
+        f"curated_limit={run_info['curated_limit']}"
+    )
+    print(
+        "Sessions: "
+        f"total={sessions['total']} success={sessions['success']} "
+        f"failed={sessions['failed']} timeout={sessions['timeout']} "
+        f"protocol_error={sessions['protocol_error']}"
+    )
+    print(
+        "Chunk gap ms: "
+        f"min={format_num(chunk_gap['min'])} "
+        f"p50={format_num(chunk_gap['p50'])} "
+        f"p95={format_num(chunk_gap['p95'])} "
+        f"p99={format_num(chunk_gap['p99'])} "
+        f"max={format_num(chunk_gap['max'])} "
+        f"threshold={format_num(chunk_gap['threshold_ms'])} "
+        f"breaches={chunk_gap['breach_count']}"
+    )
+    print(
+        "Queue wait ms: "
+        f"min={format_num(queue_wait['min'])} "
+        f"p50={format_num(queue_wait['p50'])} "
+        f"p95={format_num(queue_wait['p95'])} "
+        f"max={format_num(queue_wait['max'])}"
+    )
+    ratio = progressive["ratio"]
+    ratio_text = "n/a" if ratio is None else f"{ratio * 100:.2f}%"
+    print(
+        "Progressive streaming: "
+        f"{progressive['success_sessions']}/"
+        f"{progressive['eligible_sessions']} ({ratio_text})"
+    )
+    print(
+        "Bandwidth Mbps: "
+        f"per_session_avg={format_num(per_session_bw['avg'])} "
+        f"per_session_p95={format_num(per_session_bw['p95'])} "
+        f"global={format_num(bandwidth['global_goodput_mbps'])} "
+        f"bucket_avg={format_num(bucket_bw['avg_mbps'])} "
+        f"bucket_peak={format_num(bucket_bw['peak_mbps'])}"
+    )
+    print(f"VERDICT: {'PASS' if summary['passed'] else 'FAIL'}")
+    if summary["fail_reasons"]:
+        print("Fail reasons:")
+        for reason in summary["fail_reasons"]:
+            print(f"- {reason}")
+
+
+def distribute_sessions(total_clients: int, process_count: int) -> list[int]:
+    base = total_clients // process_count
+    remainder = total_clients % process_count
+    counts = []
+    for idx in range(process_count):
+        count = base + (1 if idx < remainder else 0)
+        counts.append(count)
+    return counts
+
+
+def run_stress(args: argparse.Namespace) -> tuple[dict[str, Any], int]:
+    if websockets is None:
+        raise RuntimeError(
+            "Missing dependency: websockets. Install it before running this "
+            "stress test."
+        )
+
+    preset_file = Path(args.preset_file).expanduser().resolve()
+    selected_preset_id, curated_prompts, total_prompt_count = load_curated_prompts(
+        preset_file=preset_file,
+        preset_id=args.preset_id,
+        curated_limit=args.curated_limit,
+    )
+
+    process_count = args.processes
+    if process_count is None:
+        process_count = min(args.clients, os.cpu_count() or 1)
+    process_count = max(1, min(process_count, args.clients))
+
+    mp_ctx = mp.get_context("spawn")
+    start_event = mp_ctx.Event()
+    ready_queue = mp_ctx.Queue()
+    result_queue = mp_ctx.Queue()
+
+    worker_config = {
+        "url": args.url,
+        "preset_id": selected_preset_id,
+        "curated_prompts": curated_prompts,
+        "connect_timeout_s": args.connect_timeout_s,
+        "session_timeout_s": args.session_timeout_s,
+        "post_complete_wait_s": args.post_complete_wait_s,
+    }
+
+    session_counts = distribute_sessions(args.clients, process_count)
+    run_start_epoch = time.time()
+    run_start_monotonic = time.monotonic()
+    run_start_iso = iso_from_epoch(run_start_epoch)
+
+    processes: list[mp.Process] = []
+    for worker_id, session_count in enumerate(session_counts):
+        proc = mp_ctx.Process(
+            target=worker_entry,
+            args=(
+                worker_id,
+                session_count,
+                worker_config,
+                start_event,
+                ready_queue,
+                result_queue,
+            ),
+        )
+        proc.start()
+        processes.append(proc)
+
+    try:
+        ready_workers = 0
+        ready_deadline = time.monotonic() + 60.0
+        while ready_workers < len(processes):
+            timeout_s = max(0.1, ready_deadline - time.monotonic())
+            if timeout_s <= 0:
+                raise RuntimeError("Timed out waiting for workers to become ready.")
+            msg = ready_queue.get(timeout=timeout_s)
+            if msg.get("status") == "ready":
+                ready_workers += 1
+
+        start_event.set()
+
+        result_deadline = (
+            time.monotonic()
+            + args.connect_timeout_s
+            + args.session_timeout_s
+            + args.post_complete_wait_s
+            + 180.0
+        )
+        worker_results: list[dict[str, Any]] = []
+        while len(worker_results) < len(processes):
+            timeout_s = max(0.1, result_deadline - time.monotonic())
+            if timeout_s <= 0:
+                break
+            try:
+                result = result_queue.get(timeout=timeout_s)
+            except Exception:
+                break
+            worker_results.append(result)
+
+        for proc in processes:
+            proc.join(timeout=5.0)
+            if proc.is_alive():
+                proc.terminate()
+                proc.join(timeout=2.0)
+
+        sessions: list[dict[str, Any]] = []
+        worker_errors: list[dict[str, Any]] = []
+        for result in worker_results:
+            if result.get("status") == "ok":
+                sessions.extend(result.get("sessions", []))
+            else:
+                worker_errors.append(
+                    {
+                        "worker_id": result.get("worker_id"),
+                        "error": result.get("error"),
+                        "traceback": result.get("traceback"),
+                    }
+                )
+
+        received_workers = {result.get("worker_id") for result in worker_results}
+        expected_workers = set(range(len(processes)))
+        missing_workers = sorted(expected_workers - received_workers)
+        for worker_id in missing_workers:
+            worker_errors.append(
+                {
+                    "worker_id": worker_id,
+                    "error": "No worker result received.",
+                }
+            )
+
+        run_end_epoch = time.time()
+        run_end_iso = iso_from_epoch(run_end_epoch)
+        run_duration_ms = (time.monotonic() - run_start_monotonic) * 1000.0
+
+        summary = build_summary(
+            sessions=sessions,
+            chunk_gap_threshold_ms=args.chunk_gap_threshold_ms,
+        )
+
+        if worker_errors:
+            summary["passed"] = False
+            summary["fail_reasons"] = list(summary["fail_reasons"]) + [
+                f"{len(worker_errors)} worker error(s) occurred."
+            ]
+
+        output_payload = {
+            "run_info": {
+                "url": args.url,
+                "clients": args.clients,
+                "processes": process_count,
+                "preset_file": str(preset_file),
+                "preset_id": selected_preset_id,
+                "curated_limit": args.curated_limit,
+                "selected_prompt_count": len(curated_prompts),
+                "preset_total_prompt_count": total_prompt_count,
+                "chunk_gap_threshold_ms": args.chunk_gap_threshold_ms,
+                "connect_timeout_s": args.connect_timeout_s,
+                "session_timeout_s": args.session_timeout_s,
+                "post_complete_wait_s": args.post_complete_wait_s,
+                "run_start_ts_utc": run_start_iso,
+                "run_end_ts_utc": run_end_iso,
+                "run_duration_ms": run_duration_ms,
+            },
+            "summary": summary,
+            "full_data": {
+                "sessions": sessions,
+                "worker_errors": worker_errors,
+            },
+        }
+        exit_code = 0 if summary["passed"] else 1
+        return output_payload, exit_code
+    finally:
+        for proc in processes:
+            if proc.is_alive():
+                proc.terminate()
+                proc.join(timeout=1.0)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Multiprocess realtime stress test for LTX2 streaming.",
+    )
+    parser.add_argument(
+        "-u",
+        "--url",
+        required=True,
+        help="WebSocket URL, e.g. wss://your-domain/ws",
+    )
+    parser.add_argument(
+        "-c",
+        "--clients",
+        type=int,
+        required=True,
+        help="Total concurrent virtual users.",
+    )
+    parser.add_argument(
+        "--processes",
+        type=int,
+        default=None,
+        help="Worker process count (default: min(clients, cpu_count)).",
+    )
+    parser.add_argument(
+        "--preset-file",
+        default=str(DEFAULT_PRESET_FILE),
+        help="Path to curated presets JSON file.",
+    )
+    parser.add_argument(
+        "--preset-id",
+        default=None,
+        help="Preset id to use (default: first preset in file).",
+    )
+    parser.add_argument(
+        "--curated-limit",
+        type=int,
+        default=6,
+        help="Number of curated prompts to send from selected preset.",
+    )
+    parser.add_argument(
+        "--chunk-gap-threshold-ms",
+        type=float,
+        default=5000.0,
+        help="Fail if any chunk gap is >= this value.",
+    )
+    parser.add_argument(
+        "--post-complete-wait-s",
+        type=float,
+        default=5.0,
+        help="Seconds to wait after target segment completion before leave.",
+    )
+    parser.add_argument(
+        "--connect-timeout-s",
+        type=float,
+        default=20.0,
+        help="WebSocket connect timeout in seconds.",
+    )
+    parser.add_argument(
+        "--session-timeout-s",
+        type=float,
+        default=180.0,
+        help="Max session runtime per virtual user in seconds.",
+    )
+    parser.add_argument(
+        "-o",
+        "--output-json",
+        required=True,
+        help="Required output JSON path (summary + full data).",
+    )
+    args = parser.parse_args()
+
+    if args.clients <= 0:
+        parser.error("--clients must be > 0")
+    if args.processes is not None and args.processes <= 0:
+        parser.error("--processes must be > 0")
+    if args.curated_limit <= 0:
+        parser.error("--curated-limit must be > 0")
+    if args.chunk_gap_threshold_ms <= 0:
+        parser.error("--chunk-gap-threshold-ms must be > 0")
+    if args.post_complete_wait_s < 0:
+        parser.error("--post-complete-wait-s must be >= 0")
+    if args.connect_timeout_s <= 0:
+        parser.error("--connect-timeout-s must be > 0")
+    if args.session_timeout_s <= 0:
+        parser.error("--session-timeout-s must be > 0")
+    return args
+
+
+def main() -> int:
+    try:
+        args = parse_args()
+        print("Starting LTX2 realtime stress test with the following parameters:")
+        for arg, value in vars(args).items():
+            print(f"  {arg}: {value}")
+        output_payload, exit_code = run_stress(args)
+
+        output_path = Path(args.output_json).expanduser().resolve()
+        output_path.parent.mkdir(parents=True, exist_ok=True)
+        output_path.write_text(
+            json.dumps(output_payload, ensure_ascii=False, indent=2) + "\n",
+            encoding="utf-8",
+        )
+
+        print_summary(
+            run_info=output_payload["run_info"],
+            summary=output_payload["summary"],
+        )
+        return exit_code
+    except SystemExit:
+        raise
+    except Exception as exc:
+        print(f"Error: {exc}", file=sys.stderr)
+        return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

apps/dreamverse/dreamverse/tests/test_session_init_image.py

@@ -0,0 +1,75 @@
+# pyright: reportMissingTypeArgument=false
+import base64
+import io
+from pathlib import Path
+
+from PIL import Image
+import pytest
+
+from dreamverse.session_init_image import (
+    MAX_SESSION_INIT_IMAGE_BYTES,
+    cleanup_session_init_image,
+    persist_session_init_image,
+)
+
+
+def make_data_url(format_name: str = "PNG") -> str:
+    image = Image.new("RGB", (2, 2), color=(16, 32, 64))
+    buffer = io.BytesIO()
+    image.save(buffer, format=format_name)
+    encoded = base64.b64encode(buffer.getvalue()).decode("ascii")
+    mime_type = "image/png" if format_name == "PNG" else "image/jpeg"
+    return f"data:{mime_type};base64,{encoded}"
+
+
+def test_persist_session_init_image_saves_normalized_file(tmp_path: Path):
+    session_image = persist_session_init_image(
+        {
+            "name": "frame.png",
+            "mime_type": "image/png",
+            "data_url": make_data_url("PNG"),
+        },
+        temp_root=tmp_path,
+    )
+
+    assert session_image is not None
+    assert session_image.display_name == "frame.png"
+    assert session_image.file_path.is_file()
+
+    cleanup_session_init_image(session_image)
+    assert not session_image.temp_dir.exists()
+
+
+def test_persist_session_init_image_returns_none_when_missing_data():
+    assert persist_session_init_image(None) is None
+    assert persist_session_init_image({"name": "frame.png", "data_url": ""}) is None
+
+
+def test_persist_session_init_image_rejects_unsupported_mime():
+    with pytest.raises(ValueError, match="PNG, JPEG, or WebP"):
+        persist_session_init_image(
+            {
+                "name": "frame.gif",
+                "mime_type": "image/gif",
+                "data_url": "data:image/gif;base64,R0lGODlhAQABAAAAACw=",
+            }
+        )
+
+
+def test_persist_session_init_image_rejects_large_payload(monkeypatch):
+    data_url = make_data_url("PNG")
+    oversized = "a" * (MAX_SESSION_INIT_IMAGE_BYTES + 1)
+
+    def fake_b64decode(value: str, validate: bool = True):
+        return oversized.encode("ascii")
+
+    monkeypatch.setattr(base64, "b64decode", fake_b64decode)
+
+    with pytest.raises(ValueError, match="15 MB or smaller"):
+        persist_session_init_image(
+            {
+                "name": "frame.png",
+                "mime_type": "image/png",
+                "data_url": data_url,
+            }
+        )