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

1#!/usr/bin/env python3 

2""" 

3Lightweight event history for Osiris CLI. 

4 

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""" 

9 

10from __future__ import annotations 

11 

12from dataclasses import dataclass, field 

13from datetime import datetime 

14from pathlib import Path 

15from typing import Dict, List, Optional 

16 

17 

18@dataclass 

19class EventRecord: 

20 kind: str 

21 timestamp: datetime = field(default_factory=datetime.utcnow) 

22 payload: Optional[Dict] = None 

23 

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" 

30 

31 

32class EventHistory: 

33 """Simple in-memory history with helpers for summaries.""" 

34 

35 def __init__(self): 

36 self.events: List[EventRecord] = [] 

37 

38 def log(self, kind: str, payload: Optional[Dict] = None) -> None: 

39 self.events.append(EventRecord(kind=kind, payload=payload or {})) 

40 

41 def recent(self, limit: int = 20) -> List[EventRecord]: 

42 return self.events[-limit:] 

43 

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) 

51 

52 def clear(self) -> None: 

53 self.events.clear() 

54 

55 

56# Global history used across the CLI session 

57event_history = EventHistory()