backend/src/core/world_engine.py

"""world_engine — the main simulation orchestrator.

Owns the tick loop (snapshot, act, perceive, decide, replan, resolve),
day handoff, proximity conversations, events, wellbeing updates, health
observation, and per-tick checkpointing; also runs headless via CLI.

Architecture: the hub of the backend — every subsystem (brain, planner,
Actions, memory, events, checkpoints) is called from here; Odin.py hosts
it as a background task.
Design: LLM calls run in worker threads and asyncio tasks so latency
never freezes the clock; deterministic phases (resolve) and stochastic
phases (decide) are strictly separated.
"""

from __future__ import annotations

import asyncio
import hashlib
import json
import math
import random
import sys
import time as _time
from collections import deque
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Callable, Dict, List, Optional
from types import SimpleNamespace

from src.core.log import get_logger

BACKEND_ROOT = Path(__file__).resolve().parents[2]
if str(BACKEND_ROOT) not in sys.path:
    sys.path.append(str(BACKEND_ROOT))


from src.agents.Actions import AgentActionManager, LocationResolver, ActionState, ActionType
from src.agents.conversation import (
    generate_conversation,
    RelationshipMatrix,
    _is_action_blocked,
)
from src.agents.Short_term import (
    save_day_plan,
    load_day_plan,
    date_from_simulation_time,
    append_conversation,
    append_event,
    archive_to_long_term,
    consolidate_to_single_day,
    clear_short_term_data,
    reset_day_runtime,
)
from src.config import (
    PERSONALITIES_DIR,
    SIM_MINUTES_PER_TICK,
    REAL_SECONDS_PER_TICK,
    MAX_CONVERSATIONS_PER_AGENT,
)
from src.core.agent_registry import AgentRegistry, AgentRuntimeState
from src.core.checkpoint_manager import (
    save_checkpoint,
    save_history,
    load_checkpoint,
    list_checkpoints,
    prune_checkpoints,
    latest_tick,
    KEEP_LAST,
)
from src.core.snapshot import take_snapshot, WorldSnapshot
from src.core.world_state import WorldState, Position, CurrentAction, AgentStatus
from src.core.world_events import WorldEventManager
from src.core.runtime_health import RuntimeHealthMonitor
from src.llm.gemini_client import ProviderFailureError
from src import config as _cfg

logger = get_logger(__name__)


