"""Game Master (orchestrator) — three-phase iteration engine. Phase 1: Planner + Executor (tool-calling loop until submit_plan) Phase 2: Writer (single LLM call with submit_step tool) Phase 3: Persist + Deferred triggers + Summary + Suggest actions """ from __future__ import annotations import json import uuid from datetime import datetime, timezone from typing import Any from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from app.core.llm import LlmClient, MockLlmClient from app.core.logging import get_logger from app.core.rag import rag_add from app.core.settings_service import get_all_settings from app.core.time_utils import advance_time, summarize_schemas from app.engine.context import ( build_orchestrator_phase1_context, build_orchestrator_phase2_context, build_orchestrator_phase3_suggest_context, build_summary_context, ) from app.engine.sse import SseEmitter from app.engine.tools.base import ToolContext, get_registry from app.engine.world_builder import _run_tool_loop from app.models import DeferredTrigger, Step, StoryEntry, World _logger = get_logger(__name__) async def run_iteration( *, db: AsyncSession, world: World, step: Step, llm: LlmClient | MockLlmClient, sse: SseEmitter, ) -> None: """Run the full three-phase orchestrator iteration for a single step.""" settings = await get_all_settings(db) try: # ============ Phase 1 ============ await sse.emit("phase_start", {"phase": 1, "name": "planner_executor"}) messages = await build_orchestrator_phase1_context( db=db, world=world, player_action=step.player_action, settings=settings, ) phase1_result = await _run_tool_loop( db=db, world=world, llm=llm, sse=sse, stage="orchestrator_phase1", system_prompt=messages[0]["content"], terminal_tool="submit_plan", max_substeps=int(settings.get("game.max_substeps_per_iteration", 8)), settings=settings, ) await sse.emit("phase_end", {"phase": 1, "duration_ms": 0}) if not phase1_result or not phase1_result.get("ok"): # Force-completion: synthesize a minimal plan phase1_result = { "ok": True, "data": { "plan": "The action was processed but no explicit plan was submitted.", "summary": [], "offscreen_events": [], }, } plan = phase1_result["data"].get("plan", "") summary = phase1_result["data"].get("summary", []) offscreen_events = phase1_result["data"].get("offscreen_events", []) # Persist tool_calls_summary on the step step.tool_calls_summary = summary await db.commit() # ============ Phase 2: Writer ============ await sse.emit("phase_start", {"phase": 2, "name": "writer"}) messages = await build_orchestrator_phase2_context( db=db, world=world, player_action=step.player_action, plan=plan, summary=summary, settings=settings, ) registry = get_registry() ctx = ToolContext(db=db, world=world, step_id=step.id, stage="orchestrator_phase2", sse_emitter=sse.emit) tools = registry.to_openai_format("orchestrator_phase2") phase2_msg: dict[str, Any] = {} phase2_retries = int(settings.get("llm.tool_retry_attempts", 3)) for retry in range(phase2_retries + 1): resp = await llm.complete( stage="orchestrator_phase2", messages=messages, tools=tools, temperature=float(settings.get("llm.temperature_writer", 0.85)), max_tokens=int(settings.get("llm.max_tokens", 2048)), world_id=world.id, step_id=step.id, session=db, ) phase2_msg = resp.get("message", {}) tcs = phase2_msg.get("tool_calls") or [] if tcs: # Execute submit_step for tc in tcs: fn = tc.get("function", {}) if fn.get("name") == "submit_step": try: args = json.loads(fn.get("arguments") or "{}") except json.JSONDecodeError: args = {} result = await registry.execute("submit_step", args, ctx) if result.ok: scene_text = result.data.get("scene_text", "") delta_time = result.data.get("delta_time", "hours_1") # Apply text replacements from app.core.settings_service import apply_text_replacements scene_text = await apply_text_replacements(db, scene_text) step.scene_text = scene_text step.scene_delta_time = delta_time await sse.emit("scene_complete", { "text": scene_text, "delta_time": delta_time, }) break if step.scene_text: break # Retry messages.append(phase2_msg) messages.append({ "role": "user", "content": "You MUST call submit_step with scene_text and delta_time.", }) else: await sse.error("writer_no_submit", "Writer failed to call submit_step after 3 retries") step.status = "failed" await db.commit() return await sse.emit("phase_end", {"phase": 2, "duration_ms": 0}) # ============ Phase 3 ============ await sse.emit("phase_start", {"phase": 3, "name": "persist_triggers_summary_suggest"}) # 3.0 Persist step.status = "completed" world.last_played_at = datetime.now(timezone.utc) world.current_time = advance_time( world.current_time, step.scene_delta_time or "hours_1", world.time_schema ) await db.commit() # 3.1 Deferred triggers if settings.get("game.deferred_triggers_enabled", True): await _process_deferred_triggers( db=db, world=world, step=step, llm=llm, sse=sse, settings=settings, ) # 3.2 Summary (if history is too long) await _maybe_generate_summary( db=db, world=world, step=step, llm=llm, sse=sse, settings=settings, ) # 3.3 Suggest actions — retry up to tool_retry_attempts if no tools returned suggest_msgs = await build_orchestrator_phase3_suggest_context( db=db, world=world, scene_text=step.scene_text or "", settings=settings, ) suggest_tools = registry.to_openai_format("orchestrator_phase3_suggest") tool_retry_limit = int(settings.get("llm.tool_retry_attempts", 3)) for retry in range(tool_retry_limit + 1): resp = await llm.complete( stage="orchestrator_phase3_suggest", messages=suggest_msgs, tools=suggest_tools, temperature=0.8, max_tokens=512, world_id=world.id, step_id=step.id, session=db, ) msg = resp.get("message", {}) # Apply text replacements content = msg.get("content", "") or "" if content: from app.core.settings_service import apply_text_replacements content = await apply_text_replacements(db, content) msg = dict(msg) msg["content"] = content tcs = msg.get("tool_calls") or [] found = False for tc in tcs: fn = tc.get("function", {}) if isinstance(tc, dict) else {} if not isinstance(fn, dict): fn = {"name": tc.get("name", ""), "arguments": tc.get("arguments", "{}")} if fn.get("name") == "suggest_actions": try: args = json.loads(fn.get("arguments") or "{}") except json.JSONDecodeError: args = {} result = await registry.execute("suggest_actions", args, ctx) if result.ok: step.suggested_actions = result.data.get("actions", []) await sse.emit("suggested_actions", {"actions": step.suggested_actions}) found = True break if found: break # No suggest_actions call — retry with nudge suggest_msgs.append(msg) suggest_msgs.append({ "role": "user", "content": ( "You did not call suggest_actions. " "You MUST call the suggest_actions tool with 1-3 short action strings. " "If you tried before and it didn't work, try again with proper JSON arguments." ), }) # If still no actions after retries, provide defaults if not step.suggested_actions: step.suggested_actions = ["Continue exploring", "Talk to someone nearby"] await sse.emit("suggested_actions", {"actions": step.suggested_actions}) await db.commit() await sse.emit("iteration_complete", { "step_id": str(step.id), "sequence_number": step.sequence_number, }) await sse.done({"step_id": str(step.id), "status": "completed"}) except Exception as e: # noqa: BLE001 _logger.exception("orchestrator_failed", step_id=str(step.id), error=str(e)) step.status = "failed" await db.commit() await sse.error("internal_error", str(e)) async def _process_deferred_triggers( *, db: AsyncSession, world: World, step: Step, llm: LlmClient | MockLlmClient, sse: SseEmitter, settings: dict[str, Any], ) -> None: """Fire all deferred triggers whose fire_at <= current_time.""" from app.core.time_utils import time_le triggers = ( await db.execute( select(DeferredTrigger).where( DeferredTrigger.world_id == world.id, DeferredTrigger.is_fired.is_(False), ) ) ).scalars().all() fired = 0 for trig in triggers: try: if not time_le(trig.fire_at, world.current_time): continue except Exception: # noqa: BLE001 continue # Simple firing: append a note to scene_text summary = f"\n\n[Offscreen event: {trig.event_type} — payload: {json.dumps(trig.payload, ensure_ascii=False)}]" if step.scene_text: step.scene_text += summary else: step.scene_text = summary trig.is_fired = True trig.fired_at = datetime.now(timezone.utc) await db.flush() await sse.emit("trigger_fired", { "trigger_id": str(trig.id), "event_type": trig.event_type, "summary": summary.strip(), }) fired += 1 # Persist the trigger event as a story entry await rag_add( db=db, world_id=world.id, content=f"Deferred trigger fired: {trig.event_type} at {trig.fire_at}", entry_type="event", metadata={"trigger_id": str(trig.id), "step_id": str(step.id)}, step_id=step.id, ) if fired: await db.commit() async def _maybe_generate_summary( *, db: AsyncSession, world: World, step: Step, llm: LlmClient | MockLlmClient, sse: SseEmitter, settings: dict[str, Any], ) -> None: """Generate a summary if recent step count exceeds the threshold.""" threshold = int(settings.get("context.compression_threshold_messages", 20)) guaranteed = int(settings.get("context.guaranteed_messages", 10)) recent_steps = list( reversed( ( await db.execute( select(Step) .where(Step.world_id == world.id, Step.deleted_at.is_(None)) .order_by(Step.sequence_number.desc()) .limit(threshold + 1) ) ).scalars().all() ) ) if len(recent_steps) <= threshold: return old_steps = recent_steps[:-guaranteed] if not old_steps: return messages = await build_summary_context( db=db, world=world, old_steps=old_steps, settings=settings, ) resp = await llm.complete( stage="orchestrator_phase3_summary", messages=messages, temperature=0.3, max_tokens=1024, world_id=world.id, step_id=step.id, session=db, ) summary_text = resp.get("message", {}).get("content", "") if not summary_text: return # Store as a story entry se = StoryEntry( world_id=world.id, content=summary_text, entry_type="event", metadata_={ "type": "summary", "step_range": [old_steps[0].sequence_number, old_steps[-1].sequence_number], }, embedding_status="pending", ) db.add(se) await db.commit() await sse.emit("summary_generated", { "summary_id": str(se.id), "message_range": [old_steps[0].sequence_number, old_steps[-1].sequence_number], })