Production lane: qualify (size + propose) → confirm → build (#508)

Derive shot count from the brief's stated duration (1 minute → 12 shots, not a fixed 3); propose the full plan (models · length · est. time) and only build on 'go', adjustable in chat. Service + CLI auto-size; poll timeout scales with shots. +3 tests (80).

services/studio/build_studio_pipe.py

@@ -189,8 +189,8 @@ class Pipe:
         hide_unavailable_lanes: bool = Field(default=True, description="Only list lanes whose backend is currently reachable — ComfyUI (:8188) for every media lane, the voice service (:8193) for the voice lane. When the ai-studio scene is down, the Studio lanes drop out of the OWUI model picker instead of listing-but-erroring. Set false to always list all lanes regardless of backend state.")
         # ── Production lane (the Director: brief → planned → finished film) ──────
         production_url: str = Field(default="http://host.docker.internal:8195", description="Studio Production service (the host-side wrapper around the 4B director + executor). The 🎬 Production lane is a thin client over it; gated by /produce/health. Bring it up on the host: python3 -m services.studio.production.server")
-        production_timeout_s: int = Field(default=1800, description="Max wait for a production to finish (a full film = keyframes + i2v shots + narration + music + assembly; 15–25 min is normal). The lane polls progress until done / error / this timeout.")
-        production_shots: int = Field(default=3, description="Target shot count the director plans the brief into.")
+        production_timeout_s: int = Field(default=1800, description="Min wait for a production to finish; the lane auto-extends this by the shot count (~3.5 min/shot) so a long film doesn't time out. Polls progress until done / error / this budget.")
+        production_shots: int = Field(default=0, description="Shot count. 0 (default) = SIZE it from the brief's stated duration (~5s/shot, e.g. “1 minute” → ~12 shots), else 4 when no length is given. Set >0 to force an exact count.")
         # Two-level UX: leave these 'auto' for the visible default stack (Wan · Chroma ·
         # storyboard), or PIN one to override. The director plans shot content within the
         # chosen stack — it never silently picks the video/image model.
@@ -772,43 +772,127 @@ class Pipe:
         # lane's ⚙️ valves — Auto by default, overridable — and never silently picked by the director.
         if "production" in model:
             loop = asyncio.get_event_loop()
-            brief = ""
-            for m in reversed(body.get("messages", [])):
+            # QUALIFY → CONFIRM → BUILD. The conversation is the state: gather the user turns,
+            # resolve a plan, PROPOSE it (models · length→shots · est. time), and only build once
+            # the user says "go". Never fire a render straight off a brief (or a greeting).
+            users = []
+            for m in body.get("messages", []):
                 if m.get("role") == "user":
                     c = m.get("content")
-                    brief = (" ".join(p.get("text", "") for p in c if isinstance(p, dict) and p.get("type") == "text").strip()
-                             if isinstance(c, list) else (c or "").strip())
-                    break
+                    t = (" ".join(p.get("text", "") for p in c if isinstance(p, dict) and p.get("type") == "text").strip()
+                         if isinstance(c, list) else (c or "").strip())
+                    if t:
+                        users.append(t)
+            last = users[-1] if users else ""
             _help = ("\U0001F3AC Tell me what film to make — a one-line brief like "
-                     "“a 45s calm documentary about lighthouses” or “a 15s noir detective short”. "
-                     "I'll plan it and render the whole thing (keyframes → video → narration → music "
-                     "→ assembly into one MP4); pick the stack in the lane's ⚙️ valves "
-                     "(video/keyframe model, continuity, music) or leave it on Auto.")
+                     "“a 1-minute documentary on the history of Pakistan” or “a 15s noir detective short”. "
+                     "I'll size it, show you the plan (models · length · render time), and build it once "
+                     "you say **go**.")
+            if not last:
+                return _help
+
+            _GREET = ("hi", "hello", "hey", "yo", "sup", "hiya", "hello there", "hola", "thanks",
+                      "thank you", "cool", "nice", "test", "testing", "ping", "help", "?")
+            _CONFIRM = ("go", "yes", "y", "start", "render", "render it", "proceed", "do it", "ok",
+                        "okay", "ok go", "yep", "yeah", "sure", "build", "build it", "make it",
+                        "let's go", "go ahead", "confirm", "\U0001F44D")
+            def _norm(t):
+                return t.strip().lower().strip(" .!?")
+            def _is_confirm(t):
+                return _norm(t) in _CONFIRM
+            def _is_greeting(t):
+                return _norm(t) in _GREET
+            def _overrides(t):
+                tl = t.lower(); words = set(re.findall(r"[a-z0-9\-]+", tl)); o = {}
+                for k in ("10eros", "sulphur", "ltx", "wan"):
+                    if k in words:
+                        o["video_lane"] = k; break
+                for k, v in (("hidream", "hidream"), ("z-image", "zimage"), ("zimage", "zimage"),
+                             ("chroma", "chroma"), ("krea", "krea")):
+                    if k in words:
+                        o["keyframe_lane"] = v; break
+                for cc in ("storyboard", "hero", "chain"):
+                    if cc in words:
+                        o["continuity"] = cc
+                if "no continuity" in tl or "independent" in words:
+                    o["continuity"] = "none"
+                if "no music" in tl or "without music" in tl:
+                    o["music"] = False
+                if "no narration" in tl or "no voice" in tl or "no voiceover" in tl:
+                    o["narration"] = False
+                s = self._target_seconds(t)
+                if s:
+                    o["seconds"] = s
+                return o
+            def _pure_override(t):
+                return bool(_overrides(t)) and len(t.split()) <= 5
+
+            # effective brief = the first real turn (not a greeting / confirmation / short tweak)
+            brief = next((t for t in users
+                          if not _is_confirm(t) and not _is_greeting(t) and not _pure_override(t)), "")
+            ov = {}
+            for t in users:
+                ov.update(_overrides(t))         # accumulate tweaks across the convo (later wins)
+            video = ov.get("video_lane") or (self.valves.production_video_lane or "auto")
+            keyf = ov.get("keyframe_lane") or (self.valves.production_keyframe_lane or "auto")
+            cont = ov.get("continuity") or (self.valves.production_continuity or "auto")
+            music = ov.get("music", bool(self.valves.production_music))
+            narr = ov.get("narration", True)
+            secs = ov.get("seconds") or 0
+            if secs:
+                shots = max(1, min(24, round(secs / 5.0)))      # ~5s per Wan shot, capped at ~2 min
+            elif int(self.valves.production_shots or 0) > 0:
+                shots = int(self.valves.production_shots)
+            else:
+                shots = 4                                       # no stated length → a short default
+            est_lo, est_hi = int(round(shots * 2.5)), int(round(shots * 3)) + 3
+
+            # no real brief yet → chit-chat (greeting / vague), never a render
             if not brief:
