Index

ipw.telemetry

Telemetry collector implementations bundled with Intelligence Per Watt.

AgentEvent dataclass

Single event recorded during agent execution.

Source code in intelligence-per-watt/src/ipw/telemetry/events.py

@dataclass
class AgentEvent:
    """Single event recorded during agent execution."""

    event_type: str
    timestamp: float  # Unix timestamp from time.time()
    metadata: Dict[str, Any] = field(default_factory=dict)

    def __repr__(self) -> str:
        return f"AgentEvent({self.event_type!r}, ts={self.timestamp:.3f}, meta={self.metadata})"

EventRecorder

Thread-safe recorder for agent execution events.

Records events with timestamps for later correlation with energy telemetry. All operations are thread-safe for use in concurrent agent execution.

Example

recorder = EventRecorder() recorder.record('tool_call_start', tool='calculator') recorder.record('tool_call_end', tool='calculator') events = recorder.get_events() len(events) 2

Source code in intelligence-per-watt/src/ipw/telemetry/events.py

class EventRecorder:
    """Thread-safe recorder for agent execution events.

    Records events with timestamps for later correlation with energy telemetry.
    All operations are thread-safe for use in concurrent agent execution.

    Example:
        >>> recorder = EventRecorder()
        >>> recorder.record('tool_call_start', tool='calculator')
        >>> recorder.record('tool_call_end', tool='calculator')
        >>> events = recorder.get_events()
        >>> len(events)
        2
    """

    def __init__(self) -> None:
        """Initialize the event recorder."""
        self._events: List[AgentEvent] = []
        self._lock = threading.Lock()

    def record(self, event_type: str, **metadata: Any) -> None:
        """Record an event with current timestamp.

        Args:
            event_type: Type of event (e.g., 'tool_call_start', 'lm_inference_end')
            **metadata: Additional metadata to attach to the event
        """
        event = AgentEvent(
            event_type=event_type,
            timestamp=time.time(),
            metadata=metadata,
        )
        with self._lock:
            self._events.append(event)

    def get_events(self) -> List[AgentEvent]:
        """Return a copy of all recorded events.

        Returns:
            List of all recorded events in chronological order.
        """
        with self._lock:
            return list(self._events)

    def clear(self) -> None:
        """Clear all recorded events."""
        with self._lock:
            self._events.clear()

    def __len__(self) -> int:
        """Return the number of recorded events."""
        with self._lock:
            return len(self._events)

__init__()

Initialize the event recorder.

Source code in intelligence-per-watt/src/ipw/telemetry/events.py

def __init__(self) -> None:
    """Initialize the event recorder."""
    self._events: List[AgentEvent] = []
    self._lock = threading.Lock()

record(event_type, **metadata)

Record an event with current timestamp.

Parameters:

Name	Type	Description	Default
event_type	str	Type of event (e.g., 'tool_call_start', 'lm_inference_end')	required
**metadata	Any	Additional metadata to attach to the event	{}

Source code in intelligence-per-watt/src/ipw/telemetry/events.py

def record(self, event_type: str, **metadata: Any) -> None:
    """Record an event with current timestamp.

    Args:
        event_type: Type of event (e.g., 'tool_call_start', 'lm_inference_end')
        **metadata: Additional metadata to attach to the event
    """
    event = AgentEvent(
        event_type=event_type,
        timestamp=time.time(),
        metadata=metadata,
    )
    with self._lock:
        self._events.append(event)

get_events()

Return a copy of all recorded events.

Returns:

Type	Description
List[AgentEvent]	List of all recorded events in chronological order.

Source code in intelligence-per-watt/src/ipw/telemetry/events.py

def get_events(self) -> List[AgentEvent]:
    """Return a copy of all recorded events.

    Returns:
        List of all recorded events in chronological order.
    """
    with self._lock:
        return list(self._events)

clear()

Clear all recorded events.

Source code in intelligence-per-watt/src/ipw/telemetry/events.py

def clear(self) -> None:
    """Clear all recorded events."""
    with self._lock:
        self._events.clear()

__len__()

Return the number of recorded events.

Source code in intelligence-per-watt/src/ipw/telemetry/events.py

def __len__(self) -> int:
    """Return the number of recorded events."""
    with self._lock:
        return len(self._events)

EventType

Bases: str, Enum

Supported event types for agent telemetry.

Source code in intelligence-per-watt/src/ipw/telemetry/events.py

class EventType(str, Enum):
    """Supported event types for agent telemetry."""

    LM_INFERENCE_START = "lm_inference_start"
    LM_INFERENCE_END = "lm_inference_end"
    TOOL_CALL_START = "tool_call_start"
    TOOL_CALL_END = "tool_call_end"
    PREFILL_START = "prefill_start"
    PREFILL_END = "prefill_end"
    DECODE_START = "decode_start"
    DECODE_END = "decode_end"
    # Submodel call events (for MCP tools that call inference servers)
    SUBMODEL_CALL_START = "submodel_call_start"
    SUBMODEL_CALL_END = "submodel_call_end"

ensure_monitor(target=DEFAULT_TARGET, *, timeout=5.0, launch=True, launch_args=None, env=None)

Ensure a monitor is reachable, optionally launching one in the background.

Yields the normalized target string. If launch is True, a background process will be spawned when the target is unavailable. The caller is responsible for any teardown when the context exits.

Source code in intelligence-per-watt/src/ipw/telemetry/launcher.py

@contextmanager
def ensure_monitor(
    target: str = DEFAULT_TARGET,
    *,
    timeout: float = 5.0,
    launch: bool = True,
    launch_args: Sequence[str] | None = None,
    env: MutableMapping[str, str] | None = None,
) -> Iterator[str]:
    """Ensure a monitor is reachable, optionally launching one in the background.

    Yields the normalized target string. If ``launch`` is True, a background
    process will be spawned when the target is unavailable. The caller is
    responsible for any teardown when the context exits.
    """

    normalized = normalize_target(target)
    pid: int | None = None
    if not wait_for_ready(normalized, timeout=timeout) and launch:
        env_map: Mapping[str, str] | None = env if env is not None else os.environ
        pid, normalized = launch_monitor(
            launch_args,
            env=env_map,
            timeout=timeout,
            target=normalized,
        )

    try:
        yield normalized
    finally:
        if pid is not None:
            try:
                os.kill(pid, signal.SIGTERM)
            except OSError:
                pass

wait_for_ready(target=DEFAULT_TARGET, *, timeout=5.0)

Return True when the energy monitor responds to a health check.

Source code in intelligence-per-watt/src/ipw/telemetry/launcher.py

def wait_for_ready(target: str = DEFAULT_TARGET, *, timeout: float = 5.0) -> bool:
    """Return True when the energy monitor responds to a health check."""

    normalized = normalize_target(target)
    bundle = get_stub_bundle()
    channel = grpc.insecure_channel(normalized)
    stub = bundle.stub_factory(channel)
    try:
        grpc.channel_ready_future(channel).result(timeout=timeout)
        stub.Health(bundle.HealthRequestCls(), timeout=timeout)
        return True
    except Exception:
        return False
    finally:
        channel.close()