"""Event types for the Yoker event system."""
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum, auto
from typing import Any
[docs]
class EventType(Enum):
"""Enumeration of all event types."""
# Session lifecycle
SESSION_START = auto()
SESSION_END = auto()
# Turn lifecycle
TURN_START = auto()
TURN_END = auto()
# Thinking (reasoning trace)
THINKING_START = auto()
THINKING_CHUNK = auto()
THINKING_END = auto()
# Content (response text)
CONTENT_START = auto()
CONTENT_CHUNK = auto()
CONTENT_END = auto()
# Tool execution
TOOL_CALL = auto()
TOOL_RESULT = auto()
TOOL_CONTENT = auto() # Content display event for write/update tools
# Command execution
COMMAND = auto()
# Error
ERROR = auto()
[docs]
@dataclass(frozen=True, kw_only=True)
class Event:
"""Base event class with common fields."""
type: EventType
timestamp: datetime = field(default_factory=datetime.now)
[docs]
@dataclass(frozen=True)
class SessionStartEvent(Event):
"""Emitted when agent session starts."""
model: str
thinking_enabled: bool
config_summary: dict[str, Any] = field(default_factory=dict)
[docs]
@dataclass(frozen=True)
class SessionEndEvent(Event):
"""Emitted when agent session ends."""
reason: str # "quit", "error", "interrupt"
[docs]
@dataclass(frozen=True)
class TurnStartEvent(Event):
"""Emitted when processing a user message begins."""
message: str
[docs]
@dataclass(frozen=True)
class TurnEndEvent(Event):
"""Emitted when processing a user message completes."""
response: str
tool_calls_count: int = 0
# Token statistics (from Ollama response)
prompt_eval_count: int = 0
eval_count: int = 0
# Duration in milliseconds
total_duration_ms: int = 0
[docs]
@dataclass(frozen=True)
class ThinkingStartEvent(Event):
"""Emitted when thinking output begins."""
pass
[docs]
@dataclass(frozen=True)
class ThinkingChunkEvent(Event):
"""Emitted for each chunk of thinking output."""
text: str
[docs]
@dataclass(frozen=True)
class ThinkingEndEvent(Event):
"""Emitted when thinking output ends."""
total_length: int
[docs]
@dataclass(frozen=True)
class ContentStartEvent(Event):
"""Emitted when content output begins."""
pass
[docs]
@dataclass(frozen=True)
class ContentChunkEvent(Event):
"""Emitted for each chunk of content output."""
text: str
[docs]
@dataclass(frozen=True)
class ContentEndEvent(Event):
"""Emitted when content output ends."""
total_length: int
[docs]
@dataclass(frozen=True)
class ToolContentEvent(Event):
"""Emitted when a tool has content to display (write/update operations).
This event is optional - tools can emit it when they have meaningful
content to show. The event handler can choose to display or ignore it.
Attributes:
tool_name: Name of the tool (e.g., "write", "update").
operation: Operation type (e.g., "write", "replace", "insert_before", "insert_after", "delete").
path: Resolved file path.
content_type: Type of content ("full", "diff", "summary").
content: Content to display (truncated if too large, None for summary type).
metadata: Additional metadata (lines, bytes, is_new_file, is_overwrite, etc.).
"""
tool_name: str
operation: str
path: str
content_type: str # "full", "diff", "summary"
content: str | None = None
metadata: dict[str, Any] = field(default_factory=dict)
[docs]
@dataclass(frozen=True)
class ErrorEvent(Event):
"""Emitted when an error occurs."""
error_type: str
message: str
details: dict[str, Any] = field(default_factory=dict)
[docs]
@dataclass(frozen=True)
class CommandEvent(Event):
"""Emitted when a slash command is executed."""
command: str # The command string (e.g., "/help")
result: str # The command output