+                if _is_confirm(last):
+                    return ("Nothing to start yet — give me a one-line brief first, e.g. "
+                            "“a 1-minute documentary on the history of Pakistan”.")
+                if not _is_greeting(last):
+                    try:
+                        _c = self._chat_gate(await loop.run_in_executor(None, self._enhance, last, False, None, "video"))
+                        if _c is not None:
+                            await status("", True)
+                            return "\U0001F3AC " + _c + "\n\n_(Give me a one-line film brief and I'll plan it.)_"
+                    except Exception:
+                        pass
                 return _help
-            # CHIT-CHAT GATE — don't spin up a full production for a greeting / vague message
-            # (otherwise the director invents + renders a random film from "hello"). Cheap
-            # exact-match fast path, then the director's CHAT: gate (same as every other lane).
-            if brief.strip().lower().strip(" .!?") in (
-                    "hi", "hello", "hey", "yo", "sup", "hiya", "hello there", "hola", "thanks",
-                    "thank you", "ok", "okay", "cool", "nice", "test", "testing", "ping", "help", "?"):
+
+            # have a brief but NOT confirming → PROPOSE the plan (qualify); no render
+            if not _is_confirm(last):
+                _vl = {"auto": "Wan2.2 (auto)", "wan": "Wan2.2", "ltx": "LTX-2.3",
+                       "sulphur": "Sulphur", "10eros": "10Eros"}.get(video, video)
+                if video not in ("auto", "wan"):
+                    _vl += " ⚠️ roadmap — only Wan renders today"
+                _kl = {"auto": "Chroma (auto)", "chroma": "Chroma", "zimage": "Z-Image",
+                       "krea": "Krea 2", "hidream": "HiDream-O1"}.get(keyf, keyf)
+                _audio = ("narration" if narr else "no narration") + " + " + ("music" if music else "no music")
+                _len = ("~%ds → %d shots" % (int(secs), shots)) if secs else \
+                       ("%d shots (~%ds — say a length like “1 minute” to size it)" % (shots, shots * 5))
                 await status("", True)
