Coverage for src / osiris_cli / event_history.py: 0%
33 statements
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-28 16:44 +0200
« prev ^ index » next coverage.py v7.13.0, created at 2025-12-28 16:44 +0200
1#!/usr/bin/env python3
2"""
3Lightweight event history for Osiris CLI.
5Gemini CLI maintains an internal event bus so the UI can summarize work even
6when the model does not stream user-facing text. This module provides a
7similar capability for the Python CLI.
8"""
10from __future__ import annotations
12from dataclasses import dataclass, field
13from datetime import datetime
14from pathlib import Path
15from typing import Dict, List, Optional
18@dataclass
19class EventRecord:
20 kind: str
21 timestamp: datetime = field(default_factory=datetime.utcnow)
22 payload: Optional[Dict] = None
24 def to_markdown(self) -> str:
25 payload_str = ""
26 if self.payload:
27 segments = [f"- **{k}**: {v}" for k, v in self.payload.items()]
28 payload_str = "\n" + "\n".join(segments)
29 return f"### {self.kind}\n- Time (UTC): {self.timestamp.isoformat()}{payload_str}\n"
32class EventHistory:
33 """Simple in-memory history with helpers for summaries."""
35 def __init__(self):
36 self.events: List[EventRecord] = []
38 def log(self, kind: str, payload: Optional[Dict] = None) -> None:
39 self.events.append(EventRecord(kind=kind, payload=payload or {}))
41 def recent(self, limit: int = 20) -> List[EventRecord]:
42 return self.events[-limit:]
44 def to_markdown(self) -> str:
45 if not self.events:
46 return "No activity recorded this session."
47 lines = ["# Session Activity"]
48 for event in self.events:
49 lines.append(event.to_markdown())
50 return "\n".join(lines)
52 def clear(self) -> None:
53 self.events.clear()
56# Global history used across the CLI session
57event_history = EventHistory()