mechaharness.core.events

Queryable structured event log.

A harness is an agent with a finite run. Parent agents share one EventLog with children and query it (including via tools). This module has no harness or injector imports.

class mechaharness.core.events.AccessCheck

Bases: CoreEvent

name: ClassVar[str] = 'access_check'

class mechaharness.core.events.AgentEnd

Bases: CoreEvent

name: ClassVar[str] = 'agent_end'

class mechaharness.core.events.AgentRef(*, agent_id: str, parent_agent_id: str | None = None, first_ts: datetime, last_ts: datetime, label: str | None = None)

Bases: BaseModel

An agent observed in the log.

agent_id: str

first_ts: datetime

label: str | None

last_ts: datetime

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parent_agent_id: str | None

class mechaharness.core.events.AgentStart

Bases: CoreEvent

name: ClassVar[str] = 'agent_start'

class mechaharness.core.events.CoreEvent

Bases: EventType

Built-in MechaHarness events (core:*).

namespace: ClassVar[str] = 'core'

class mechaharness.core.events.Cost

Bases: CoreEvent

name: ClassVar[str] = 'cost'

class mechaharness.core.events.Event(*, ts: datetime = <factory>, type: str, agent_id: str, run_id: str, parent_agent_id: str | None = None, payload: dict[str, ~typing.Any]=<factory>)

Bases: BaseModel

One structured log record.

agent_id: str

model_config = {}

Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

parent_agent_id: str | None

payload: dict[str, Any]

run_id: str

ts: datetime

type: str

class mechaharness.core.events.EventLog

Bases: ABC

Write and query structured agent events.

abstractmethod agents(*, parent_id: str | None = None) → list[AgentRef]

Agents seen in the log. parent_id limits to direct children.

abstractmethod emit(event: mechaharness.core.events.Event) → None

Append one event.

abstractmethod query(*, agent_id: str | None = None, run_id: str | None = None, descendants: bool = False, types: Sequence[str | type[EventType]] | None = None) → list[TypeAliasForwardRef('mechaharness.core.events.Event')]

Return matching events in emit order.

class mechaharness.core.events.EventType

Bases: object

Namespaced event identity.

Subclass to add types. The wire form is namespace:name (for example core:agent_start). Intermediate classes set namespace; leaves set name.

classmethod key() → str

name: ClassVar[str] = ''

namespace: ClassVar[str] = ''

classmethod parse(value: str) → type[TypeAliasForwardRef('mechaharness.core.events.EventType')]

Return the class registered for namespace:name.

class mechaharness.core.events.FanoutEventLog(*logs: mechaharness.core.events.EventLog)

Bases: EventLog

Emit to every log; query / agents use the first non-logging store.

__init__(*logs: mechaharness.core.events.EventLog) → None

agents(*, parent_id: str | None = None) → list[AgentRef]

Agents seen in the log. parent_id limits to direct children.

emit(event: mechaharness.core.events.Event) → None

Append one event.

query(*, agent_id: str | None = None, run_id: str | None = None, descendants: bool = False, types: Sequence[str | type[EventType]] | None = None) → list[TypeAliasForwardRef('mechaharness.core.events.Event')]

Return matching events in emit order.

class mechaharness.core.events.InMemoryEventLog

Bases: EventLog

List-backed store with agent / run / parent indexes.

__init__() → None

agents(*, parent_id: str | None = None) → list[AgentRef]

Agents seen in the log. parent_id limits to direct children.

emit(event: mechaharness.core.events.Event) → None

Append one event.

query(*, agent_id: str | None = None, run_id: str | None = None, descendants: bool = False, types: Sequence[str | type[EventType]] | None = None) → list[TypeAliasForwardRef('mechaharness.core.events.Event')]

Return matching events in emit order.

class mechaharness.core.events.Inference

Bases: CoreEvent

name: ClassVar[str] = 'inference'

class mechaharness.core.events.LoggingEventLog(logger: Logger | None = None)

Bases: EventLog

Write adapter: one JSON line per event on mechaharness.events.

Not a store: query / agents return empty lists.

__init__(logger: Logger | None = None) → None

agents(*, parent_id: str | None = None) → list[AgentRef]

Agents seen in the log. parent_id limits to direct children.

emit(event: mechaharness.core.events.Event) → None

Append one event.

query(*, agent_id: str | None = None, run_id: str | None = None, descendants: bool = False, types: Sequence[str | type[EventType]] | None = None) → list[TypeAliasForwardRef('mechaharness.core.events.Event')]

Return matching events in emit order.

class mechaharness.core.events.MaxTurns

Bases: CoreEvent

name: ClassVar[str] = 'max_turns'

class mechaharness.core.events.RunEnd

Bases: CoreEvent

name: ClassVar[str] = 'run_end'

class mechaharness.core.events.RunStart

Bases: CoreEvent

name: ClassVar[str] = 'run_start'

class mechaharness.core.events.ToolCall

Bases: CoreEvent

name: ClassVar[str] = 'tool_call'

class mechaharness.core.events.ToolResult

Bases: CoreEvent

name: ClassVar[str] = 'tool_result'

class mechaharness.core.events.TurnStart

Bases: CoreEvent

name: ClassVar[str] = 'turn_start'

mechaharness.core.events.default_event_log() → FanoutEventLog

In-memory store plus stdlib logging.

mechaharness.core.events.event_type_key(value: object) → str

Normalize a class or namespace:name string to the wire key.