-                return _help
-            try:
-                _crafted = await loop.run_in_executor(None, self._enhance, brief, False, None, "video")
-                _chat = self._chat_gate(_crafted)
-                if _chat is not None:           # director judged it chit-chat / too vague → no render
-                    await status("", True)
-                    return ("\U0001F3AC " + _chat +
-                            "\n\n_(Give me a one-line film brief and I'll plan + render it.)_")
-            except Exception:
-                pass    # director unreachable → fall through; the production service surfaces the error
+                return (
+                    "\U0001F3AC **Plan — " + brief + "**\n\n"
+                    "| | |\n|---|---|\n"
+                    "| \U0001F3A5 video | **" + _vl + "** |\n"
+                    "| \U0001F5BC️ keyframes | **" + _kl + "** |\n"
+                    "| \U0001F39E️ continuity | **" + str(cont) + "**  ·  \U0001F50A audio **" + _audio + "** |\n"
+                    "| ⏱️ length | **" + _len + "** |\n"
+                    "| ⚙️ est. render | **~" + str(est_lo) + "–" + str(est_hi) + " min** on 1× 3090 |\n\n"
+                    "Reply **go** to start — or tell me what to change: _“use LTX” · “30 seconds” · "
+                    "“no music” · “hidream keyframes” · “hero continuity”_.\n\n"
+                    "_(Video renders on **Wan** today; LTX/Sulphur/10Eros are roadmap.)_"
+                )
+
+            # CONFIRMED + have a brief → build with the resolved plan
+            await status("\U0001F3AC Starting “" + brief + "” — " + str(shots) + " shots, ~" +
+                         str(est_lo) + "–" + str(est_hi) + " min…")
             base_prod = self.valves.production_url.rstrip("/")
