from __future__ import annotations

from datetime import datetime, timedelta, timezone

from device_gateway.integrations.pi import (
    DEVICE_HISTORY_MAX_BYTES,
    SESSION_HISTORY_MAX_BYTES,
    PiSessionEvent,
    PiSessionRegistry,
)


def event(name: str, when: datetime, *, failed: bool = False, session_name: str = "esp32-dev") -> PiSessionEvent:
    return PiSessionEvent(
        event=name,  # type: ignore[arg-type]
        session_id="pi-session-1",
        name=session_name,
        cwd="/home/lighthouse/esp32-dev",
        failed=failed,
        timestamp=when,
    )


def test_pi_session_start_run_and_done() -> None:
    registry = PiSessionRegistry()
    started = datetime(2026, 10, 2, 8, 0, tzinfo=timezone.utc)
    assert registry.apply(event("session_start", started)).state == "IDLE"
    assert registry.apply(event("agent_start", started + timedelta(seconds=5))).state == "RUNNING"
    done = registry.apply(event("agent_end", started + timedelta(seconds=36)))
    assert done.state == "DONE"
    assert done.public()["elapsed_seconds"] == 31


def test_pi_session_run_timer_survives_low_level_agent_restarts() -> None:
    """Retries, compaction and steering reopen a low-level run inside one user turn.

    The Watch Running timer is anchored to agent_start, so those extra events must
    not restart it at 00:00.
    """
    registry = PiSessionRegistry()
    started = datetime(2026, 10, 2, 8, 0, tzinfo=timezone.utc)

    def lifecycle(name: str, offset: int, turn_id: str) -> PiSessionEvent:
        return PiSessionEvent(
            event=name,  # type: ignore[arg-type]
            session_id="pi-session-1",
            name="esp32-dev",
            cwd="/home/lighthouse/esp32-dev",
            turn_id=turn_id,
            timestamp=started + timedelta(seconds=offset),
        )

    registry.apply(event("session_start", started))
    registry.apply(lifecycle("agent_start", 1, "turn-1"))
    # Same user turn: overflow recovery then compaction retry, each a new low-level run.
    registry.apply(lifecycle("agent_start", 40, "turn-2"))
    session = registry.apply(lifecycle("agent_start", 95, "turn-3"))
    assert session.state == "RUNNING"
    assert session.turn_id == "turn-3"
    assert session.public(now=started + timedelta(seconds=95))["elapsed_seconds"] == 94
    assert session.public(now=started + timedelta(seconds=126))["elapsed_seconds"] == 125

    # A genuinely new user turn after the terminal event re-anchors the timer.
    registry.apply(lifecycle("agent_end", 130, "turn-3"))
    restarted = registry.apply(lifecycle("agent_start", 200, "turn-4"))
    assert restarted.public(now=started + timedelta(seconds=200))["elapsed_seconds"] == 0
    assert restarted.public(now=started + timedelta(seconds=215))["elapsed_seconds"] == 15


def test_pi_session_rename_failure_and_shutdown() -> None:
    registry = PiSessionRegistry()
    started = datetime(2026, 10, 2, 8, 0, tzinfo=timezone.utc)
    registry.apply(event("session_start", started))
    renamed = registry.apply(
        event("session_info_changed", started + timedelta(seconds=1), session_name="pager-work")
    )
    assert renamed.name == "pager-work"
    registry.apply(event("agent_start", started + timedelta(seconds=2)))
    assert registry.apply(event("agent_end", started + timedelta(seconds=3), failed=True)).state == "FAILED"
    assert registry.apply(event("session_shutdown", started + timedelta(seconds=4))).state == "OFFLINE"


def test_pi_activity_keeps_utf8_safe_bounded_history_and_tool_preview() -> None:
    registry = PiSessionRegistry()
    started = datetime(2026, 10, 2, 8, 0, tzinfo=timezone.utc)
    registry.apply(event("session_start", started))
    for offset in range(20):
        registry.apply(
            PiSessionEvent(
                event="activity",
                session_id="pi-session-1",
                name="esp32-dev",
                cwd="/home/lighthouse/esp32-dev",
                activity_type="text",
                activity=("实时输出🙂" * 100) + str(offset),
                timestamp=started + timedelta(seconds=offset + 1),
            )
        )
    session = registry.get("pi-session-1")
    assert session is not None
    assert len(session.recent_text.encode("utf-8")) <= SESSION_HISTORY_MAX_BYTES
    assert session.recent_text.endswith("19")
    assert len(str(session.public()["recent_text"]).encode("utf-8")) <= 480
    assert len(str(session.public(include_history=True)["recent_text"]).encode("utf-8")) <= DEVICE_HISTORY_MAX_BYTES

    registry.apply(
        PiSessionEvent(
            event="activity",
            session_id="pi-session-1",
            name="esp32-dev",
            cwd="/home/lighthouse/esp32-dev",
            activity_type="tool",
            activity="Reading\nserver/device_gateway/app.py",
            tool_name="read",
            timestamp=started + timedelta(seconds=30),
        )
    )
    assert session.activity_type == "tool"
    assert session.latest_activity == "Reading\nserver/device_gateway/app.py"
    assert "[TOOL] Reading" in session.recent_text


def test_terminal_notification_occurs_once_per_running_transition() -> None:
    registry = PiSessionRegistry()
    started = datetime(2026, 10, 2, 8, 0, tzinfo=timezone.utc)
    registry.apply(event("session_start", started))
    registry.apply(event("agent_start", started + timedelta(seconds=1)))
    done = registry.apply(event("agent_end", started + timedelta(seconds=2)))
    assert done.pending_notification == "done"
    duplicate_done = registry.apply(event("agent_end", started + timedelta(seconds=3)))
    assert duplicate_done.pending_notification is None

    registry.apply(event("agent_start", started + timedelta(seconds=4)))
    failed = registry.apply(event("agent_end", started + timedelta(seconds=5), failed=True))
    assert failed.pending_notification == "failed"
    # agent_settled is encoded as a second, non-failed agent_end by the extension.
    # It must neither notify again nor overwrite the terminal FAILED state.
    duplicate_failed = registry.apply(event("agent_end", started + timedelta(seconds=6)))
    assert duplicate_failed.pending_notification is None
    assert duplicate_failed.state == "FAILED"