from puppet_theater.actions import request_finale, summon_actor, throw_prop import puppet_theater.director as director_module from puppet_theater.director import ( BEAT_ARC, build_director_prompt, choose_director_decision, parse_director_decision, run_one_beat, run_full_act, story_phase, ) from puppet_theater.models import DirectorDecision, ToolRequest from puppet_theater.backends import build_actor_line_prompt, parse_actor_output from puppet_theater.session import create_show_from_premise, resolve_show_length from puppet_theater.tools import run_actor_tool_request def test_show_length_presets_are_resolved() -> None: assert resolve_show_length("short") == ("short", 5, 7, 8) assert resolve_show_length("standard") == ("standard", 7, 10, 12) assert resolve_show_length("extended") == ("extended", 10, 14, 16) assert resolve_show_length("unknown") == ("standard", 7, 10, 12) def test_director_prompt_includes_audience_context_and_schema() -> None: session = create_show_from_premise( "A moon detective interrogates a suspicious toaster", director_mode="hf_api", ) session = throw_prop(session, "tiny crown") session = summon_actor(session, "Professor Button") prompt = build_director_prompt(session) assert session.show_title in prompt assert "A moon detective interrogates a suspicious toaster" in prompt assert "tiny crown" in prompt assert "Audience summoned Professor Button." in prompt assert "Professor Button" in prompt assert any(actor.secret in prompt for actor in session.actors) assert "Minimum beats before normal ending: 7" in prompt assert "Target beats: 10" in prompt assert "Maximum beats: 12" in prompt assert "Current progress: 0%" in prompt assert "Story phase: opening" in prompt assert '"next_speaker"' in prompt assert "Allowed beat types: " in prompt def test_actor_prompt_formats_literal_tool_argument_examples() -> None: session = create_show_from_premise("A moon detective interrogates a suspicious toaster") decision = DirectorDecision( next_speaker=session.actors[0].name, beat_type="evidence_or_prop", instruction="Inspect the clue in one sharp line.", stage_effect="prop_table_glow", uses_prop=True, reveal_secret=False, should_end_scene=False, reason_summary="Move the clue forward.", ) prompt = build_actor_line_prompt(session, decision, session.actors[0], "tiny crown") assert '{"prop": "short prop name"}' in prompt assert '{"question": "short stage question"}' in prompt assert '{"mood": "short lighting mood"}' in prompt assert "Latest prop: tiny crown" in prompt def test_story_phase_uses_target_progress() -> None: session = create_show_from_premise("A teacup files a weather report", show_length="standard") session.beat_index = 0 assert story_phase(session) == "opening" session.beat_index = 2 assert story_phase(session) == "complication" session.beat_index = 5 assert story_phase(session) == "reveal" session.beat_index = 7 assert story_phase(session) == "chaos" session.beat_index = 9 assert story_phase(session) == "finale" def test_invalid_llm_speaker_is_rejected() -> None: session = create_show_from_premise("A castle argues with a thunderstorm") raw_decision = { "next_speaker": "Imaginary Stand-In", "beat_type": "setup", "instruction": "Open the scene in one line.", "stage_effect": "warm_spotlight", "uses_prop": False, "reveal_secret": False, "should_end_scene": False, "reason_summary": "Start the scene.", } decision, status = parse_director_decision(raw_decision, session) assert decision is None assert status == "invalid_speaker" def test_old_actor_output_defaults_agentic_fields() -> None: response, status = parse_actor_output( { "line": "The moon denies everything with suspicious confidence.", "emotion": "suspicious", "gesture": "point_accusingly", "stage_effect": "spotlight", "tool_request": None, } ) assert response is not None assert response.intent == "Respond to the Director's cue." assert response.memory_update == "" assert "compat_intent_defaulted" in status assert "compat_memory_defaulted" in status def test_actor_output_accepts_structured_tool_request() -> None: response, status = parse_actor_output( { "intent": "Inspect the suspicious prop.", "line": "This crown is practically sweating clues.", "emotion": "suspicious", "gesture": "present_prop", "stage_effect": "prop_table_glow", "memory_update": "Noticed the crown acting suspicious.", "tool_request": { "tool_name": "inspect_prop", "arguments": {"prop": "tiny crown"}, "reason": "The prop should reveal a clue.", }, } ) assert status == "valid" assert response is not None assert response.tool_request is not None assert response.tool_request.tool_name == "inspect_prop" assert response.tool_request.arguments == {"prop": "tiny crown"} def test_invalid_tool_request_is_ignored_with_trace() -> None: session = create_show_from_premise("A moon detective interrogates a suspicious toaster") actor = session.actors[0] result = run_actor_tool_request( session, actor, ToolRequest(tool_name="web_search", arguments={"query": "secrets"}, reason="Look outside the stage."), ) assert result is None event_types = [event["type"] for event in session.trace_events if isinstance(event, dict)] assert "tool_requested" in event_types assert "tool_ignored" in event_types assert session.latest_tool_result is None def test_invalid_tool_arguments_are_ignored_with_trace() -> None: session = create_show_from_premise("A moon detective interrogates a suspicious toaster") actor = session.actors[0] result = run_actor_tool_request( session, actor, ToolRequest(tool_name="inspect_prop", arguments={"prop": {"nested": "unsafe"}}, reason="Inspect it."), ) assert result is None ignored = [event for event in session.trace_events if isinstance(event, dict) and event["type"] == "tool_ignored"] assert ignored assert ignored[-1]["validation_status"] == "invalid_tool_arguments" def test_actor_state_updates_after_beat() -> None: session = create_show_from_premise("A moon detective interrogates a suspicious toaster") actor = session.actors[0] run_one_beat(session) assert session.transcript[-1].intent assert actor.mood == session.transcript[-1].emotion assert actor.current_goal == session.transcript[-1].intent assert actor.recent_memory event_types = {event["type"] for event in session.trace_events if isinstance(event, dict)} assert "actor_intent" in event_types assert "actor_memory_update" in event_types assert "actor_state_update" in event_types def test_prop_tool_runs_and_influences_following_beat() -> None: session = create_show_from_premise("A moon detective interrogates a suspicious toaster") throw_prop(session, "tiny crown") run_one_beat(session) assert session.latest_tool_result is not None assert session.latest_tool_result.tool_name == "inspect_prop" assert "tiny crown" in session.latest_tool_result.result first_event_types = [event["type"] for event in session.trace_events if isinstance(event, dict)] assert "tool_requested" in first_event_types assert "tool_executed" in first_event_types assert "tool_result" in first_event_types run_one_beat(session) assert "last stage clue" in session.transcript[-1].line def test_fallback_used_when_hf_api_director_cannot_run(monkeypatch) -> None: session = create_show_from_premise("A kitchen council debates a heroic toaster", director_mode="hf_api") monkeypatch.setattr(director_module, "get_backend", lambda *args, **kwargs: object()) generation = choose_director_decision(session) assert generation.director_mode == "hf_api" assert generation.fallback_used is True assert generation.decision.next_speaker in {actor.name for actor in session.actors} assert generation.decision.beat_type in BEAT_ARC def test_finale_request_forces_bounded_finale_decision() -> None: session = create_show_from_premise("A library book demands a curtain call") session = request_finale(session) raw_decision = { "next_speaker": session.actors[0].name, "beat_type": "chaos_or_intervention", "instruction": "Try to keep the scene going.", "stage_effect": "confetti_rustle", "uses_prop": False, "reveal_secret": False, "should_end_scene": False, "reason_summary": "The audience asked for momentum.", } decision, status = parse_director_decision(raw_decision, session) assert status == "valid" assert decision is not None assert decision.beat_type == "finale" assert decision.should_end_scene is True assert decision.stage_effect == "curtain_fall" def test_early_should_end_scene_is_bounded_until_target() -> None: session = create_show_from_premise("A teacup runs for mayor") session.beat_index = session.min_beats - 1 decision = DirectorDecision( next_speaker=session.actors[0].name, beat_type="setup", instruction="Set up the campaign.", stage_effect="warm_spotlight", should_end_scene=True, reason_summary="The model tried to end too early.", ) validated, status = parse_director_decision(decision, session) assert status == "valid" assert validated is not None assert validated.should_end_scene is False def test_early_llm_finale_is_rewritten_before_target() -> None: session = create_show_from_premise("A lighthouse hosts a talent show") decision = DirectorDecision( next_speaker=session.actors[0].name, beat_type="finale", instruction="End immediately.", stage_effect="curtain_fall", should_end_scene=True, reason_summary="The model tried to close early.", ) validated, status = parse_director_decision(decision, session) assert status == "valid" assert validated is not None assert validated.beat_type != "finale" assert validated.should_end_scene is False def test_full_act_standard_runs_to_target_and_stays_bounded() -> None: session = create_show_from_premise("A moon detective interrogates a suspicious toaster") run_full_act(session) assert len(session.transcript) == session.target_beats assert len(session.transcript) > 6 assert session.beat_index == session.max_beats assert len(session.transcript) <= session.max_beats assert session.transcript[-1].stage_effect == "curtain_fall" def test_length_presets_change_full_act_bounds() -> None: short_session = create_show_from_premise("A tiny crown demands applause", show_length="short") extended_session = create_show_from_premise("A telescope opens a bakery", show_length="extended") run_full_act(short_session) run_full_act(extended_session) assert (short_session.min_beats, short_session.target_beats, short_session.max_beats) == (5, 7, 8) assert (extended_session.min_beats, extended_session.target_beats, extended_session.max_beats) == (10, 14, 16) assert len(short_session.transcript) == 7 assert len(extended_session.transcript) == 14 assert len(extended_session.transcript) <= extended_session.max_beats def test_finale_request_can_end_before_minimum() -> None: session = create_show_from_premise("A spoon wants a curtain call", show_length="extended") session = request_finale(session) run_full_act(session) assert len(session.transcript) == 1 assert session.transcript[-1].stage_effect == "curtain_fall" def test_llm_decision_is_forced_to_finale_at_max() -> None: session = create_show_from_premise("A clock refuses bedtime", show_length="short") session.beat_index = session.max_beats - 1 decision = DirectorDecision( next_speaker=session.actors[0].name, beat_type="chaos_or_intervention", instruction="Keep escalating.", stage_effect="confetti_rustle", should_end_scene=False, reason_summary="The model tried to continue at the hard bound.", ) validated, status = parse_director_decision(decision, session) assert status == "valid" assert validated is not None assert validated.beat_type == "finale" assert validated.should_end_scene is True assert validated.stage_effect == "curtain_fall"