-            payload = {"brief": brief, "shots": int(self.valves.production_shots),
-                       "video_lane": self.valves.production_video_lane,
-                       "keyframe_lane": self.valves.production_keyframe_lane,
-                       "continuity": self.valves.production_continuity,
-                       "music": bool(self.valves.production_music)}
+            payload = {"brief": brief, "shots": shots, "video_lane": video, "keyframe_lane": keyf,
+                       "continuity": cont, "music": music, "narration": narr}
 
             def _prod_post():
                 req = urllib.request.Request(base_prod + "/produce", data=json.dumps(payload).encode(),
@@ -842,7 +926,7 @@ class Pipe:
             stack_line = ("**Stack** — video: `" + str(st.get("video_lane")) + "` · keyframes: `" +
                           str(st.get("keyframe_lane")) + "` · continuity: `" + str(st.get("continuity")) +
                           "` · " + ("music on" if st.get("music") else "no music"))
-            deadline = time.time() + int(self.valves.production_timeout_s)
+            deadline = time.time() + max(int(self.valves.production_timeout_s), shots * 210 + 300)
             last_phase = ""
             job = {}
             while time.time() < deadline:

services/studio/production/planner.py

@@ -13,7 +13,9 @@ model required (mirrors the synthetic lane backend).
 from __future__ import annotations
 
 import json
+import math
 import os
+import re
 import urllib.request
 
 from . import config
@@ -27,6 +29,37 @@ class PlannerError(RuntimeError):
     pass
 
 
+# -- sizing: derive the shot count from the brief's requested duration --------
+SECONDS_PER_SHOT = 5.0   # Wan native window ≈ 81 frames @ 16 fps ≈ 5 s
+MAX_SHOTS = 24           # cap auto-derivation (~120 s) so a stray "10 minute" can't queue hours
+DEFAULT_SHOTS = 4        # when the brief states no duration
+
+
+def parse_duration_seconds(text: str):
+    """Best-effort 'how long' from a brief: '1 minute' -> 60.0, '45s' -> 45.0. None if absent."""
+    t = text or ""
+    m = re.search(r"(\d+(?:\.\d+)?)\s*(?:minutes?|mins?|m)\b", t, re.I)
+    if m:
+        return float(m.group(1)) * 60.0
+    m = re.search(r"(\d+(?:\.\d+)?)\s*(?:seconds?|secs?|s)\b", t, re.I)
+    if m:
+        return float(m.group(1))
+    return None
+
+
+def derive_shots(brief: str, *, default: int = DEFAULT_SHOTS):
+    """Map a brief's requested duration to a shot count (~5 s/shot, capped at MAX_SHOTS).
+
+    The agent SIZES the film from the request instead of a fixed test count —
+    'a 1 minute video' -> ~12 shots. Returns (shots, requested_seconds | None).
+    """
+    secs = parse_duration_seconds(brief)
+    if not secs or secs <= 0:
+        return default, None
+    shots = max(1, min(MAX_SHOTS, math.ceil(secs / SECONDS_PER_SHOT)))
+    return shots, secs
+
+
 # -- director call (the injectable boundary) ----------------------------------
 def director_call(messages: list[dict], *, max_tokens: int, temperature: float,
                   base: str | None = None, timeout: int = 120) -> str:

services/studio/production/run.py

@@ -64,7 +64,9 @@ def main(argv: list[str] | None = None) -> int:
     ap.add_argument("plan", nargs="?", help="path to a ProductionPlanV1 JSON (v0a path)")
     ap.add_argument("--brief", default=None,
                     help='a one-line brief; the 4B director plans it (v0b), e.g. --brief "60s doc on lighthouses"')
-    ap.add_argument("--shots", type=int, default=3, help="target shot count for --brief planning")
+    ap.add_argument("--shots", type=int, default=0,
+                    help="shot count for --brief planning. 0 (default) = SIZE it from the brief's "
+                         "stated duration (~5s/shot, e.g. \"1 minute\" → ~12 shots); >0 = explicit.")
     # -- the production stack (operator-chosen; 'auto' = the visible default) --
     ap.add_argument("--video-lane", default="auto",
                     help="video model for every shot: wan (default, renders today) · "
@@ -119,14 +121,21 @@ def main(argv: list[str] | None = None) -> int:
             # brief path: the CLI stack pins the lanes; the director plans within it.
             print(describe_stack(stack), file=sys.stderr)
             from . import planner, registry
+            # Size the film from the brief's stated duration unless --shots was given.
+            if args.shots > 0:
+                shots = args.shots
+            else:
+                shots, _secs = planner.derive_shots(args.brief)
+                if _secs:
+                    print(f"[plan] brief asks for ~{_secs:.0f}s → ~{shots} shots", file=sys.stderr)
             job_id = args.job_id or _job_id(args.brief)
             prod_dir = os.path.join(args.productions_dir, job_id)
             os.makedirs(prod_dir, exist_ok=True)
             try:
                 reg = registry.load()
-                print(f"[plan] director planning the brief into ~{args.shots} shots…", file=sys.stderr)
+                print(f"[plan] director planning the brief into ~{shots} shots…", file=sys.stderr)
                 plan, extra_artifacts = planner.plan_from_brief(
-                    args.brief, reg, n_shots=args.shots, stack=stack,
+                    args.brief, reg, n_shots=shots, stack=stack,
                     prompts_dir=os.path.join(prod_dir, "prompts"),
                 )
             except planner.PlannerError as e:

services/studio/production/server.py

@@ -153,17 +153,29 @@ class Handler(BaseHTTPRequestHandler):
         except StackError as e:
             return self._json(400, {"error": str(e), "renders_today": {
                 "video": wired_video_lanes(), "keyframe": wired_keyframe_lanes()}})
+        # Shot count: an explicit shots>0 wins; otherwise SIZE it from the brief's stated
+        # duration ("1 minute" -> ~12 shots) instead of a fixed test count (#production).
+        try:
+            req_shots = int(b.get("shots", 0) or 0)
+        except (TypeError, ValueError):
+            req_shots = 0
+        if req_shots > 0:
+            shots, req_secs = req_shots, None
+        else:
+            shots, req_secs = planner.derive_shots(brief)
         if _active():
             return self._json(409, {"error": "a production is already running", "job_id": _active()[0]})
         job_id = _job_id(brief)
         _set(job_id, status="planning", phase="queued", frac=0.0, brief=brief, title="…",
-             stack=stack.to_dict(), stack_desc=describe_stack(stack))
+             stack=stack.to_dict(), stack_desc=describe_stack(stack),
+             shots=shots, requested_seconds=req_secs)
         threading.Thread(
             target=_run_job,
-            args=(job_id, brief, stack, int(b.get("shots", 3)), b.get("backend", "live")),
+            args=(job_id, brief, stack, shots, b.get("backend", "live")),
             daemon=True,
         ).start()
-        return self._json(200, {"job_id": job_id, "stack": stack.to_dict()})
+        return self._json(200, {"job_id": job_id, "stack": stack.to_dict(),
+                                "shots": shots, "requested_seconds": req_secs})
 
 
 def main() -> None:

services/studio/production/tests/test_v0b.py

@@ -131,5 +131,25 @@ class TestProvenanceNotCountedAsValidator(unittest.TestCase):
             self.assertIn("media", types)
 
 
+class TestDeriveShots(unittest.TestCase):
+    """Size the film from the brief's stated duration (~5s/shot), not a fixed test count."""
+
+    def test_minutes_and_seconds(self):
+        from ..planner import derive_shots
+        self.assertEqual(derive_shots("a 1 minute video on the history of pakistan")[0], 12)
+        self.assertEqual(derive_shots("a 45 second noir short")[0], 9)
+        self.assertEqual(derive_shots("make a 30s clip")[0], 6)
+
+    def test_caps_runaway_requests(self):
+        from ..planner import derive_shots
+        self.assertEqual(derive_shots("a 10 minute epic")[0], 24)   # capped, not 120 shots
+
+    def test_default_when_no_duration(self):
+        from ..planner import derive_shots
+        shots, secs = derive_shots("a noir detective short")
+        self.assertEqual(shots, 4)
+        self.assertIsNone(secs)
+
+
 if __name__ == "__main__":
     unittest.main()

services/studio/studio_pipe.py

@@ -82,8 +82,8 @@ class Pipe:
         hide_unavailable_lanes: bool = Field(default=True, description="Only list lanes whose backend is currently reachable — ComfyUI (:8188) for every media lane, the voice service (:8193) for the voice lane. When the ai-studio scene is down, the Studio lanes drop out of the OWUI model picker instead of listing-but-erroring. Set false to always list all lanes regardless of backend state.")
         # ── Production lane (the Director: brief → planned → finished film) ──────
         production_url: str = Field(default="http://host.docker.internal:8195", description="Studio Production service (the host-side wrapper around the 4B director + executor). The 🎬 Production lane is a thin client over it; gated by /produce/health. Bring it up on the host: python3 -m services.studio.production.server")
-        production_timeout_s: int = Field(default=1800, description="Max wait for a production to finish (a full film = keyframes + i2v shots + narration + music + assembly; 15–25 min is normal). The lane polls progress until done / error / this timeout.")
-        production_shots: int = Field(default=3, description="Target shot count the director plans the brief into.")
+        production_timeout_s: int = Field(default=1800, description="Min wait for a production to finish; the lane auto-extends this by the shot count (~3.5 min/shot) so a long film doesn't time out. Polls progress until done / error / this budget.")
+        production_shots: int = Field(default=0, description="Shot count. 0 (default) = SIZE it from the brief's stated duration (~5s/shot, e.g. “1 minute” → ~12 shots), else 4 when no length is given. Set >0 to force an exact count.")
         # Two-level UX: leave these 'auto' for the visible default stack (Wan · Chroma ·
         # storyboard), or PIN one to override. The director plans shot content within the
         # chosen stack — it never silently picks the video/image model.
@@ -665,43 +665,127 @@ class Pipe:
         # lane's ⚙️ valves — Auto by default, overridable — and never silently picked by the director.
         if "production" in model:
             loop = asyncio.get_event_loop()
-            brief = ""
-            for m in reversed(body.get("messages", [])):
+            # QUALIFY → CONFIRM → BUILD. The conversation is the state: gather the user turns,
+            # resolve a plan, PROPOSE it (models · length→shots · est. time), and only build once
+            # the user says "go". Never fire a render straight off a brief (or a greeting).
+            users = []
+            for m in body.get("messages", []):
                 if m.get("role") == "user":
                     c = m.get("content")
-                    brief = (" ".join(p.get("text", "") for p in c if isinstance(p, dict) and p.get("type") == "text").strip()
-                             if isinstance(c, list) else (c or "").strip())
-                    break
+                    t = (" ".join(p.get("text", "") for p in c if isinstance(p, dict) and p.get("type") == "text").strip()
+                         if isinstance(c, list) else (c or "").strip())
+                    if t:
+                        users.append(t)
+            last = users[-1] if users else ""
             _help = ("\U0001F3AC Tell me what film to make — a one-line brief like "
-                     "“a 45s calm documentary about lighthouses” or “a 15s noir detective short”. "
-                     "I'll plan it and render the whole thing (keyframes → video → narration → music "
-                     "→ assembly into one MP4); pick the stack in the lane's ⚙️ valves "
-                     "(video/keyframe model, continuity, music) or leave it on Auto.")
+                     "“a 1-minute documentary on the history of Pakistan” or “a 15s noir detective short”. "
+                     "I'll size it, show you the plan (models · length · render time), and build it once "
+                     "you say **go**.")
+            if not last:
+                return _help
+
+            _GREET = ("hi", "hello", "hey", "yo", "sup", "hiya", "hello there", "hola", "thanks",
+                      "thank you", "cool", "nice", "test", "testing", "ping", "help", "?")
+            _CONFIRM = ("go", "yes", "y", "start", "render", "render it", "proceed", "do it", "ok",
+                        "okay", "ok go", "yep", "yeah", "sure", "build", "build it", "make it",
+                        "let's go", "go ahead", "confirm", "\U0001F44D")
+            def _norm(t):
+                return t.strip().lower().strip(" .!?")
+            def _is_confirm(t):
+                return _norm(t) in _CONFIRM
+            def _is_greeting(t):
+                return _norm(t) in _GREET
+            def _overrides(t):
+                tl = t.lower(); words = set(re.findall(r"[a-z0-9\-]+", tl)); o = {}
+                for k in ("10eros", "sulphur", "ltx", "wan"):
+                    if k in words:
+                        o["video_lane"] = k; break
+                for k, v in (("hidream", "hidream"), ("z-image", "zimage"), ("zimage", "zimage"),
+                             ("chroma", "chroma"), ("krea", "krea")):
+                    if k in words:
+                        o["keyframe_lane"] = v; break
+                for cc in ("storyboard", "hero", "chain"):
+                    if cc in words:
+                        o["continuity"] = cc
+                if "no continuity" in tl or "independent" in words:
+                    o["continuity"] = "none"
+                if "no music" in tl or "without music" in tl:
+                    o["music"] = False
+                if "no narration" in tl or "no voice" in tl or "no voiceover" in tl:
+                    o["narration"] = False
+                s = self._target_seconds(t)
+                if s:
+                    o["seconds"] = s
+                return o
+            def _pure_override(t):
+                return bool(_overrides(t)) and len(t.split()) <= 5
+
+            # effective brief = the first real turn (not a greeting / confirmation / short tweak)
+            brief = next((t for t in users
+                          if not _is_confirm(t) and not _is_greeting(t) and not _pure_override(t)), "")
+            ov = {}
+            for t in users:
+                ov.update(_overrides(t))         # accumulate tweaks across the convo (later wins)
+            video = ov.get("video_lane") or (self.valves.production_video_lane or "auto")
+            keyf = ov.get("keyframe_lane") or (self.valves.production_keyframe_lane or "auto")
+            cont = ov.get("continuity") or (self.valves.production_continuity or "auto")
+            music = ov.get("music", bool(self.valves.production_music))
+            narr = ov.get("narration", True)
+            secs = ov.get("seconds") or 0
+            if secs:
+                shots = max(1, min(24, round(secs / 5.0)))      # ~5s per Wan shot, capped at ~2 min
+            elif int(self.valves.production_shots or 0) > 0:
+                shots = int(self.valves.production_shots)
+            else:
+                shots = 4                                       # no stated length → a short default
+            est_lo, est_hi = int(round(shots * 2.5)), int(round(shots * 3)) + 3
+
+            # no real brief yet → chit-chat (greeting / vague), never a render
             if not brief:
