from __future__ import annotations from dataclasses import dataclass from datetime import datetime, timezone from typing import Literal from pydantic import BaseModel, Field, model_validator from device_gateway.display_text import device_final_text PiState = Literal["IDLE", "RUNNING", "DONE", "FAILED", "OFFLINE"] PiActivityType = Literal["text", "tool"] PiNotificationKind = Literal["done", "failed"] PiEventName = Literal[ "session_start", "session_info_changed", "agent_start", "agent_end", "session_shutdown", "activity", ] SESSION_HISTORY_MAX_BYTES = 12 * 1024 DEVICE_HISTORY_MAX_BYTES = 4 * 1024 ADMIN_PREVIEW_MAX_BYTES = 480 LATEST_ACTIVITY_MAX_BYTES = 480 def utc_now() -> datetime: return datetime.now(timezone.utc) def utf8_tail(value: str, maximum_bytes: int) -> str: encoded = value.encode("utf-8") if len(encoded) <= maximum_bytes: return value return encoded[-maximum_bytes:].decode("utf-8", errors="ignore") def normalize_activity(value: str) -> str: return "".join( character for character in value.replace("\r\n", "\n") if character in "\n\t" or ord(character) >= 32 ) class PiSessionEvent(BaseModel): type: Literal["pi_session_state"] = "pi_session_state" event: PiEventName session_id: str = Field(min_length=1, max_length=160) name: str = Field(min_length=1, max_length=160) cwd: str = Field(min_length=1, max_length=1024) turn_id: str | None = Field(default=None, max_length=64) final_text: str = Field(default="", max_length=16000) failed: bool = False timestamp: datetime = Field(default_factory=utc_now) waiting_reason: str | None = None activity_type: PiActivityType | None = None activity: str | None = Field(default=None, max_length=2048) tool_name: str | None = Field(default=None, max_length=80) @model_validator(mode="after") def validate_activity(self) -> PiSessionEvent: if self.event == "activity" and (self.activity_type is None or not self.activity): raise ValueError("activity events require activity_type and activity") return self @dataclass class PiSession: session_id: str name: str cwd: str state: PiState started_at: datetime last_activity: datetime run_started_at: datetime | None = None run_elapsed_seconds: int = 0 waiting_reason: str | None = None latest_activity: str = "" recent_text: str = "" activity_type: PiActivityType | None = None activity_updated_at: datetime | None = None turn_id: str | None = None final_text: str = "" terminal_notified: bool = False pending_notification: PiNotificationKind | None = None def append_activity(self, event: PiSessionEvent) -> None: activity = normalize_activity(event.activity or "") if not activity: return history_entry = f"\n[TOOL] {activity}\n" if event.activity_type == "tool" else activity self.recent_text = utf8_tail( self.recent_text + history_entry, SESSION_HISTORY_MAX_BYTES ) self.latest_activity = utf8_tail(activity, LATEST_ACTIVITY_MAX_BYTES) self.activity_type = event.activity_type self.activity_updated_at = event.timestamp.astimezone(timezone.utc) def public( self, *, include_history: bool = False, now: datetime | None = None ) -> dict[str, object]: now = now or utc_now() elapsed = self.run_elapsed_seconds if self.state == "RUNNING" and self.run_started_at is not None: elapsed = max(0, int((now - self.run_started_at).total_seconds())) history_bytes = DEVICE_HISTORY_MAX_BYTES if include_history else ADMIN_PREVIEW_MAX_BYTES return { "session_id": self.session_id, "name": self.name, "cwd": self.cwd, "state": self.state, "started_at": self.started_at.isoformat(), "last_activity": self.last_activity.isoformat(), "elapsed_seconds": elapsed, "waiting_reason": self.waiting_reason, "latest_activity": self.latest_activity, "recent_text": utf8_tail(self.recent_text, history_bytes), "final_text": device_final_text(self.final_text) if self.state in ("DONE", "FAILED") else "", "activity_type": self.activity_type, "activity_updated_at": ( self.activity_updated_at.isoformat() if self.activity_updated_at is not None else None ), } class PiSessionRegistry: def __init__(self) -> None: self.sessions: dict[str, PiSession] = {} def apply(self, event: PiSessionEvent) -> PiSession: timestamp = event.timestamp.astimezone(timezone.utc) session = self.sessions.get(event.session_id) if session is None: session = PiSession( session_id=event.session_id, name=event.name, cwd=event.cwd, state="IDLE", started_at=timestamp, last_activity=timestamp, ) self.sessions[event.session_id] = session session.pending_notification = None session.name = event.name session.cwd = event.cwd session.last_activity = timestamp session.waiting_reason = event.waiting_reason if event.event == "session_start": session.state = "IDLE" session.started_at = timestamp session.run_started_at = None session.run_elapsed_seconds = 0 session.latest_activity = "" session.recent_text = "" session.activity_type = None session.activity_updated_at = None session.terminal_notified = False elif event.event == "agent_start": if event.turn_id and event.turn_id == session.turn_id: return session session.turn_id = event.turn_id session.final_text = "" session.state = "RUNNING" session.run_started_at = timestamp session.run_elapsed_seconds = 0 session.terminal_notified = False elif event.event == "agent_end": if event.turn_id and event.turn_id != session.turn_id: return session was_running = session.state == "RUNNING" if was_running: if session.run_started_at is not None: session.run_elapsed_seconds = max( 0, int((timestamp - session.run_started_at).total_seconds()) ) session.state = "FAILED" if event.failed else "DONE" if was_running and not session.terminal_notified: session.final_text = event.final_text session.pending_notification = "failed" if event.failed else "done" session.terminal_notified = True elif event.event == "session_shutdown": session.state = "OFFLINE" elif event.event == "activity": session.append_activity(event) return session def get(self, session_id: str) -> PiSession | None: return self.sessions.get(session_id) def public(self) -> list[dict[str, object]]: sessions = sorted( self.sessions.values(), key=lambda item: item.last_activity, reverse=True ) return [session.public() for session in sessions]