ai-rpg/backend/app/workers/trigger_runner.py

"""Trigger checker: scans due deferred triggers and fires them.

The actual firing = creating a new narrative step for the player to see,
OR a hidden technical message if the event is "offscreen".
"""
from __future__ import annotations

import asyncio
import json
import re
import uuid
from typing import Any, Dict, List

from sqlalchemy import select

from app.core.llm import LlmClient
from app.core.settings_service import cast_setting, get_all_settings
from app.core.state_validator import apply_patch, validate_state
from app.db import AsyncSessionLocal
from app.logging_setup import get_logger, setup_logging
from app.models import DeferredTrigger, Message, Session, World
from app.prompts.templates import get_prompt

log = get_logger("trigger_runner")


def _parse_time(t: str) -> int:
    m = re.match(r"day_(\d+)_hour_(\d+)", t or "")
    if m:
        return int(m.group(1)) * 24 * 60 + int(m.group(2)) * 60
    try:
        from datetime import datetime
        return int(datetime.fromisoformat(t).timestamp() // 60)
    except Exception:
        return 0


async def check_and_fire_triggers() -> int:
    """Find all unfired triggers whose fire_at <= current world time, fire them.

    Returns the number of triggers fired.
    """
    setup_logging()
    async with AsyncSessionLocal() as db:
        result = await db.execute(
            select(DeferredTrigger, Session, World)
            .join(Session, DeferredTrigger.session_id == Session.id)
            .join(World, Session.world_id == World.id)
            .where(DeferredTrigger.fired.is_(False))
        )
        rows = result.all()
        if not rows:
            return 0

        fired = 0
        for trigger, session, world in rows:
            cur = _parse_time(world.current_time or "")
            fire_at = _parse_time(trigger.fire_at)
            if fire_at > cur:
                continue
            try:
                await _fire_trigger(db, trigger, session, world)
                fired += 1
            except Exception as e:
                log.error("trigger_fire_failed", trigger_id=str(trigger.id), error=str(e))
        if fired:
            await db.commit()
        return fired


async def _fire_trigger(db, trigger: DeferredTrigger, session: Session, world: World) -> None:
    """Fire a single trigger: produce narrative + apply state patch."""
    settings_map = await get_all_settings(db)
    llm = LlmClient(settings_map)

    system_prompt = get_prompt("trigger_runner", world.language).format(
        description=trigger.description,
        payload=json.dumps(trigger.payload, ensure_ascii=False)[:600],
        state=json.dumps(world.state, ensure_ascii=False)[:1000],
    )

    response = await llm.chat(
        messages=[{"role": "system", "content": system_prompt}],
        temperature=0.5,
        max_tokens=500,
        purpose="trigger",
        session_id=session.id,
        db=db,
    )

    # Parse response
    parsed: Dict[str, Any] = {}
    m = re.search(r"\{[\s\S]*\}", response.text or "")
    if m:
        try:
            parsed = json.loads(m.group(0))
        except json.JSONDecodeError:
            pass

    # Apply state patch
    state_patch = parsed.get("state_patch", {})
    if state_patch:
        new_state = apply_patch(world.state, state_patch)
        schema = world.definition.get("world_schema", {})
        ok, errors = validate_state(new_state, schema)
        if ok:
            world.state = new_state

    narrative = parsed.get("narrative", "")
    should_notify = bool(parsed.get("should_notify_player", True))

    # Save as message
    next_seq_result = await db.execute(
        select(Message.seq).where(Message.session_id == session.id).order_by(Message.seq.desc()).limit(1)
    )
    row = next_seq_result.first()
    next_seq = (row[0] + 1) if row else 1

    if should_notify and narrative:
        msg = Message(
            session_id=session.id,
            seq=next_seq,
            role="system",
            kind="narrative_step",
            content=f"[Событие] {narrative}",
            payload={
                "trigger_id": str(trigger.id),
                "triggered_at": trigger.fire_at,
                "outcome": parsed.get("outcome", trigger.description),
                "world_time": world.current_time,
                "player_state": world.state.get("player", {}),
                "options": [],  # triggers don't usually offer choices
            },
            is_pinned=True,
            hidden=False,
        )
    else:
        msg = Message(
            session_id=session.id,
            seq=next_seq,
            role="system",
            kind="technical_offscreen",
            content=f"[Trigger fired: {trigger.description}] Outcome: {parsed.get('outcome', '')}",
            payload={
                "trigger_id": str(trigger.id),
                "outcome": parsed.get("outcome", ""),
                "state_patch": state_patch,
            },
            is_pinned=False,
            hidden=True,
        )
    db.add(msg)
    trigger.fired = True
    log.info("trigger_fired", trigger_id=str(trigger.id), session_id=str(session.id))


async def main_loop():
    """Main worker loop. Polls every N seconds for due triggers.

    Resilient to transient DB errors: any error inside an iteration is logged
    and the loop sleeps for a fallback interval before retrying, instead of
    crashing the worker process.
    """
    setup_logging()
    log.info("trigger_worker_started")
    fallback_interval = 30  # used when settings table is unreadable
    while True:
        interval = fallback_interval
        try:
            async with AsyncSessionLocal() as db:
                enabled = await _get_setting(db, "triggers.enabled", True)
                interval = int(await _get_setting(db, "triggers.check_interval", fallback_interval))
            if enabled:
                fired = await check_and_fire_triggers()
                if fired:
                    log.info("triggers_fired", count=fired)
        except Exception as e:
            log.error("trigger_worker_iteration_failed", error=f"{type(e).__name__}: {e}")
        # Always sleep with a safe positive interval; never let a DB error
        # escape this loop and crash the worker.
        try:
            sleep_for = max(5, int(interval))
        except Exception:
            sleep_for = fallback_interval
        await asyncio.sleep(sleep_for)


async def _get_setting(db, key: str, default):
    from app.models import Setting
    result = await db.execute(select(Setting).where(Setting.key == key))
    row = result.scalars().first()
    if row is None:
        return default
    return cast_setting(key, row.value)


async def _get_setting_sleep() -> int:
    async with AsyncSessionLocal() as db:
        return int(await _get_setting(db, "triggers.check_interval", 30))


if __name__ == "__main__":
    asyncio.run(main_loop())