+                if _is_confirm(last):
+                    return ("Nothing to start yet — give me a one-line brief first, e.g. "
+                            "“a 1-minute documentary on the history of Pakistan”.")
+                if not _is_greeting(last):
+                    try:
+                        _c = self._chat_gate(await loop.run_in_executor(None, self._enhance, last, False, None, "video"))
+                        if _c is not None:
+                            await status("", True)
+                            return "\U0001F3AC " + _c + "\n\n_(Give me a one-line film brief and I'll plan it.)_"
+                    except Exception:
+                        pass
                 return _help
-            # CHIT-CHAT GATE — don't spin up a full production for a greeting / vague message
-            # (otherwise the director invents + renders a random film from "hello"). Cheap
-            # exact-match fast path, then the director's CHAT: gate (same as every other lane).
-            if brief.strip().lower().strip(" .!?") in (
-                    "hi", "hello", "hey", "yo", "sup", "hiya", "hello there", "hola", "thanks",
-                    "thank you", "ok", "okay", "cool", "nice", "test", "testing", "ping", "help", "?"):
+
+            # have a brief but NOT confirming → PROPOSE the plan (qualify); no render
+            if not _is_confirm(last):
+                _vl = {"auto": "Wan2.2 (auto)", "wan": "Wan2.2", "ltx": "LTX-2.3",
+                       "sulphur": "Sulphur", "10eros": "10Eros"}.get(video, video)
+                if video not in ("auto", "wan"):
+                    _vl += " ⚠️ roadmap — only Wan renders today"
+                _kl = {"auto": "Chroma (auto)", "chroma": "Chroma", "zimage": "Z-Image",
+                       "krea": "Krea 2", "hidream": "HiDream-O1"}.get(keyf, keyf)
+                _audio = ("narration" if narr else "no narration") + " + " + ("music" if music else "no music")
+                _len = ("~%ds → %d shots" % (int(secs), shots)) if secs else \
+                       ("%d shots (~%ds — say a length like “1 minute” to size it)" % (shots, shots * 5))
                 await status("", True)