class WorldEngine:
    """Main simulation orchestrator."""

    def __init__(
        self,
        sim_start_date: str = "2026-07-03",
        sim_start_hhmm: str = "00:00",
        sim_end_hhmm: str = "24:00",
    ):
        self.sim_start_date = sim_start_date
        self.sim_start_hhmm = sim_start_hhmm
        self.sim_end_hhmm = sim_end_hhmm

        self.world = WorldState()
        self.registry = AgentRegistry()
        self.resolver = LocationResolver()
        self.relationship_matrix = RelationshipMatrix()
        self.event_manager = WorldEventManager()
        self._applied_event_effects: set[str] = set()
        self.health_monitor = RuntimeHealthMonitor()

        self._day_index = 0
        # Per-agent set of subject ids currently within the perception circle
        # (edge-triggered so we only record a memory when someone *enters* range).
        self._in_range: Dict[str, set] = {}
        # Per-agent Brain (cognition) instances, built lazily. Each commands a
        # Body (the action state machine) — the human-like brain/body split.
        self._brains: Dict[str, Any] = {}
        # Rolling feed of recently-completed conversations (for the UI feed).
        self._recent_convs: deque = deque(maxlen=12)
        # Pixels already used at spawn, so agents don't stack on each other.
        self._spawn_points: List[tuple] = []
        # Per-agent observation fingerprint from the previous tick (novelty detection).
        # Key: agent_id, Value: frozenset of (agent_id, action_description) tuples.
        self._last_obs: Dict[str, frozenset] = {}
        # Per-agent cached observations from the current tick's perceive phase,
        # so Phase 4 (LLM decide) can reuse them without recomputing.
        self._tick_observations: Dict[str, list] = {}
        # Last tick at which each agent started an advisory LLM decision.  A
        # changing proximity observation must not generate an API call per tick.
        self._last_decision_tick: Dict[str, int] = {}
        # Background LLM work is not serializable, but its deterministic input
        # envelope is.  Keep that envelope so a restore can restart an
        # in-flight conversation instead of silently losing it.
        self._pending_conversations: Dict[str, Dict[str, Any]] = {}
        self._conversation_tasks: Dict[str, asyncio.Task] = {}
        # Decisions are advisory; a slow provider response must not freeze the
        # simulation clock or WebSocket snapshots at an action boundary.
        self._decision_tasks: Dict[str, asyncio.Task] = {}
        # Advisory throttle: last tick an "unscheduled downtime" recovery
        # replan was attempted per agent.  Not checkpointed — on restore a
        # fresh attempt is harmless.
        self._downtime_replan_tick: Dict[str, int] = {}

    @staticmethod
    def _conversation_key(first_id: str, second_id: str) -> str:
        return "|".join(sorted((first_id, second_id)))

    def checkpoint_state(self) -> Dict[str, Any]:
        """Serialize engine-owned state that is not held by WorldState."""
        return {
            "sim_start_date": self.sim_start_date,
            "sim_start_hhmm": self.sim_start_hhmm,
            "sim_end_hhmm": self.sim_end_hhmm,
            "day_index": self._day_index,
            "in_range": {
                agent_id: sorted(subject_ids)
                for agent_id, subject_ids in self._in_range.items()
            },
            # Novelty fingerprints affect whether an agent makes an LLM
            # decision next tick, so they are behavioural state rather than a
            # disposable UI cache.
            "last_observations": {
                agent_id: [list(item) for item in observations]
                for agent_id, observations in self._last_obs.items()
            },
            "last_decision_tick": dict(self._last_decision_tick),
            "recent_conversations": list(self._recent_convs),
            "relationship_matrix": self.relationship_matrix.snapshot(),
            "event_attendance": self.event_manager.attendance_snapshot(),
            "applied_event_effects": sorted(self._applied_event_effects),
            "pending_conversations": list(self._pending_conversations.values()),
        }

    def restore_checkpoint_state(self, state: Dict[str, Any]) -> None:
        """Restore engine-owned state after the world and registry are loaded.

        The caller replaces ``self.registry`` with newly deserialized manager
        instances before this method runs.  Brains own BodyControllers, which
        retain a manager reference, so cached brains must never survive that
        replacement.  Otherwise the clock advances an old manager while the
        restored registry (and therefore the UI) remains frozen on its old
        action.
        """
        self._brains.clear()
        if not state:
            return
        self.sim_start_date = state.get("sim_start_date", self.sim_start_date)
        self.sim_start_hhmm = state.get("sim_start_hhmm", self.sim_start_hhmm)
        self.sim_end_hhmm = state.get("sim_end_hhmm", self.sim_end_hhmm)
        self._day_index = state.get("day_index", self._day_index)
        self._in_range = {
            agent_id: set(subject_ids)
            for agent_id, subject_ids in state.get("in_range", {}).items()
        }
        self._last_obs = {
            agent_id: frozenset(tuple(item) for item in observations)
            for agent_id, observations in state.get("last_observations", {}).items()
        }
        self._last_decision_tick = {
            agent_id: int(tick)
            for agent_id, tick in state.get("last_decision_tick", {}).items()
        }
        self._recent_convs = deque(state.get("recent_conversations", []), maxlen=12)
        relationship_state = state.get("relationship_matrix")
        if relationship_state is not None:
            self.relationship_matrix.restore(relationship_state)
        self.event_manager.restore_attendance(state.get("event_attendance", {}))
        self._applied_event_effects = set(state.get("applied_event_effects", []))

        # Older checkpoints did not store a personality baseline. Upgrade
        # them on restore so a resumed day gets the same mood normalization as
        # a fresh day without invalidating its current action state.
        for agent in self.registry.all_states():
            if agent.emotion_baseline == 0.5:
                agent.emotion_baseline = self._emotion_baseline(agent.persona)

        self._pending_conversations = {}
        for pending in state.get("pending_conversations", []):
            first_id = pending.get("agent_a_id")
            second_id = pending.get("agent_b_id")
            if first_id and second_id:
                self._pending_conversations[self._conversation_key(first_id, second_id)] = pending

    async def resume_pending_conversations(self) -> None:
        """Restart durable conversation requests after a checkpoint restore.

        An API call already executing at save time cannot be serialized.  Its
        request envelope is retained, so this resumes the same logical work
        from the original participants, prompt inputs, and simulation time.
        """
        for key, pending in list(self._pending_conversations.items()):
            try:
                first = self.registry.get(pending["agent_a_id"])
                second = self.registry.get(pending["agent_b_id"])
            except KeyError:
                self._pending_conversations.pop(key, None)
                continue
            if not first.paused or not second.paused:
                self._pending_conversations.pop(key, None)
                continue
            self._start_conversation_task(first, second, pending)

    # ------------------------------------------------------------------ #
    # Persona discovery
    # ------------------------------------------------------------------ #

    def _discover_personas(self) -> List[Dict[str, Any]]:
        """Find all persona JSON files and return their parsed data."""
        personas = []
        for json_path in sorted(PERSONALITIES_DIR.glob("**/*.json")):
            try:
                data = json.loads(json_path.read_text(encoding="utf-8"))
                if "Name" in data and "Hostel" in data:
                    personas.append(data)
            except Exception as e:
                logger.warning("[WorldEngine] failed to load %s: %s", json_path, e)
        logger.info(
            "[WorldEngine] discovered %d personas from %s", len(personas), PERSONALITIES_DIR
        )
        return personas

    def _apply_event_opportunities(self, date: str) -> None:
        """Overlay deterministic, conflict-safe optional event attendance.

        The overlay happens after cached or generated plans are available, so
        it works equally for fresh, resumed, and handoff days.  Only entirely
        flexible windows can be replaced; the event manager rejects hard
        commitments before this method sees a changed plan.
        """
        states = self.registry.all_states()
        if not states:
            return
        plans = {state.agent_id: state.day_plan for state in states}
        personas = {state.agent_id: state.persona for state in states}
        updated = self.event_manager.apply_opportunities(
            date, plans, personas, self.relationship_matrix.get,
        )
        for state in states:
            plan = updated.get(state.agent_id, state.day_plan)
            if plan == state.day_plan:
                continue
            state.day_plan = plan
            if state.manager:
                state.manager.replace_day_plan(plan)
            save_day_plan(state.persona_name, date, plan)

    def _apply_finished_event_effects(self, date: str, hhmm: str) -> None:
        """Persist bounded social/mood effects once an attended event ends."""
        now = self._hhmm_to_minutes(hhmm)
        changed_relationships = False
        for event in self.event_manager.for_date(date):
            if event.id in self._applied_event_effects or now < self._hhmm_to_minutes(event.end_time):
                continue
            attendees = self.event_manager.attendance_snapshot().get(event.id, [])
            for agent_id in attendees:
                try:
                    state = self.registry.get(agent_id)
                except KeyError:
                    continue
                state.energy_level = max(0.0, min(1.0, state.energy_level + event.energy_effect))
                state.emotion_state = max(0.0, min(1.0, state.emotion_state + event.emotion_effect))
                append_event(state.persona_name, date, {
                    "type": "world_event", "action": event.name,
                    "location": event.location_id,
                    "summary": f"Attended {event.name}: {event.description}",
                    "details": {"event_id": event.id, "ended_at": event.end_time},
                })
            if event.relationship_effect and len(attendees) > 1:
                for first in attendees:
                    for second in attendees:
                        if first != second:
                            self.relationship_matrix.update(first, second, event.relationship_effect)
                            changed_relationships = True
            self._applied_event_effects.add(event.id)
            logger.info("[Events] completed %s with %d attendee(s)", event.id, len(attendees))
        if changed_relationships:
            self.relationship_matrix.save()

    def _resolve_hostel_position(self, hostel_id: str) -> Position:
        """Convert a Hostel field value to a pixel Position."""
        pos = self.resolver.resolve(hostel_id)
        if pos is None:
            logger.warning(
                "[WorldEngine] could not resolve hostel '%s', using default (0,0)", hostel_id
            )
            pos = Position(x=0, y=0, location_id=hostel_id)
        return pos

    def _persona_name_to_id(self, name: str) -> str:
        """Convert a display name like 'Parv Singla' to a safe agent_id."""
        import re
        safe = re.sub(r"[^A-Za-z0-9._-]+", "_", name.strip()).strip("._-").lower()
        return safe or "unknown"

    @staticmethod
    def _emotion_baseline(persona: Dict[str, Any]) -> float:
        """Derive a stable, modest mood baseline from stated personality traits.

        This is deliberately narrow: students can have a good or difficult day,
        but their normal state should cluster around neutral rather than begin
        at an arbitrary emotional extreme.
        """
        traits = " ".join(str(persona.get(key, "")) for key in ("innate", "lifestyle", "learned")).lower()
        baseline = 0.50
        for marker in ("friendly", "warm", "approachable", "outgoing", "energetic", "playful", "resilient", "optimistic"):
            if marker in traits:
                baseline += 0.015
        for marker in ("introverted", "quiet", "reserved", "awkward", "irregular sleep", "bad sleep", "procrastinate"):
            if marker in traits:
                baseline -= 0.015
        return max(0.40, min(0.62, baseline))

    @staticmethod
    def _energy_baseline(persona: Dict[str, Any]) -> float:
        """Return a personality-informed morning energy level, not 100%."""
        traits = " ".join(str(persona.get(key, "")) for key in ("innate", "lifestyle", "learned")).lower()
        baseline = 0.74
        for marker in ("regular sleep", "early riser", "exercise", "sport", "disciplined", "morning person"):
            if marker in traits:
                baseline += 0.025
        for marker in ("irregular sleep", "bad sleep", "late-night", "procrastinate", "overcommitted", "insomni"):
            if marker in traits:
                baseline -= 0.04
        return max(0.56, min(0.86, baseline))

    def _action_wellbeing_deltas(self, state: AgentRuntimeState, action: Any, duration: int) -> tuple[float, float]:
        """Return the total wellbeing effect for one action.

        When the day planner declares an energy_target for the action, the
        runtime glides the agent's energy from its current level toward that
        declared cumulative target (the LLM owns every value; this is only a
        smooth, deterministic path to it).  The exponential progress factor
        means short actions barely move energy while long ones converge, and
        because it depends only on stored state and the plan it stays
        checkpoint-reproducible.

        Without a declared target the legacy delta path applies: the planner's
        energy_change drives the level, with the small deterministic jitter
        keyed by agent/action so replays remain reproducible.
        """
        description = (getattr(action, "description", "") or "").lower()
        emotion = float(getattr(action, "emotion_change", 0.0) or 0.0)

        target = getattr(action, "energy_target", None)
        if target is not None:
            target = max(0.0, min(1.0, float(target)))
            remaining = target - state.energy_level
            progress = 1.0 - math.exp(-_cfg.SIM_ENERGY_FOLLOW_RATE * max(1, duration))
            energy = remaining * progress
            return energy, emotion

        energy = float(getattr(action, "energy_change", 0.0) or 0.0)

        variation = _cfg.SIM_WELLBEING_VARIABILITY
        token = f"{state.agent_id}|{getattr(action, 'start_time', '')}|{getattr(action, 'end_time', '')}|{description}"
        digest = hashlib.blake2s(token.encode("utf-8"), digest_size=4).digest()
        jitter = (int.from_bytes(digest, "big") / 0xFFFFFFFF) * 2.0 - 1.0
        energy += jitter * 0.035 * variation
        emotion += jitter * 0.045 * variation
        return energy, emotion

    def _memory_context(self, persona_name, persona, before_date=None, query_hint=""):
        """Build (relevant_memories, rolling_summary) for a day-planner call.

        Uses the configured long-term memory backend (keyword by default). This
        is 0-LLM for the keyword backend. Failures degrade to empty memory so
        planning never breaks.
        """
        try:
            from src.agents.Long_term import get_retriever
            retr = get_retriever()
            query = query_hint or " ".join(
                str(persona.get(k, ""))
                for k in ("goals", "hobbies", "daily_plan_req", "Branch")
            )
            memories = retr.retrieve(persona_name, query, k=6)
            summary = retr.rolling_summary(persona_name, days=2, before_date=before_date)
            if memories or summary:
                logger.info(
                    "[WorldEngine]   memory for '%s': %d recalls, summary=%s",
                    persona_name, len(memories), "yes" if summary else "no",
                )
            return memories, summary
        except Exception as e:
            logger.warning("[WorldEngine] memory context failed for %s: %s", persona_name, e)
            return [], None

    # ------------------------------------------------------------------ #
    # Initialization
    # ------------------------------------------------------------------ #

    async def initialize(self) -> None:
        """
        Phase 1: Discover personas, generate day plans, register agents.
        Does NOT advance the clock — all agents are ready at tick 0.
        """
        logger.info("[WorldEngine] ========== INITIALIZATION ==========")
        t0 = _time.perf_counter()

        # Initialise the Qdrant-only long-term memory backend once at startup.
        try:
            from src.agents.Long_term import get_retriever
            get_retriever()
        except Exception as exc:
            logger.warning("[WorldEngine] memory backend initialization failed: %s", exc)

        personas = self._discover_personas()
        if not personas:
            logger.error("[WorldEngine] no personas found — aborting")
            return
        self.relationship_matrix.ensure_grounded_seed([
            {"agent_id": self._persona_name_to_id(persona["Name"]), "persona_name": persona["Name"]}
            for persona in personas
        ])

        from src.agents.day_planner import run as day_planner_run

        # ── Phase 1a (sequential, fast): resolve positions, check saved plans ──
        logger.info("[WorldEngine] preparing %d agents ...", len(personas))
        agent_setups = []
        for persona in personas:
            try:
                name = persona.get("Name", "unknown")
                hostel = persona.get("Hostel", "")
                agent_id = self._persona_name_to_id(name)
                position = self.resolver.random_interior_point(hostel, occupied=self._spawn_points)
                self._spawn_points.append((position.x, position.y))
                consolidate_to_single_day(name, self.sim_start_date)
                existing_plan = load_day_plan(name, self.sim_start_date)
                if existing_plan:
                    reset_day_runtime(name, self.sim_start_date)
                    logger.info(
                        "[WorldEngine]   reusing saved day plan for '%s' (%d actions) — no LLM",
                        name, len(existing_plan),
                    )
                    day_plan = existing_plan
                    error = None
                    manager = AgentActionManager(agent_id, day_plan, position, self.resolver)
                    self.registry.register(
                        AgentRuntimeState(
                            agent_id=agent_id, persona=persona, persona_name=name,
                            manager=manager, position=position, day_plan=day_plan,
                            emotion_state=self._emotion_baseline(persona),
                            emotion_baseline=self._emotion_baseline(persona), energy_level=self._energy_baseline(persona),
                        )
                    )
                    self.world.register_agent(agent_id, position)
                    logger.info("[WorldEngine]   registered '%s' (hostel=%s) — %d plan actions (cached)",
                                name, hostel, len(day_plan))
                else:
                    agent_setups.append((name, hostel, agent_id, position, persona))
            except Exception as e:
                logger.error("[WorldEngine]   failed to prepare agent '%s': %s — skipping",
                             persona.get("Name", "unknown"), e)
                import traceback; traceback.print_exc()
                continue

        # ── Phase 1b (parallel, slow): generate day plans for agents that need one ──
        # A full plan makes several provider calls. Serialize startup planning
        # to prevent concurrent agents from stampeding the shared key ring.
        logger.info("[WorldEngine] generating day plans for %d agents (sequential) ...", len(agent_setups))
        plan_t0 = _time.perf_counter()

        async def _plan_one(name: str, hostel: str, agent_id: str,
                            position, persona: dict) -> None:
            logger.info("[WorldEngine]   starting day plan for '%s' ...", name)
            mem, yday = self._memory_context(name, persona, before_date=self.sim_start_date)
            proxy = SimpleNamespace(persona=persona, relevant_memories=mem, yesterday_summary=yday)
            try:
                _loop = asyncio.get_event_loop()
                plan_result = await _loop.run_in_executor(
                    None,
                    lambda: day_planner_run(
                        proxy,
                        {
                            "current_time": f"{self.sim_start_date} {self.sim_start_hhmm}",
                            "places": None,
                            "persona_name": name,
                            "mode": "full_day",
                            "current_location_id": hostel,
                            "energy_level": self._energy_baseline(persona),
                            "emotion_state": self._emotion_baseline(persona),
                            "upcoming_events": self.event_manager.snapshot(self.sim_start_date, self.sim_start_hhmm).get("upcoming", []),
                        },
                    ),
                )
                day_plan = plan_result.get("day_plan", [])
                error = plan_result.get("error")
                if error:
                    logger.warning("[WorldEngine]   day plan for '%s' has issues: %s", name, error)
                manager = AgentActionManager(agent_id, day_plan, position, self.resolver)
                self.registry.register(
                    AgentRuntimeState(
                        agent_id=agent_id, persona=persona, persona_name=name,
                        manager=manager, position=position, day_plan=day_plan,
                        emotion_state=self._emotion_baseline(persona),
                        emotion_baseline=self._emotion_baseline(persona), energy_level=self._energy_baseline(persona),
                    )
                )
                self.world.register_agent(agent_id, position)
                logger.info("[WorldEngine]   registered '%s' (hostel=%s) — %d plan actions",
                            name, hostel, len(day_plan))
            except ProviderFailureError:
                # Never continue startup with an empty-plan agent after every
                # available key rejects a request. Odin reports this cleanly.
                raise
            except Exception as e:
                logger.error("[WorldEngine]   day plan failed for '%s': %s — skipping", name, e)
                import traceback; traceback.print_exc()
                # Register anyway with empty plan so the sim doesn't crash
                manager = AgentActionManager(agent_id, [], position, self.resolver)
                self.registry.register(
                    AgentRuntimeState(
                        agent_id=agent_id, persona=persona, persona_name=name,
                        manager=manager, position=position, day_plan=[],
                        emotion_state=self._emotion_baseline(persona),
                        emotion_baseline=self._emotion_baseline(persona), energy_level=self._energy_baseline(persona),
                    )
                )
                self.world.register_agent(agent_id, position)
                logger.info("[WorldEngine]   registered '%s' with empty plan (fallback)", name)

        for name, hostel, agent_id, position, persona in agent_setups:
            await _plan_one(name, hostel, agent_id, position, persona)

        self._apply_event_opportunities(self.sim_start_date)

        plan_elapsed = _time.perf_counter() - plan_t0
        logger.info("[WorldEngine] all day plans generated in %.1fs", plan_elapsed)

        # Assign frontend colors
        _COLORS = [
            "#ffcc33", "#ff6b6b", "#51cf66", "#339af0", "#cc5de8",
            "#f76707", "#20c997", "#f06595", "#748ffc", "#ffd43b",
        ]
        for i, state in enumerate(self.registry.all_states()):
            state.color = _COLORS[i % len(_COLORS)]

        # Register hostel rooms as world resources
        for state in self.registry.all_states():
            hostel = state.persona.get("Hostel", "")
            if hostel:
                self.world.register_resource(hostel)

        elapsed = _time.perf_counter() - t0
        # Start the clock at the configured time of day (e.g. 08:00) so the day
        # begins with real activity instead of everyone asleep at midnight.
        self.world.tick = self._hhmm_to_minutes(self.sim_start_hhmm)
        # Persist the assembled roster before the first driver tick. This lets
        # an observer use stopped-only roster controls even when an API quota
        # failure prevents the very first tick from running.
        save_checkpoint(self.world, self.registry, self.world.tick, self.checkpoint_state())
        logger.info(
            "[WorldEngine] initialized %d agents in %.1fs — starting simulation at %s",
            len(self.registry), elapsed, self.sim_start_hhmm,
        )

    # ------------------------------------------------------------------ #
    # Tick loop
    # ------------------------------------------------------------------ #

    async def run_tick(self) -> Dict[str, Any]:
        """
        Execute one simulation tick in six phases:

          1. Snapshot  — freeze world state (one copy for all agents).
          2. Act       — all agents advance their body state machines in parallel.
          3. Perceive  — build observations from snapshot, detect novelty, set
                         flags for LLM decision; conversation detection happens here.
          4. LLM Decide— parallel LLM decision for agents with novel observations.
          5. Replan    — regenerate remaining-day plan for agents that chose replan.
          6. Resolve   — sequential: timeouts, day transitions, collisions, sync.

        Returns a dict snapshot for frontend use.
        """
        from src.llm.gemini_client import provider_failure
        failure = provider_failure()
        if failure:
            raise failure
        current_tick = self.world.tick
        hhmm = self._minutes_to_hhmm(current_tick % (24 * 60))

        # A generated conversation remains a real, visible simulation state
        # until its declared end, rather than disappearing with the LLM task.
        self._complete_finished_conversations(current_tick)

        agent_states = self.registry.all_states()

        # ══════ PHASE 2: Act (parallel) ══════
        if agent_states:
            await self._phase_act(agent_states, current_tick, hhmm)

        # Perception must observe the positions and actions produced by this
        # tick's act phase, rather than a one-tick-old world mirror.
        self._sync_to_world_state()
        snapshot = take_snapshot(self.world)

        # ══════ PHASE 3: Perceive (parallel) + conversation detection (sequential) ══════
        novelty_flags: Dict[str, bool] = {}  # agent_id -> has novel observations
        self._tick_observations.clear()
        if agent_states:
            perceive_results = await asyncio.gather(
                *[self._phase_perceive(s, snapshot) for s in agent_states],
                return_exceptions=True,
            )
            for state, result in zip(agent_states, perceive_results):
                if isinstance(result, Exception):
                    logger.error(
                        "[WorldEngine] agent '%s' perceive failed: %s", state.agent_id, result,
                    )
                else:
                    novelty_flags[state.agent_id], obs = result
                    self._tick_observations[state.agent_id] = obs

        # Phase 3b: Conversation detection (sequential)
        await self._check_conversations(current_tick, hhmm)
        self._timeout_conversations(current_tick)

        # ══════ PHASE 4: LLM Decide (parallel, only non-paused agents with novelty) ══════
        decide_results: Dict[str, Any] = self._collect_finished_decisions()
        decide_agents = [
            s for s in agent_states
            if not s.paused and novelty_flags.get(s.agent_id, False)
            and s.energy_level >= _cfg.DECIDE_MIN_ENERGY
            and s.emotion_state >= _cfg.DECIDE_MIN_EMOTION
            and current_tick - self._last_decision_tick.get(
                s.agent_id, -_cfg.DECIDE_COOLDOWN_TICKS
            ) >= _cfg.DECIDE_COOLDOWN_TICKS
        ]
        if decide_agents:
            for state in decide_agents:
                if state.agent_id in self._decision_tasks:
                    continue
                self._last_decision_tick[state.agent_id] = current_tick
                self._decision_tasks[state.agent_id] = asyncio.create_task(
                    self._phase_llm_decide(state, current_tick, hhmm),
                    name=f"decide:{state.agent_id}:{current_tick}",
                )

        # ══════ PHASE 5: Replan (parallel, agents whose LLM decision was "replan") ══════
        replan_agents = [
            s for s in agent_states
            if decide_results.get(s.agent_id) == "replan"
            and s.replan_count < _cfg.MAX_REPLANS_PER_AGENT_PER_DAY
        ]
        if replan_agents:
            await asyncio.gather(
                *[self._phase_replan(s, current_tick, hhmm) for s in replan_agents],
                return_exceptions=True,
            )

        # ══════ PHASE 5b: Unscheduled downtime recovery (parallel, deterministic) ══════
        # A force-accepted fallback plan strands the agent on "Unscheduled
        # downtime" for the rest of the day.  The LLM decide path may never
        # fire for such an agent, so detect it here and replan the remaining
        # whole day explicitly.
        downtime_agents = [
            s for s in agent_states if self._has_unscheduled_downtime(s, current_tick)
        ]
        if downtime_agents:
            for s in downtime_agents:
                self._downtime_replan_tick[s.agent_id] = current_tick
                logger.info(
                    "[WorldEngine] agent '%s' stuck on unscheduled downtime — replanning remaining day",
                    s.persona_name,
                )
            await asyncio.gather(
                *[self._phase_replan(s, current_tick, hhmm) for s in downtime_agents],
                return_exceptions=True,
            )

        # ══════ PHASE 6: Resolve (sequential) ══════
        await self._check_last_action_triggers(current_tick, hhmm)
        self._apply_finished_event_effects(self.sim_start_date, hhmm)
        self._resolve_collisions()
        self._sync_to_world_state()
        # ----- Advance tick -----
        self.world.advance_tick(minutes=SIM_MINUTES_PER_TICK)
        snapshot_tick = self.world.tick
        snapshot_hhmm = self._minutes_to_hhmm(snapshot_tick % (24 * 60))
        # A frontend frame must describe the same clock state as the engine
        # and checkpoint it was built from.  Returning the pre-advance clock
        # made the UI one tick behind rewind/fast-forward controls.
        health = self.health_monitor.observe(self, snapshot_tick, snapshot_hhmm)

        # ----- Save checkpoint -----
        save_checkpoint(
            self.world,
            self.registry,
            self.world.tick,
            engine_state=self.checkpoint_state(),
        )
        prune_checkpoints(keep_last=KEEP_LAST)

        # Build frontend snapshot
        return self._frontend_snapshot(snapshot_tick, snapshot_hhmm, health)

    def _frontend_snapshot(self, tick: int, hhmm: str, health: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
        """Return the current UI-safe projection without advancing the world."""
        real_min_per_day = (1440 * _cfg.REAL_SECONDS_PER_SIM_MINUTE
                            / max(_cfg.TICK_SPEED, 1e-9)) / 60.0
        return {
            "tick": tick,
            "time": hhmm,
            "day": self._day_index + 1,
            "speed": {
                "multiplier": _cfg.TICK_SPEED,
                "real_min_per_day": round(real_min_per_day),
                "real_ms_per_sim_minute": round((_cfg.REAL_SECONDS_PER_SIM_MINUTE / max(_cfg.TICK_SPEED, 1e-9)) * 1000),
            },
            "recent_conversations": list(self._recent_convs),
            "events": self.event_manager.snapshot(self.sim_start_date, hhmm),
            "health": health if health is not None else self.health_monitor.latest,
            "agents": {
                s.agent_id: {
                    "name": s.persona_name,
                    "color": s.color,
                    "position": {
                        "x": s.position.x,
                        "y": s.position.y,
                        "location_id": s.position.location_id,
                    },
                    "current_action": (
                        s.manager.current_action.model_dump()
                        if s.manager and s.manager.current_action
                        else None
                    ),
                    "activity": (
                        s.manager.current_action.description
                        if s.manager and s.manager.current_action
                        else "Idle"
                    ),
                    "paused": s.paused,
                    "in_conversation": bool(s.paused and s.active_conversation),
                    "in_last_action": s.manager.is_last_action if s.manager else False,
                    "energy_level": s.energy_level,
                    "emotion_state": s.emotion_state,
                    "conversation": s.active_conversation,
                }
                for s in self.registry.all_states()
            },
        }

    def current_frontend_snapshot(self) -> Dict[str, Any]:
        """Expose restored state immediately after a checkpoint rewind."""
        tick = self.world.tick
        hhmm = self._minutes_to_hhmm(tick % (24 * 60))
        return self._frontend_snapshot(tick, hhmm)

    async def _phase_act(self, agent_states, current_tick, hhmm) -> None:
        """Phase 2: all agents advance their body state machines in parallel."""
        act_results = await asyncio.gather(
            *[self._run_agent_act(s, current_tick, hhmm) for s in agent_states],
            return_exceptions=True,
        )
        for state, result in zip(agent_states, act_results):
            if isinstance(result, Exception):
                logger.error(
                    "[WorldEngine] agent '%s' act failed: %s", state.agent_id, result,
                )

    def _collect_finished_decisions(self) -> Dict[str, str]:
        """Consume completed advisory decisions without delaying a tick."""
        completed: Dict[str, str] = {}
        for agent_id, task in list(self._decision_tasks.items()):
            if not task.done():
                continue
            self._decision_tasks.pop(agent_id, None)
            try:
                completed[agent_id] = task.result()
            except asyncio.CancelledError:
                continue
            except Exception as exc:
                from src.llm.gemini_client import ProviderFailureError
                if isinstance(exc, ProviderFailureError):
                    raise
                logger.error("[WorldEngine] agent '%s' LLM decide failed: %s", agent_id, exc)
                completed[agent_id] = "continue"
        return completed

    async def _phase_perceive(self, state: AgentRuntimeState, snapshot: WorldSnapshot) -> tuple:
        """Phase 3: build observations for one agent, detect novelty.
        Returns (novel: bool, observations: list).
        """
        if state.paused:
            return False, []
        observations = self._build_observations(state, snapshot)
        # Compute fingerprint: frozenset of (agent_id, action_description)
        fingerprint = frozenset(
            (o["agent_id"], o.get("current_action", "")) for o in observations
        )
        prev = self._last_obs.get(state.agent_id, frozenset())
        novel = fingerprint != prev
        self._last_obs[state.agent_id] = fingerprint
        return novel, observations

    async def _phase_llm_decide(self, state: AgentRuntimeState, tick: int, hhmm: str) -> str:
        """Phase 4: call the brain's LLM decision for one agent.
        Returns "continue" or "replan".
        """
        from src.core.budget import GOVERNOR
        if not GOVERNOR.can_afford("decide", cost=1):
            logger.info("[WorldEngine] budget denied decide for '%s' — continuing", state.persona_name)
            return "continue"
        brain = self._brain_for(state)
        observations = self._tick_observations.get(state.agent_id, [])
        query = " ".join(str(item.get("current_action", "")) for item in observations)
        memories, _ = self._memory_context(state.persona_name, state.persona, query_hint=query)
        # Gemini is a synchronous SDK call.  Run it in the default worker
        # pool so a slow response, rate-limit wait, or retry never prevents
        # WebSocket broadcasts and the rest of the simulation from advancing.
        decision = await asyncio.to_thread(
            brain.decide_tick,
            tick=tick,
            hhmm=hhmm,
            observations=observations,
            day_plan=state.day_plan,
            replan_count=state.replan_count,
            max_replans=_cfg.MAX_REPLANS_PER_AGENT_PER_DAY,
            energy_level=state.energy_level,
            emotion_state=state.emotion_state,
            relevant_memories=memories,
        )
        return decision.decision

    async def _phase_replan(self, state: AgentRuntimeState, tick: int, hhmm: str) -> None:
        """Phase 5: regenerate remaining-day plan for one agent.
        Called only when the LLM returned "replan" and budget allows.
        """
        from src.core.budget import GOVERNOR
        if not GOVERNOR.can_afford("replan", cost=4):
            logger.info("[WorldEngine] budget denied replan for '%s' — skipping", state.persona_name)
            return
        from src.agents.day_planner import run as day_planner_run
        from types import SimpleNamespace

        loop = asyncio.get_event_loop()
        mem, yday = self._memory_context(
            state.persona_name, state.persona,
            before_date=date_from_simulation_time(f"{self.sim_start_date} {hhmm}"),
        )
        proxy = SimpleNamespace(
            persona=state.persona,
            relevant_memories=mem,
            yesterday_summary=yday,
        )
        try:
            plan_result = await loop.run_in_executor(
                None,
                lambda: day_planner_run(
                    proxy,
                    {
                        "current_time": f"{self.sim_start_date} {hhmm}",
                        "places": None,
                        "persona_name": state.persona_name,
                        "mode": "remaining",
                        "current_location_id": state.position.location_id,
                        "energy_level": state.energy_level,
                        "emotion_state": state.emotion_state,
                        "upcoming_events": self.event_manager.snapshot(self.sim_start_date, hhmm).get("upcoming", []),
                    },
                ),
            )
            new_plan = plan_result.get("day_plan", [])
            if plan_result.get("replan_rejected"):
                logger.warning(
                    "[WorldEngine] replan rejected for '%s'; retaining prior %d-action plan: %s",
                    state.persona_name,
                    len(state.day_plan),
                    plan_result.get("error", "validation retries exhausted"),
                )
                return
            if new_plan and state.manager:
                state.day_plan = new_plan
                state.manager.replace_day_plan(new_plan)
                state.replan_count += 1
                logger.info(
                    "[WorldEngine] replan for '%s': %d actions (count=%d)",
                    state.persona_name, len(new_plan), state.replan_count,
                )
        except Exception as e:
            logger.error(
                "[WorldEngine] replan failed for '%s': %s", state.persona_name, e,
            )

    def _has_unscheduled_downtime(self, state: AgentRuntimeState, tick: int) -> bool:
        """Detect agents stranded on the deterministic fallback schedule.

        The LLM decide path is gated (novelty, energy/emotion, cooldown,
        budget), so a force-accepted fallback day can leave an agent stuck on
        "Unscheduled downtime" for hours with no replan ever firing.  This
        backstop scans the remaining plan every tick and flags it."""
        if state.paused:
            return False
        if state.day_archived:
            return False
        if state.manager is None:
            return False
        if state.replan_count >= _cfg.MAX_REPLANS_PER_AGENT_PER_DAY:
            return False
        if (
            tick - self._downtime_replan_tick.get(state.agent_id, -10**9)
            < _cfg.DOWNTIME_REPLAN_COOLDOWN_TICKS
        ):
            return False
        now_minutes = tick % (24 * 60)
        if now_minutes >= 24 * 60 - _cfg.DOWNTIME_REPLAN_MIN_HORIZON:
            # Too little of the day remains to justify a replan.
            return False
        for action in state.day_plan:
            end = self._hhmm_to_minutes(str(action.get("end", "")))
            if end > now_minutes and "unscheduled downtime" in str(
                action.get("action", "")
            ).lower():
                return True
        return False

    async def _run_agent_act(
        self, state: AgentRuntimeState, tick: int, hhmm: str
    ) -> None:
        """Phase 2: execute the decision — advance the body's state machine.

        Updates registry position and action from the manager after the body
        advances. Paused agents are skipped.
        """
        if state.paused:
            return
        manager = state.manager
        if manager is None:
            return
        brain = self._brain_for(state)
        action = brain.act(tick)
        state.position = manager.position
        state.current_action = action.model_dump() if action else None

        # Per-tick energy/emotion update
        self._update_energy_emotion(state)

    def _timeout_conversations(self, current_tick: int) -> None:
        """Fail safe only for LLM requests that are still generating."""
        for state in self.registry.all_states():
            if (
                not state.paused
                or not state.active_conversation
                or state.active_conversation.get("status") != "generating"
            ):
                continue
            if current_tick - state.conversation_start_tick < 30:
                continue
            partner_id = state.last_conversation_partner
            logger.info(
                "[WorldEngine] conversation timeout for '%s' (started tick %d, now tick %d)",
                state.persona_name, state.conversation_start_tick, current_tick,
            )
            self._end_conversation(state, current_tick)
            if partner_id:
                try:
                    partner = self.registry.get(partner_id)
                    self._end_conversation(partner, current_tick)
                    self._discard_pending_conversation(state.agent_id, partner.agent_id)
                except KeyError:
                    pass

    @staticmethod
    def _end_conversation(state: AgentRuntimeState, current_tick: Optional[int] = None) -> None:
        """Restore agent state after a conversation ends (timeout or normal)."""
        state.paused = False
        state.conversation_start_tick = 0
        state.active_conversation = None
        mgr = state.manager
        if mgr:
            mgr.resume_from_conversation(world_tick=current_tick)

    def _complete_finished_conversations(self, current_tick: int) -> None:
        """Resume each completed conversation once its simulated duration ends."""
        completed: set[str] = set()
        for state in self.registry.all_states():
            conversation = state.active_conversation or {}
            partner_id = conversation.get("partner_id")
            if (
                not state.paused
                or conversation.get("status") != "active"
                or current_tick < conversation.get("ends_at_tick", current_tick + 1)
                or not partner_id
                or state.agent_id in completed
            ):
                continue
            try:
                partner = self.registry.get(partner_id)
            except KeyError:
                self._end_conversation(state, current_tick)
                continue
            self._end_conversation(state, current_tick)
            self._end_conversation(partner, current_tick)
            self.world.record_conversation(state.agent_id, current_tick)
            self.world.record_conversation(partner.agent_id, current_tick)
            completed.update((state.agent_id, partner.agent_id))
            logger.info(
                "[WorldEngine] conversation '%s' <-> '%s' ended after its simulated duration",
                state.persona_name, partner.persona_name,
            )

    def _resolve_collisions(self, min_gap: int = 4) -> None:
        """Ensure no two agents share (nearly) the same pixel — nudge later ones
        a few pixels away. Keeps the map readable and honours 'never same pixel'."""
        seen: List[tuple] = []
        g2 = min_gap * min_gap
        for s in self.registry.all_states():
            x, y = s.position.x, s.position.y
            tries = 0
            while tries < 15 and any((x - ox) ** 2 + (y - oy) ** 2 < g2 for ox, oy in seen):
                x = s.position.x + random.randint(-7, 7)
                y = s.position.y + random.randint(-7, 7)
                tries += 1
            if (x, y) != (s.position.x, s.position.y):
                # Registry and action manager are two views of the live body.
                # Keep them atomic here; otherwise _run_agent_act restores the
                # old manager position on the next tick and the collision
                # reappears.
                corrected = Position(x=x, y=y, location_id=s.position.location_id)
                s.position = corrected
                if s.manager is not None:
                    s.manager.position = corrected
            seen.append((x, y))

    def _update_energy_emotion(self, state: AgentRuntimeState) -> None:
        """Apply the active action's bounded, personality-aware wellbeing effect."""
        manager = state.manager
        if manager is None or manager.current_action is None:
            return
        action = manager.current_action
        try:
            start_min = self._hhmm_to_minutes(action.start_time)
            end_min = self._hhmm_to_minutes(action.end_time)
            duration = max(1, end_min - start_min)
            tick_step = _cfg.SIM_MINUTES_PER_TICK
            action_energy_change, action_emotion_change = self._action_wellbeing_deltas(state, action, duration)
            energy_tick = (action_energy_change / duration) * tick_step
            emotion_tick = (action_emotion_change / duration) * tick_step
            state.energy_level = max(0.0, min(1.0, state.energy_level + energy_tick))
            state.emotion_state = max(0.0, min(1.0, state.emotion_state + emotion_tick))
        except Exception:
            pass

    def _brain_for(self, state: AgentRuntimeState):
        """Lazily build and cache one Brain (+ Body) per agent."""
        brain = self._brains.get(state.agent_id)
        if brain is None:
            from src.agents.body import BodyController
            from src.agents.brain import Brain
            brain = Brain(
                agent_id=state.agent_id,
                persona_name=state.persona_name,
                body=BodyController(state.manager),
                persona=state.persona,
            )
            self._brains[state.agent_id] = brain
        return brain

    def _build_observations(self, state: AgentRuntimeState, w_snapshot: WorldSnapshot) -> list:
        """Build structured observations for one agent from the frozen snapshot.

        Reports every agent within the configured pixel radius, including their
        position and current action. Placeholder for future world-events.
        """
        radius = float(_cfg.PERCEPTION_RADIUS_PX)
        nearby_snaps = w_snapshot.agents_within_px(state.agent_id, radius)
        observed = []
        for ns in nearby_snaps:
            try:
                rs = self.registry.get(ns.agent_id)
                name = rs.persona_name if rs else ns.agent_id
            except KeyError:
                name = ns.agent_id
            observed.append({
                "agent_id": ns.agent_id,
                "name": name,
                "position": {"x": ns.position.x, "y": ns.position.y},
                "current_action": ns.current_action.description if ns.current_action else "idle",
            })
        return observed

    # ------------------------------------------------------------------ #
    # Perception (0-LLM) + spatial helpers
    # ------------------------------------------------------------------ #

    def _neighbors_within_px(self, state, radius_px: float) -> List:
        """Registry states whose pixel position is within radius_px of `state`
        (excludes self). Nearest first."""
        out = []
        for other in self.registry.all_states():
            if other.agent_id == state.agent_id:
                continue
            dx = other.position.x - state.position.x
            dy = other.position.y - state.position.y
            d2 = dx * dx + dy * dy
            if d2 <= radius_px * radius_px:
                out.append((d2, other))
        out.sort(key=lambda t: t[0])
        return [o for _d, o in out]

    def _perceive_and_record(self, tick: int, hhmm: str) -> None:
        """For each agent, find who is within the perception circle and record a
        memory the moment someone *enters* range (edge-triggered, de-duplicated).
        Pure spatial math — no LLM. Cheap keyword-backend writes only.
        """
        radius = float(_cfg.PERCEPTION_RADIUS_PX)
        try:
            from src.agents.Long_term import get_retriever
            retriever = get_retriever()
        except Exception:
            retriever = None

        date_str = date_from_simulation_time(f"{self.sim_start_date} {hhmm}")

        for state in self.registry.all_states():
            neighbors = self._neighbors_within_px(state, radius)
            current_ids = {n.agent_id for n in neighbors}
            previous_ids = self._in_range.get(state.agent_id, set())
            newly_entered = current_ids - previous_ids
            self._in_range[state.agent_id] = current_ids

            if not newly_entered or retriever is None:
                continue

            loc = state.position.location_id or "campus"
            crowd = len(neighbors)
            for n in neighbors:
                if n.agent_id not in newly_entered:
                    continue
                doing = ""
                if n.manager and n.manager.current_action:
                    doing = f" ({n.manager.current_action.description})"
                text = (
                    f"At {hhmm} near {loc}, saw {n.persona_name}{doing}"
                    + (f"; {crowd} people around." if crowd > 1 else ".")
                )
                try:
                    retriever.store(state.persona_name, text, kind="observation",
                                    importance=0.5, date_str=date_str)
                except Exception as e:
                    logger.debug("[WorldEngine] observation store failed: %s", e)

    def _conversation_request(
        self,
        a: AgentRuntimeState,
        b: AgentRuntimeState,
        loc_id: str,
        tick: int,
        hhmm: str,
    ) -> Dict[str, Any]:
        """Capture the inputs needed to restart an in-flight LLM request."""
        date_ctx = date_from_simulation_time(f"{self.sim_start_date} {hhmm}")
        memories_a, _ = self._memory_context(
            a.persona_name, a.persona, before_date=date_ctx,
            query_hint=f"{b.persona_name} {loc_id}",
        )
        memories_b, _ = self._memory_context(
            b.persona_name, b.persona, before_date=date_ctx,
            query_hint=f"{a.persona_name} {loc_id}",
        )
        return {
            "agent_a_id": a.agent_id,
            "agent_b_id": b.agent_id,
            "location_id": loc_id,
            "start_tick": tick,
            "hhmm": hhmm,
            "action_a": a.manager.current_action.description if a.manager and a.manager.current_action else "unknown",
            "action_b": b.manager.current_action.description if b.manager and b.manager.current_action else "unknown",
            "relationship_a_to_b": self.relationship_matrix.get(a.agent_id, b.agent_id),
            "relationship_b_to_a": self.relationship_matrix.get(b.agent_id, a.agent_id),
            "relationship_context_a": self.relationship_matrix.context(a.agent_id, b.agent_id).model_dump(),
            "relationship_context_b": self.relationship_matrix.context(b.agent_id, a.agent_id).model_dump(),
            "memories_a": memories_a,
            "memories_b": memories_b,
            "energy_a": a.energy_level,
            "emotion_a": a.emotion_state,
            "energy_b": b.energy_level,
            "emotion_b": b.emotion_state,
        }

    @staticmethod
    def _relationship_context_text(record: Dict[str, Any]) -> str:
        """Compact structured relationship context for a conversation prompt."""
        if not isinstance(record, dict):
            return ""
        tags = ", ".join(str(tag) for tag in record.get("tags", [])[:4])
        context = str(record.get("context", "")).strip()
        return "; ".join(part for part in (tags, context) if part)

    def _start_conversation_task(
        self,
        a: AgentRuntimeState,
        b: AgentRuntimeState,
        request: Dict[str, Any],
    ) -> None:
        """Run one durable conversation request and clear its bookkeeping."""
        key = self._conversation_key(a.agent_id, b.agent_id)
        if key in self._conversation_tasks:
            return

        async def _runner() -> None:
            try:
                await self._run_conversation_pipeline(a, b, request)
            except asyncio.CancelledError:
                raise
            except Exception:
                logger.exception(
                    "[WorldEngine] conversation pipeline crashed for '%s' <-> '%s'",
                    a.persona_name, b.persona_name,
                )
                self._end_conversation(a)
                self._end_conversation(b)
            finally:
                self._pending_conversations.pop(key, None)
                self._conversation_tasks.pop(key, None)

        self._conversation_tasks[key] = asyncio.create_task(_runner())

    def _discard_pending_conversation(self, first_id: str, second_id: str) -> None:
        """Prevent a timed-out conversation from being restarted on restore."""
        self._pending_conversations.pop(self._conversation_key(first_id, second_id), None)

    async def _check_conversations(self, tick: int, hhmm: str) -> None:
        """
        Detect close one-to-one pairs and fire a background LLM conversation.
        Agents may meet while travelling: the action manager pauses their
        routes and resumes them from the exact path point when the chat ends.
        The pair-only rule, cooldown, per-day cap, and blocked-action (e.g.
        sleeping) guard still apply.
        Triggered pairs are paused immediately and resume when the background
        task completes.
        """
        radius = float(_cfg.CONVERSATION_RADIUS_PX)

        def is_conversation_candidate(state: AgentRuntimeState) -> bool:
            return (
                not state.paused
                and not (state.manager and state.manager.is_last_action)
                and bool(state.position.location_id)
                and state.energy_level >= _cfg.CONVERSATION_MIN_ENERGY
                and state.emotion_state >= _cfg.CONVERSATION_MIN_EMOTION
            )

        candidates = [s for s in self.registry.all_states() if is_conversation_candidate(s)]
        candidate_ids = {state.agent_id for state in candidates}
        busy: set = set()  # agents already committed to a conversation this tick

        for a in candidates:
            if a.agent_id in busy:
                continue
            neighbors = [
                n for n in self._neighbors_within_px(a, radius)
                if n.agent_id in candidate_ids
                and n.agent_id not in busy
            ]
            if not neighbors:
                continue
            # Pair-only rule: a clean 1:1 encounter. More than one other agent
            # inside the circle is a crowd — skip (mirrors the old >2 rule).
            if len(neighbors) > 1:
                logger.debug(
                    "[WorldEngine] %d agents within %.0fpx of '%s' — skipping (pair-only rule)",
                    len(neighbors) + 1, radius, a.persona_name,
                )
                continue
            b = neighbors[0]
            # Require a symmetric clean pair: b's circle must also contain only a.
            b_neighbors = [
                n for n in self._neighbors_within_px(b, radius)
                if n.agent_id in candidate_ids
            ]
            if len(b_neighbors) > 1:
                continue

            # Cooldown check
            if not self.world.can_converse(a.agent_id, b.agent_id):
                continue
            # No back-to-back repeats: they can't talk again until at least one
            # of them has spoken with someone else since (avoids the two of them
            # looping conversations while standing together).
            if a.last_conversation_partner == b.agent_id or b.last_conversation_partner == a.agent_id:
                continue
            # Max conversations cap
            if a.conversation_count >= MAX_CONVERSATIONS_PER_AGENT:
                continue
            if b.conversation_count >= MAX_CONVERSATIONS_PER_AGENT:
                continue
            # Both must be in compatible (non-blocked, e.g. not sleeping) actions
            if a.manager and a.manager.current_action:
                if _is_action_blocked(CurrentAction(
                    description=a.manager.current_action.description,
                    start_tick=tick, end_tick=tick + 10,
                )):
                    continue
            if b.manager and b.manager.current_action:
                if _is_action_blocked(CurrentAction(
                    description=b.manager.current_action.description,
                    start_tick=tick, end_tick=tick + 10,
                )):
                    continue

            # A travelling pair may still carry different source-location ids.
            # Label the dialogue honestly instead of claiming they are already
            # inside either endpoint building.
            a_is_moving = bool(a.manager and a.manager.current_action and a.manager.current_action.action_type == ActionType.MOVE)
            b_is_moving = bool(b.manager and b.manager.current_action and b.manager.current_action.action_type == ActionType.MOVE)
            loc_id = (
                "campus_path"
                if a_is_moving or b_is_moving or a.position.location_id != b.position.location_id
                else a.position.location_id
            ) or "campus"

            logger.info(
                "[WorldEngine] triggering conversation: '%s' <-> '%s' near %s (within %.0fpx)",
                a.persona_name, b.persona_name, loc_id, radius,
            )

            # Capture the original request before replacing the actions with a
            # visible conversation state.  This payload is checkpointed while
            # the provider call is in flight.
            request = self._conversation_request(a, b, loc_id, tick, hhmm)

            # Immediately set agents to conversation mode (frozen until bg task completes)
            if a.manager:
                a.manager.set_conversation_action(b.persona_name, tick)
                a.current_action = a.manager.current_action.model_dump()
                a.paused = True
                a.conversation_count += 1
                a.last_conversation_partner = b.agent_id
                a.conversation_start_tick = tick
                a.active_conversation = {
                    "partner_name": b.persona_name,
                    "partner_id": b.agent_id,
                    "location_id": loc_id,
                    "started_tick": tick,
                    "status": "generating",
                }
            if b.manager:
                b.manager.set_conversation_action(a.persona_name, tick)
                b.current_action = b.manager.current_action.model_dump()
                b.paused = True
                b.conversation_count += 1
                b.last_conversation_partner = a.agent_id
                b.conversation_start_tick = tick
                b.active_conversation = {
                    "partner_name": a.persona_name,
                    "partner_id": a.agent_id,
                    "location_id": loc_id,
                    "started_tick": tick,
                    "status": "generating",
                }

            self._pending_conversations[self._conversation_key(a.agent_id, b.agent_id)] = request
            self._start_conversation_task(a, b, request)
            busy.add(a.agent_id)
            busy.add(b.agent_id)

            logger.info(
                "[WorldEngine] conversation '%s' <-> '%s' started (async, waiting for LLM)",
                a.persona_name, b.persona_name,
            )

    async def _run_conversation_pipeline(
        self,
        a: AgentRuntimeState,
        b: AgentRuntimeState,
        request: Dict[str, Any],
    ) -> None:
        """
        Background task: generates the conversation via LLM, saves results,
        and resumes both agents with new remaining-day plans.
        Runs in a thread pool so it doesn't block the event loop.
        """
        loop = asyncio.get_event_loop()

        loc_id = request["location_id"]
        tick = request["start_tick"]
        hhmm = request["hhmm"]
        a_action = CurrentAction(
            description=request["action_a"],
            start_tick=tick,
            end_tick=tick + 10,
            target_location_id=loc_id,
        )
        b_action = CurrentAction(
            description=request["action_b"],
            start_tick=tick,
            end_tick=tick + 10,
            target_location_id=loc_id,
        )
        # Step 1: Generate conversation (blocking LLM call → thread pool)
        conv_result = await loop.run_in_executor(
            None,
            lambda: generate_conversation(
                a.agent_id, b.agent_id,
                a.persona, b.persona,
                a.day_plan, b.day_plan,
                a_action, b_action,
                request["relationship_a_to_b"], request["relationship_b_to_a"],
                loc_id, hhmm,
                request["memories_a"], request["memories_b"],
                energy_a=request["energy_a"], emotion_a=request["emotion_a"],
                energy_b=request["energy_b"], emotion_b=request["emotion_b"],
                relationship_context_a=self._relationship_context_text(request.get("relationship_context_a", {})),
                relationship_context_b=self._relationship_context_text(request.get("relationship_context_b", {})),
            ),
        )

        if conv_result is None:
            logger.warning(
                "[WorldEngine] conversation LLM failed for '%s' <-> '%s' — unpausing",
                a.persona_name, b.persona_name,
            )
            self._end_conversation(a)
            self._end_conversation(b)
            return

        # If the conversation was already ended by timeout while the LLM was
        # running, discard the result and return silently.
        if not a.paused or not b.paused:
            logger.info(
                "[WorldEngine] conversation '%s' <-> '%s' ended by timeout while LLM ran — discarding result",
                a.persona_name, b.persona_name,
            )
            return

        # Step 2: Save conversation to Short_term
        date_str = date_from_simulation_time(f"{self.sim_start_date} {hhmm}")
        conv_entry = {
            "participants": [a.persona_name, b.persona_name],
            "messages": [
                {"speaker": m.speaker, "text": m.text}
                for m in conv_result.messages
            ],
            "summary": conv_result.summary,
        }
        append_conversation(a.persona_name, date_str, conv_entry)
        append_conversation(b.persona_name, date_str, conv_entry)

        # Add to the rolling UI feed (last N conversations across the campus).
        self._recent_convs.appendleft({
            "time": hhmm,
            "participants": [a.persona_name, b.persona_name],
            "summary": conv_result.summary,
            "sentiment": conv_result.sentiment,
            "location": loc_id,
        })

        # Step 3: Store conversation messages on agent states (for frontend snapshot)
        messages = conv_entry["messages"]
        a.active_conversation = {
            "partner_name": b.persona_name,
            "partner_id": b.agent_id,
            "messages": messages,
            "duration_minutes": conv_result.duration_minutes,
            "started_tick": self.world.tick,
            "ends_at_tick": self.world.tick + conv_result.duration_minutes,
            "status": "active",
        }
        b.active_conversation = {
            "partner_name": a.persona_name,
            "partner_id": a.agent_id,
            "messages": messages,
            "duration_minutes": conv_result.duration_minutes,
            "started_tick": self.world.tick,
            "ends_at_tick": self.world.tick + conv_result.duration_minutes,
            "status": "active",
        }
        # The action placeholder is intentionally open-ended while the LLM is
        # generating. Once a duration exists, make the UI/action snapshot tell
        # the same truth as ``ends_at_tick`` instead of displaying 23:59.
        conversation_end = self._minutes_to_hhmm((self.world.tick + conv_result.duration_minutes) % (24 * 60))
        for state in (a, b):
            if state.manager and state.manager.current_action:
                state.manager.current_action.end_time = conversation_end
                state.current_action = state.manager.current_action.model_dump()

        # Step 4: Update relationship matrix
        self.relationship_matrix.update(a.agent_id, b.agent_id, conv_result.relationship_delta)
        self.relationship_matrix.update(b.agent_id, a.agent_id, conv_result.relationship_delta)
        self.relationship_matrix.save()
        # Conversations affect the people having them, not only their stored
        # relationship score.  The LLM decides each participant's net energy
        # and mood change for the chat; only a 0..1 safety clamp is applied.
        for state, energy_delta, emotion_delta in (
            (a, conv_result.energy_delta_a, conv_result.emotion_delta_a),
            (b, conv_result.energy_delta_b, conv_result.emotion_delta_b),
        ):
            state.energy_level = max(0.0, min(1.0, state.energy_level + float(energy_delta or 0.0)))
            state.emotion_state = max(0.0, min(1.0, state.emotion_state + float(emotion_delta or 0.0)))
        logger.info(
            "[WorldEngine] conversation '%s' <-> '%s' active until tick %d",
            a.persona_name, b.persona_name, self.world.tick + conv_result.duration_minutes,
        )
        return

        # Step 5: Generate remaining-day plans (blocking → thread pool)
        from src.agents.day_planner import run as day_planner_run

        def _run_planner(state: AgentRuntimeState) -> list:
            mem, yday = self._memory_context(
                state.persona_name, state.persona,
                before_date=date_from_simulation_time(f"{self.sim_start_date} {hhmm}"),
            )
            proxy = SimpleNamespace(
                persona=state.persona,
                relevant_memories=mem,
                yesterday_summary=yday,
            )
            plan_result = day_planner_run(
                proxy,
                {
                    "current_time": f"{self.sim_start_date} {hhmm}",
                    "places": None,
                    "persona_name": state.persona_name,
                    "mode": "remaining",
                    "current_location_id": state.position.location_id,
                    "energy_level": state.energy_level,
                    "emotion_state": state.emotion_state,
                },
            )
            return plan_result.get("day_plan", [])

        # Step 5: Resume both agents. Replanning is disabled by default — kept
        # behind a flag for future re-enablement.
        _ENABLE_POST_CONVERSATION_REPLAN = False

        for state in (a, b):
            try:
                if _ENABLE_POST_CONVERSATION_REPLAN:
                    should_replan = bool(getattr(conv_result, "should_replan", False))
                    try:
                        from src.core.budget import GOVERNOR
                    except Exception:
                        GOVERNOR = None
                    allow_replan = (
                        should_replan
                        and state.replan_count < _cfg.MAX_REPLANS_PER_AGENT_PER_DAY
                        and (GOVERNOR is None or GOVERNOR.can_afford("replan", cost=4))
                    )
                    if allow_replan:
                        new_plan = await loop.run_in_executor(None, _run_planner, state)
                        if new_plan and state.manager:
                            state.manager.resume_from_conversation(new_plan)
                            state.day_plan = new_plan
                            state.replan_count += 1
                            logger.info(
                                "[WorldEngine] post-conversation replan for '%s' (%d actions): %s",
                                state.persona_name, len(new_plan),
                                getattr(conv_result, "plan_change", "") or "",
                            )
                        elif state.manager:
                            state.manager.resume_from_conversation(state.day_plan)
                    else:
                        if state.manager:
                            state.manager.resume_from_conversation(state.day_plan)
                else:
                    # No replan — resume the day already planned (0 LLM).
                    if state.manager:
                        state.manager.resume_from_conversation(state.day_plan)
                state.active_conversation = None
            except Exception as e:
                logger.error(
                    "[WorldEngine] resume after conversation failed for '%s': %s",
                    state.persona_name, e,
                )
                if state.manager:
                    try:
                        state.manager.resume_from_conversation(state.day_plan)
                    except Exception:
                        pass

        # Step 6: Unpause agents
        a.paused = False
        b.paused = False

        logger.info(
            "[WorldEngine] conversation '%s' <-> '%s' complete — agents resumed",
            a.persona_name, b.persona_name,
        )

    async def _check_last_action_triggers(self, tick: int, hhmm: str) -> None:
        """
        When an agent enters their last action of the day plan:
        Archive durable day memories to Qdrant.
        """
        for state in self.registry.all_states():
            manager = state.manager
            if manager is None:
                continue
            if not manager._entered_last_action:
                continue
            # Already triggered? Use a flag on the registry state
            if state.day_archived:
                continue

            logger.info(
                "[WorldEngine] '%s' entered last action — archiving day",
                state.persona_name,
            )

            # Archive current day
            date_str = date_from_simulation_time(f"{self.sim_start_date} {hhmm}")
            try:
                archive_result = await archive_to_long_term(state.persona_name, date_str)
                logger.info(
                    "[WorldEngine]   archive: %s", archive_result.get("summary", "")[:80],
                )
                state.day_archived = True
            except Exception as e:
                logger.error("[WorldEngine]   archive failed for '%s': %s", state.persona_name, e)

    def _sync_to_world_state(self) -> None:
        """Mirror the agent registry into WorldState for frontend queries."""
        for state in self.registry.all_states():
            agent_id = state.agent_id
            try:
                ws_agent = self.world.get_agent(agent_id)
            except KeyError:
                continue

            # Update position
            if state.position != ws_agent.position:
                self.world.move_agent(agent_id, state.position)

            # Update current action
            if state.manager and state.manager.current_action:
                action = state.manager.current_action
                if state.active_conversation:
                    start_tick = state.active_conversation.get(
                        "started_tick", state.conversation_start_tick,
                    )
                    end_tick = state.active_conversation.get("ends_at_tick", start_tick + 30)
                else:
                    day_start_tick = self.world.tick - (self.world.tick % (24 * 60))
                    start_minute = self._hhmm_to_minutes(action.start_time)
                    end_minute = self._hhmm_to_minutes(action.end_time)
                    start_tick = day_start_tick + start_minute
                    end_tick = day_start_tick + end_minute
                    if end_minute <= start_minute:
                        end_tick += 24 * 60
                ca = CurrentAction(
                    description=action.description,
                    start_tick=start_tick,
                    end_tick=end_tick,
                    target_location_id=action.location_id or state.position.location_id,
                )
                self.world.set_agent_action(agent_id, ca)
            else:
                self.world.clear_agent_action(agent_id)

    async def _drain_conversations_for_handoff(self, timeout_seconds: float) -> int:
        """Give active conversations a bounded chance to persist their results."""
        tasks = list(self._conversation_tasks.values())
        if not tasks:
            return 0

        logger.info(
            "[WorldEngine] day handoff: waiting up to %.0fs for %d conversation(s)",
            timeout_seconds, len(tasks),
        )
        _done, pending = await asyncio.wait(tasks, timeout=max(0.0, timeout_seconds))
        if not pending:
            return 0

        logger.warning(
            "[WorldEngine] day handoff timed out with %d conversation(s); cancelling them",
            len(pending),
        )
        timed_out_requests = list(self._pending_conversations.values())
        for task in pending:
            task.cancel()
        await asyncio.gather(*pending, return_exceptions=True)

        for request in timed_out_requests:
            try:
                self._end_conversation(self.registry.get(request["agent_a_id"]))
                self._end_conversation(self.registry.get(request["agent_b_id"]))
            except KeyError:
                pass
            self._pending_conversations.pop(
                self._conversation_key(request["agent_a_id"], request["agent_b_id"]), None,
            )
        return len(pending)

    async def _discard_pending_decisions_for_handoff(self) -> int:
        """Prevent a prior day's advisory decision from affecting a new day.

        Decision tasks deliberately run in the background so provider latency
        cannot stall a tick.  Their observation and plan inputs belong to one
        simulated day, however, so a result that arrives after midnight must
        never be consumed as a next-day replan.
        """
        tasks = list(self._decision_tasks.values())
        if not tasks:
            return 0

        pending = [task for task in tasks if not task.done()]
        for task in pending:
            task.cancel()
        # Consume every outcome before releasing the references.  Cancelling a
        # to_thread wrapper cannot stop its SDK call, but it does ensure its
        # eventual result cannot mutate this engine on the following day.
        await asyncio.gather(*tasks, return_exceptions=True)
        self._decision_tasks.clear()
        logger.info(
            "[WorldEngine] discarded %d prior-day decision task(s) at handoff",
            len(tasks),
        )
        return len(tasks)

    async def handoff_to_next_day(
        self,
        next_date: str,
        conversation_timeout_seconds: Optional[float] = None,
    ) -> Dict[str, Any]:
        """Finish one calendar day without replacing the engine or agents.

        Conversation output is allowed to settle first, then every agent's
        completed day is compressed into long-term memory.  New plans are
        generated with the ending action, location, and wellbeing as explicit
        continuity context; positions, brains, relationships, and action
        managers remain owned by this same engine instance.
        """
        old_date = self.sim_start_date
        timeout = (
            _cfg.DAY_HANDOFF_CONVERSATION_TIMEOUT_SECONDS
            if conversation_timeout_seconds is None else conversation_timeout_seconds
        )
        cancelled = await self._drain_conversations_for_handoff(timeout)
        await self._discard_pending_decisions_for_handoff()

        # Preserve the full replay before resetting the per-day action log.
        # History/checkpoint storage is observability, not a reason to stop a
        # living simulation if a file is temporarily unavailable.
        handoff_warnings: list[str] = []
        try:
            save_history(self.world, old_date)
        except Exception as exc:
            logger.exception("[WorldEngine] unable to save replay for %s", old_date)
            handoff_warnings.append(f"history save failed: {exc}")

        async def _archive_and_clear(state: AgentRuntimeState) -> bool:
            try:
                # Re-index even when we made the early final-activity archive:
                # conversations and actions completed afterward must be part of
                # the authoritative end-of-day Qdrant snapshot.
                await archive_to_long_term(state.persona_name, old_date)
                state.day_archived = True
                clear_short_term_data(state.persona_name, old_date)
                return True
            except Exception as exc:
                logger.error(
                    "[WorldEngine] final archive failed for '%s': %s",
                    state.persona_name, exc,
                )
                return False

        archive_results = await asyncio.gather(
            *[_archive_and_clear(state) for state in self.registry.all_states()],
            return_exceptions=False,
        )

        from src.agents.day_planner import run as day_planner_run

        def _handoff_context(state: AgentRuntimeState) -> str:
            action = state.manager.current_action if state.manager else None
            action_text = action.description if action else "no active action"
            location = state.position.location_id or "their current campus position"
            return (
                f"The previous day ended while the agent was {action_text} at {location}. "
                f"Energy is {state.energy_level:.2f}/1.0 and emotion is "
                f"{state.emotion_state:.2f}/1.0. Continue naturally from this "
                "physical and emotional state; do not abruptly relocate them. "
                "When assigning energy_change/emotion_change for the new day, "
                "keep the cumulative energy and mood totals between 0.0 and "
                "1.0 at all times."
            )

        async def _plan_next_day(state: AgentRuntimeState) -> tuple[AgentRuntimeState, list]:
            memories, yesterday = self._memory_context(
                state.persona_name, state.persona, before_date=next_date,
            )
            proxy = SimpleNamespace(
                persona=state.persona,
                relevant_memories=memories,
                yesterday_summary=yesterday,
            )
            try:
                result = await asyncio.to_thread(
                    day_planner_run,
                    proxy,
                    {
                        "current_time": f"{next_date} 00:00",
                        "places": None,
                        "persona_name": state.persona_name,
                        "mode": "next_day",
                        "current_location_id": state.position.location_id,
                        "handoff_context": _handoff_context(state),
                        "energy_level": state.energy_level,
                        "emotion_state": state.emotion_state,
                        "upcoming_events": self.event_manager.snapshot(next_date, "00:00").get("upcoming", []),
                    },
                )
                plan = result.get("day_plan", [])
                if plan:
                    return state, plan
            except Exception as exc:
                logger.error(
                    "[WorldEngine] next-day plan failed for '%s': %s",
                    state.persona_name, exc,
                )
            # A failed planner must not strand the agent. Reusing the previous
            # schedule is the safest fallback because it preserves continuity.
            return state, state.day_plan

        # Keep day-handoff planner traffic serial.  A single key ring is shared
        # across agents, and concurrent full pipelines caused a burst of
        # provider failures precisely when a new day was being installed.
        planned = []
        for state in self.registry.all_states():
            planned.append(await _plan_next_day(state))

        self.sim_start_date = next_date
        self.sim_start_hhmm = "00:00"
        self._day_index += 1
        self._recent_convs.clear()
        self._in_range.clear()
        self._last_decision_tick.clear()
        self._downtime_replan_tick.clear()
        self._last_obs.clear()
        self._tick_observations.clear()
        self._applied_event_effects.clear()
        self.world.history.clear()

        for state, plan in planned:
            state.day_plan = plan
            state.day_archived = False
            state.conversation_count = 0
            state.replan_count = 0
            state.last_conversation_partner = None
            state.conversation_start_tick = 0
            state.active_conversation = None
            state.paused = False
            if state.manager:
                state.manager.begin_new_day(plan)
                action = state.manager.tick(self.world.tick)
                state.position = state.manager.position
                state.current_action = action.model_dump() if action else None

        try:
            self._apply_event_opportunities(next_date)
        except Exception as exc:
            logger.exception("[WorldEngine] unable to apply events for %s", next_date)
            handoff_warnings.append(f"event overlay failed: {exc}")

        self._sync_to_world_state()
        try:
            save_checkpoint(
                self.world,
                self.registry,
                self.world.tick,
                engine_state=self.checkpoint_state(),
            )
            prune_checkpoints(keep_last=KEEP_LAST)
        except Exception as exc:
            logger.exception("[WorldEngine] unable to checkpoint handoff at tick %d", self.world.tick)
            handoff_warnings.append(f"handoff checkpoint failed: {exc}")

        snapshot = self._frontend_snapshot(self.world.tick, "00:00")
        snapshot.update({
            "type": "day_initialized",
            "phase": "initialized",
            "date": next_date,
            "cancelled_conversations": cancelled,
            "archived_agents": sum(bool(result) for result in archive_results),
            "warnings": handoff_warnings,
            "recent_conversations": [],
        })
        return snapshot

    # ------------------------------------------------------------------ #
    # Main run loop
    # ------------------------------------------------------------------ #

    async def run(
        self,
        max_tick: int = 1440,
        on_tick: Optional[callable] = None,
        tick_speed: Optional[float] = None,
        tick_lock: Optional[asyncio.Lock] = None,
    ) -> None:
        """Run simulation until world.tick reaches max_tick (absolute).

        If ``on_tick`` is provided, it is called with each tick's snapshot dict
        (after the tick is processed). This is used by Odin.py to broadcast
        state to WebSocket clients.

        ``tick_speed`` is a speed multiplier (1.0 = configured real-time, 2.0 =
        double speed). When None, the value from config (env/CLI) is used.
        """
        dynamic_speed = tick_speed is None
        initial_speed = _cfg.TICK_SPEED if dynamic_speed else tick_speed
        effective_sleep = _cfg.REAL_SECONDS_PER_TICK / max(initial_speed, 1e-9)
        logger.info("[WorldEngine] ========== STARTING SIMULATION ==========")
        logger.info(
            "[WorldEngine] %d agents, max tick %d (current=%d), %.1f real sec/tick (speed=%.1fx)",
            len(self.registry), max_tick, self.world.tick, effective_sleep, initial_speed,
        )

        while self.world.tick < max_tick:
            if tick_lock is None:
                result = await self.run_tick()
            else:
                async with tick_lock:
                    result = await self.run_tick()
            if not result:
                logger.info("[WorldEngine] simulation complete at tick %d", self.world.tick)
                break

            # Invoke callback if provided
            if on_tick is not None:
                try:
                    await on_tick(result)
                except Exception:
                    logger.exception("[WorldEngine] on_tick callback failed; simulation will continue")

            # Log periodic status
            elapsed = self.world.tick
            if elapsed % 60 == 0:
                active = sum(
                    1 for s in self.registry.all_states()
                    if s.current_action is not None
                )
                paused = sum(1 for s in self.registry.all_states() if s.paused)
                hhmm = self._minutes_to_hhmm(self.world.tick % (24 * 60))
                logger.info(
                    "[WorldEngine] tick=%d time=%s agents=%d active=%d paused=%d",
                    elapsed, hhmm, len(self.registry), active, paused,
                )

            # Read the mutable runtime speed on every iteration so UI controls
            # take effect without restarting a day-long run.
            active_speed = _cfg.TICK_SPEED if dynamic_speed else tick_speed
            await asyncio.sleep(_cfg.REAL_SECONDS_PER_TICK / max(active_speed, 1e-9))

        logger.info("[WorldEngine] ========== SIMULATION ENDED ==========")
        if on_tick is not None:
            try:
                await on_tick({"type": "simulation_ended", "tick": self.world.tick})
            except Exception:
                logger.exception("[WorldEngine] final on_tick callback failed")

    # ------------------------------------------------------------------ #
    # Helpers
    # ------------------------------------------------------------------ #

    @staticmethod
    def _minutes_to_hhmm(minutes: int) -> str:
        h, m = divmod(minutes, 60)
        return f"{h:02d}:{m:02d}"

    @staticmethod
    def _hhmm_to_minutes(hhmm: str) -> int:
        h, m = hhmm.split(":")
        return int(h) * 60 + int(m)


# ------------------------------------------------------------------ #
# Standalone CLI entry point
# ------------------------------------------------------------------ #

if __name__ == "__main__":
    import argparse

    from src.core.log import setup_logging
    setup_logging(run_id="world_engine", console=True)

    parser = argparse.ArgumentParser(description="WorldEngine — multi-agent simulation")
    parser.add_argument(
        "--days", type=int, default=1,
        help="Number of simulation days to run (default: 1)",
    )
    parser.add_argument(
        "--start-date", default=_cfg.SIM_START_DATE,
        help="Simulation start date (default: from SIM_START_DATE / 2026-07-03)",
    )
    parser.add_argument(
        "--start-time", default="00:00",
        help="Simulation start time HH:MM (default: 00:00)",
    )
    parser.add_argument(
        "--resume", nargs="?", const=None, default=False,
        help="Resume from checkpoint. Pass a tick number, or no arg to prompt.",
    )
    # Register all simulation toggles (--tick-speed, --reflex-llm,
    # --perception-radius-px, --real-seconds-per-sim-minute, etc.)
    _cfg.add_cli_arguments(parser)
    args = parser.parse_args()

    # Apply CLI overrides onto config (CLI > .env > built-in defaults).
    _cfg.apply_overrides(**_cfg.overrides_from_args(args))
    logger.info("[WorldEngine] active settings:\n%s", _cfg.describe_settings())

    async def _main() -> None:
        if args.resume is not False:
            ticks = list_checkpoints()
            if not ticks:
                logger.error("[WorldEngine] no checkpoints found — cannot resume")
                return

            if args.resume is None:
                # --resume with no value → prompt
                print(f"\nAvailable checkpoints ({len(ticks)} total, showing last 20):")
                for t in ticks[-20:]:
                    print(f"  {t:>5}")
                print()

                while True:
                    tick_str = input("Resume from which tick? (or 'new' to start fresh): ").strip()
                    if tick_str.lower() == "new":
                        break
                    try:
                        resume_tick = int(tick_str)
                        if resume_tick in ticks:
                            break
                        print(f"Tick {resume_tick} not found. Try again.")
                    except ValueError:
                        print("Invalid input. Enter a tick number or 'new'.")

                if tick_str.lower() == "new":
                    pass  # fall through to fresh start
                else:
                    engine = WorldEngine(
                        sim_start_date=args.start_date,
                        sim_start_hhmm=args.start_time,
                    )
                    world, registry, checkpoint_state = load_checkpoint(
                        resume_tick, engine.resolver, return_metadata=True,
                    )
                    engine.world = world
                    engine.registry = registry
                    engine.restore_checkpoint_state(checkpoint_state)
                    await engine.resume_pending_conversations()
                    logger.info(
                        "[WorldEngine] resumed from tick %d — %d agents",
                        resume_tick, len(registry),
                    )
                    day_end_tick = ((engine.world.tick // (24 * 60)) + 1) * (24 * 60)
                    await engine.run(max_tick=day_end_tick)
                    save_history(engine.world, engine.sim_start_date)
                    return
            else:
                # --resume <tick>
                resume_tick = int(args.resume)
                if resume_tick not in ticks:
                    logger.error("[WorldEngine] tick %d not found in checkpoints", resume_tick)
                    return
                engine = WorldEngine(
                    sim_start_date=args.start_date,
                    sim_start_hhmm=args.start_time,
                )
                world, registry, checkpoint_state = load_checkpoint(
                    resume_tick, engine.resolver, return_metadata=True,
                )
                engine.world = world
                engine.registry = registry
                engine.restore_checkpoint_state(checkpoint_state)
                await engine.resume_pending_conversations()
                logger.info(
                    "[WorldEngine] resumed from tick %d — %d agents",
                    resume_tick, len(registry),
                )
                day_end_tick = ((engine.world.tick // (24 * 60)) + 1) * (24 * 60)
                await engine.run(max_tick=day_end_tick)
                save_history(engine.world, engine.sim_start_date)
                return

        # Fresh start
        engine = WorldEngine(
            sim_start_date=args.start_date,
            sim_start_hhmm=args.start_time,
        )
        await engine.initialize()
        for day in range(args.days):
            logger.info("[WorldEngine] --- Day %d ---", day + 1)
            day_end_tick = ((engine.world.tick // (24 * 60)) + 1) * (24 * 60)
            await engine.run(max_tick=day_end_tick)
            if day < args.days - 1:
                next_date = (
                    datetime.fromisoformat(engine.sim_start_date) + timedelta(days=1)
                ).strftime("%Y-%m-%d")
                await engine.handoff_to_next_day(next_date)
            else:
                save_history(engine.world, engine.sim_start_date)

    asyncio.run(_main())