-                return _help
-            try:
-                _crafted = await loop.run_in_executor(None, self._enhance, brief, False, None, "video")
-                _chat = self._chat_gate(_crafted)
-                if _chat is not None:           # director judged it chit-chat / too vague → no render
-                    await status("", True)
-                    return ("\U0001F3AC " + _chat +
-                            "\n\n_(Give me a one-line film brief and I'll plan + render it.)_")
-            except Exception:
-                pass    # director unreachable → fall through; the production service surfaces the error
+                return (
+                    "\U0001F3AC **Plan — " + brief + "**\n\n"
+                    "| | |\n|---|---|\n"
+                    "| \U0001F3A5 video | **" + _vl + "** |\n"
+                    "| \U0001F5BC️ keyframes | **" + _kl + "** |\n"
+                    "| \U0001F39E️ continuity | **" + str(cont) + "**  ·  \U0001F50A audio **" + _audio + "** |\n"
+                    "| ⏱️ length | **" + _len + "** |\n"
+                    "| ⚙️ est. render | **~" + str(est_lo) + "–" + str(est_hi) + " min** on 1× 3090 |\n\n"
+                    "Reply **go** to start — or tell me what to change: _“use LTX” · “30 seconds” · "
+                    "“no music” · “hidream keyframes” · “hero continuity”_.\n\n"
+                    "_(Video renders on **Wan** today; LTX/Sulphur/10Eros are roadmap.)_"
+                )
+
+            # CONFIRMED + have a brief → build with the resolved plan
+            await status("\U0001F3AC Starting “" + brief + "” — " + str(shots) + " shots, ~" +
+                         str(est_lo) + "–" + str(est_hi) + " min…")
             base_prod = self.valves.production_url.rstrip("/")
-            payload = {"brief": brief, "shots": int(self.valves.production_shots),
-                       "video_lane": self.valves.production_video_lane,
-                       "keyframe_lane": self.valves.production_keyframe_lane,
-                       "continuity": self.valves.production_continuity,
-                       "music": bool(self.valves.production_music)}
+            payload = {"brief": brief, "shots": shots, "video_lane": video, "keyframe_lane": keyf,
+                       "continuity": cont, "music": music, "narration": narr}
 
             def _prod_post():
                 req = urllib.request.Request(base_prod + "/produce", data=json.dumps(payload).encode(),
@@ -735,7 +819,7 @@ class Pipe:
             stack_line = ("**Stack** — video: `" + str(st.get("video_lane")) + "` · keyframes: `" +
                           str(st.get("keyframe_lane")) + "` · continuity: `" + str(st.get("continuity")) +
                           "` · " + ("music on" if st.get("music") else "no music"))
-            deadline = time.time() + int(self.valves.production_timeout_s)
+            deadline = time.time() + max(int(self.valves.production_timeout_s), shots * 210 + 300)
             last_phase = ""
             job = {}
             while time.time() < deadline: