From d17895dae22df5b091cee19f50de221ca63adee8 Mon Sep 17 00:00:00 2001 From: Aditya Chhabra Date: Mon, 2 Mar 2026 11:54:53 +0000 Subject: [PATCH] feat(events): add EventBus protocol and ReactiveEventBus implementation - Add EventType StrEnum with 38 typed domain event identifiers across 8 domains (plan lifecycle, decision, actor, tool, resource, sandbox, validation, session, budget) - Add frozen DomainEvent Pydantic model with event_type, auto-UTC timestamp, auto-ULID correlation_id, plan_id, root_plan_id, session_id, actor_name, project_name, and details fields - Add @runtime_checkable EventBus Protocol with emit() and subscribe() - Add ReactiveEventBus backed by RxPY Subject for in-process reactive streaming with synchronous handler dispatch and raw Observable stream - Add LoggingEventBus using structlog for audit-trail-only environments - Wire DecisionService to emit DECISION_CREATED on record_decision() - Wire PlanLifecycleService to emit PLAN_CREATED and PLAN_PHASE_CHANGED on use_action() and execute_plan() respectively - Register ReactiveEventBus as Singleton in DI container and inject into DecisionService and PlanLifecycleService providers - Add Behave BDD unit tests: 27 scenarios, 75 steps, 100% branch coverage - Add Robot Framework integration tests: 9 smoke cases - Add ASV performance benchmarks: 5 suites covering emit throughput and subscriber fan-out - Add reference documentation at docs/reference/event_bus.md ISSUES CLOSED: #473 --- CHANGELOG.md | 19 + benchmarks/event_bus_bench.py | 159 ++++++ docs/reference/event_bus.md | 220 ++++++++ features/event_bus.feature | 165 ++++++ features/steps/event_bus_steps.py | 476 ++++++++++++++++++ robot/event_bus.robot | 91 ++++ robot/helper_event_bus.py | 203 ++++++++ src/cleveragents/application/container.py | 6 + .../application/services/decision_service.py | 21 + .../services/plan_lifecycle_service.py | 33 ++ .../infrastructure/events/__init__.py | 37 ++ .../infrastructure/events/logging_bus.py | 96 ++++ .../infrastructure/events/models.py | 68 +++ .../infrastructure/events/protocol.py | 53 ++ .../infrastructure/events/reactive.py | 100 ++++ .../infrastructure/events/types.py | 86 ++++ 16 files changed, 1833 insertions(+) create mode 100644 benchmarks/event_bus_bench.py create mode 100644 docs/reference/event_bus.md create mode 100644 features/event_bus.feature create mode 100644 features/steps/event_bus_steps.py create mode 100644 robot/event_bus.robot create mode 100644 robot/helper_event_bus.py create mode 100644 src/cleveragents/infrastructure/events/__init__.py create mode 100644 src/cleveragents/infrastructure/events/logging_bus.py create mode 100644 src/cleveragents/infrastructure/events/models.py create mode 100644 src/cleveragents/infrastructure/events/protocol.py create mode 100644 src/cleveragents/infrastructure/events/reactive.py create mode 100644 src/cleveragents/infrastructure/events/types.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e52baa96..c15e5582 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -2,6 +2,25 @@ ## Unreleased +- Added general-purpose domain event system under + `cleveragents.infrastructure.events`. `EventType` StrEnum defines 40 typed + event identifiers across 8 domains (plan lifecycle, decision, actor, tool, + resource, sandbox, validation, session, budget). `DomainEvent` is a frozen + Pydantic model with `event_type`, auto-UTC `timestamp`, auto-ULID + `correlation_id`, `plan_id`, `root_plan_id`, `session_id`, `actor_name`, + `project_name`, and `details` fields. `EventBus` is a `@runtime_checkable` + Protocol with `emit()` and `subscribe()` methods. `ReactiveEventBus` is an + RxPY `Subject`-backed in-process bus that dispatches synchronously to + type-filtered handlers and exposes a raw `rx.Observable` stream for advanced + operators. `LoggingEventBus` is a structlog-based bus for audit logging that + requires no RxPY dependency. `DecisionService` and `PlanLifecycleService` + accept an optional `event_bus` parameter (backward-compatible) and emit + `DECISION_CREATED`, `PLAN_CREATED`, and `PLAN_PHASE_CHANGED` on significant + state changes. `ReactiveEventBus` is registered as a Singleton in the DI + container and wired into both services automatically. Includes Behave BDD unit + tests (27 scenarios, 75 steps, 100% coverage on all new source files), Robot + Framework smoke tests (9 cases), ASV performance benchmarks (5 suites), and + reference documentation (`docs/reference/event_bus.md`). (#473) - Replaced behave-parallel subprocess-per-feature execution model (342 Python interpreter startups) with in-process execution via behave's `Runner` API. Sequential mode runs all features in a single `Runner.run()` call; parallel mode uses `multiprocessing.Pool` with diff --git a/benchmarks/event_bus_bench.py b/benchmarks/event_bus_bench.py new file mode 100644 index 00000000..cba94c3d --- /dev/null +++ b/benchmarks/event_bus_bench.py @@ -0,0 +1,159 @@ +"""ASV benchmarks for EventBus infrastructure. + +Measures: +- DomainEvent construction throughput +- ReactiveEventBus emit overhead (no subscribers) +- ReactiveEventBus emit with one subscriber +- ReactiveEventBus subscribe registration +- LoggingEventBus emit overhead +- Fan-out: emit to N subscribers +""" + +from __future__ import annotations + +import importlib +import sys +from pathlib import Path +from typing import ClassVar + +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) + +import cleveragents # noqa: E402 + +importlib.reload(cleveragents) + +from cleveragents.infrastructure.events import ( # noqa: E402 + DomainEvent, + EventType, + LoggingEventBus, + ReactiveEventBus, +) + +# --------------------------------------------------------------------------- +# DomainEvent construction +# --------------------------------------------------------------------------- + + +class DomainEventSuite: + """Benchmark DomainEvent construction overhead.""" + + timeout = 60 + + def setup(self) -> None: + self.event_type = EventType.PLAN_CREATED + self.plan_id = "01ARZ3NDEKTSV4RRFFQ69G5FAV" + + def time_create_minimal(self) -> None: + DomainEvent(event_type=self.event_type) + + def time_create_with_plan_id(self) -> None: + DomainEvent(event_type=self.event_type, plan_id=self.plan_id) + + def time_create_with_details(self) -> None: + DomainEvent( + event_type=self.event_type, + plan_id=self.plan_id, + details={"phase": "strategize", "actor": "strategists/planner"}, + ) + + +# --------------------------------------------------------------------------- +# ReactiveEventBus emit +# --------------------------------------------------------------------------- + + +class ReactiveEmitSuite: + """Benchmark ReactiveEventBus.emit() throughput.""" + + timeout = 60 + + def setup(self) -> None: + self.bus = ReactiveEventBus() + self.event = DomainEvent(event_type=EventType.PLAN_CREATED) + self.decision_event = DomainEvent(event_type=EventType.DECISION_CREATED) + + # Pre-register a subscriber for the with-subscriber benchmarks + self._received: list[DomainEvent] = [] + self.bus.subscribe(EventType.PLAN_CREATED, self._received.append) + + def time_emit_no_subscribers(self) -> None: + self.bus.emit(self.decision_event) + + def time_emit_with_one_subscriber(self) -> None: + self.bus.emit(self.event) + + def time_emit_100_events(self) -> None: + for _ in range(100): + self.bus.emit(self.event) + + +# --------------------------------------------------------------------------- +# ReactiveEventBus subscribe +# --------------------------------------------------------------------------- + + +class ReactiveSubscribeSuite: + """Benchmark ReactiveEventBus.subscribe() overhead.""" + + timeout = 60 + + def setup(self) -> None: + self.event_type = EventType.PLAN_CREATED + + def time_subscribe_single(self) -> None: + bus = ReactiveEventBus() + bus.subscribe(self.event_type, lambda e: None) + + def time_subscribe_10(self) -> None: + bus = ReactiveEventBus() + for _ in range(10): + bus.subscribe(self.event_type, lambda e: None) + + +# --------------------------------------------------------------------------- +# Fan-out benchmark +# --------------------------------------------------------------------------- + + +class FanOutSuite: + """Benchmark event fan-out to multiple subscribers.""" + + timeout = 60 + params: ClassVar[list[int]] = [1, 5, 10, 50] + param_names: ClassVar[list[str]] = ["num_subscribers"] + + def setup(self, num_subscribers: int) -> None: + self.bus = ReactiveEventBus() + for _ in range(num_subscribers): + self.bus.subscribe(EventType.PLAN_CREATED, lambda e: None) + self.event = DomainEvent(event_type=EventType.PLAN_CREATED) + + def time_emit(self, num_subscribers: int) -> None: + self.bus.emit(self.event) + + +# --------------------------------------------------------------------------- +# LoggingEventBus emit +# --------------------------------------------------------------------------- + + +class LoggingEmitSuite: + """Benchmark LoggingEventBus.emit() overhead.""" + + timeout = 60 + + def setup(self) -> None: + import logging + + logging.disable(logging.CRITICAL) + self.bus = LoggingEventBus() + self.event = DomainEvent(event_type=EventType.DECISION_CREATED) + + def time_emit_single(self) -> None: + self.bus.emit(self.event) + + def time_emit_100(self) -> None: + for _ in range(100): + self.bus.emit(self.event) diff --git a/docs/reference/event_bus.md b/docs/reference/event_bus.md new file mode 100644 index 00000000..679d0d76 --- /dev/null +++ b/docs/reference/event_bus.md @@ -0,0 +1,220 @@ +# Event Bus + +The **EventBus** infrastructure provides a general-purpose domain event system +for CleverAgents. Domain services emit typed events at significant state changes +(plan creation, decision recording, phase transitions, etc.), enabling audit +logging, reactive subscribers, and observability without coupling services to +one another. + +--- + +## Overview + +The system consists of four cooperating components: + +| Component | Role | +|-----------|------| +| `EventType` | `StrEnum` of all domain event identifiers | +| `DomainEvent` | Frozen Pydantic model carrying event data | +| `EventBus` | `Protocol` that all bus implementations satisfy | +| `ReactiveEventBus` | RxPY-backed in-process bus (default) | +| `LoggingEventBus` | Structured-log-only bus (audit / test) | + +--- + +## EventType + +`cleveragents.infrastructure.events.EventType` is a `StrEnum` whose members +cover all domains: + +```python +from cleveragents.infrastructure.events import EventType + +print(EventType.PLAN_CREATED) # "plan.created" +print(EventType.DECISION_CREATED) # "decision.created" +print(EventType.ACTOR_COMPLETED) # "actor.completed" +``` + +**Categories:** + +| Category | Examples | +|----------|---------| +| Plan lifecycle | `plan.created`, `plan.phase_changed`, `plan.applied` | +| Decision | `decision.created`, `decision.superseded` | +| Actor | `actor.invoked`, `actor.completed`, `actor.errored` | +| Tool | `tool.invoked`, `tool.completed`, `tool.errored` | +| Resource | `resource.accessed`, `resource.modified` | +| Sandbox | `sandbox.created`, `checkpoint.restored` | +| Validation | `validation.started`, `validation.passed` | +| Session | `session.created`, `session.message_sent` | +| Budget | `budget.warning`, `budget.exceeded` | + +--- + +## DomainEvent + +All events are instances of the frozen Pydantic model +`cleveragents.infrastructure.events.DomainEvent`: + +```python +from cleveragents.infrastructure.events import DomainEvent, EventType + +event = DomainEvent( + event_type=EventType.PLAN_CREATED, + plan_id="01ARZ3NDEKTSV4RRFFQ69G5FAV", + actor_name="strategists/planner", + details={"phase": "strategize"}, +) +``` + +| Field | Type | Default | Description | +|-------|------|---------|-------------| +| `event_type` | `EventType` | required | The type of event | +| `timestamp` | `datetime` | `now(UTC)` | UTC time of emission | +| `correlation_id` | `str` | auto ULID | Links related events in a request chain | +| `plan_id` | `str \| None` | `None` | Associated plan ULID | +| `root_plan_id` | `str \| None` | `None` | Root plan in a hierarchy | +| `session_id` | `str \| None` | `None` | Associated session ULID | +| `actor_name` | `str \| None` | `None` | Actor that emitted the event | +| `project_name` | `str \| None` | `None` | Project context | +| `details` | `dict` | `{}` | Event-type-specific payload | + +`DomainEvent` is **immutable** (`frozen=True`). Attempting to set an attribute +after construction raises a `ValidationError`. + +--- + +## EventBus Protocol + +`cleveragents.infrastructure.events.EventBus` is a `@runtime_checkable` +structural protocol. Any object with `emit()` and `subscribe()` methods +satisfies it: + +```python +from cleveragents.infrastructure.events import EventBus, ReactiveEventBus + +bus = ReactiveEventBus() +assert isinstance(bus, EventBus) # True +``` + +**Interface:** + +```python +class EventBus(Protocol): + def emit(self, event: DomainEvent) -> None: ... + def subscribe(self, event_type: EventType, + handler: Callable[[DomainEvent], None]) -> None: ... +``` + +--- + +## ReactiveEventBus + +`cleveragents.infrastructure.events.ReactiveEventBus` is the default +in-process implementation. It uses an RxPY `Subject` as its hot observable +backbone. + +```python +from cleveragents.infrastructure.events import ( + DomainEvent, EventType, ReactiveEventBus, +) + +bus = ReactiveEventBus() + +# Subscribe a plain callback +bus.subscribe(EventType.PLAN_CREATED, lambda e: print("Plan created:", e.plan_id)) + +# Use the raw RxPY stream for advanced operators +bus.stream.pipe( + rx.operators.filter(lambda e: e.plan_id is not None), +).subscribe(lambda e: print("Plan event:", e)) + +# Emit an event +bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED, plan_id="01ABC...")) +``` + +- `emit(event)` pushes to the RxPY Subject **and** calls all type-specific + handlers synchronously. +- `subscribe(event_type, handler)` registers a callback for one event type. +- `stream` exposes the raw `rx.Observable` for RxPY operators. + +--- + +## LoggingEventBus + +`cleveragents.infrastructure.events.LoggingEventBus` logs every event to +`structlog` at `INFO` level. It is useful for audit trails, CI pipelines, and +environments where RxPY overhead is undesirable. + +```python +from cleveragents.infrastructure.events import LoggingEventBus, EventType, DomainEvent + +bus = LoggingEventBus() +bus.emit(DomainEvent(event_type=EventType.DECISION_CREATED, plan_id="01ABC...")) +# → structured log: {"event": "domain_event", "event_type": "decision.created", ...} +``` + +--- + +## Dependency Injection + +The DI container (`cleveragents.application.container.Container`) registers +`ReactiveEventBus` as a **Singleton** provider named `event_bus`. It is +automatically wired into `DecisionService` and `PlanLifecycleService`: + +```python +from cleveragents.application.container import Container + +container = Container() +bus = container.event_bus() # ReactiveEventBus singleton +``` + +To substitute a different implementation (e.g. in tests): + +```python +from cleveragents.application.container import Container +from cleveragents.infrastructure.events import LoggingEventBus +from dependency_injector import providers + +container = Container() +container.event_bus.override(providers.Singleton(LoggingEventBus)) +``` + +--- + +## Service Event Emission + +### DecisionService + +`cleveragents.application.services.decision_service.DecisionService.record_decision` +emits **`DECISION_CREATED`** after a decision is persisted. + +### PlanLifecycleService + +| Method | Event emitted | +|--------|---------------| +| `use_action(...)` | `PLAN_CREATED` | +| `execute_plan(...)` | `PLAN_PHASE_CHANGED` | + +Both services accept an optional `event_bus` constructor parameter; when +`None`, event emission is silently skipped (backward-compatible). + +--- + +## Example: Subscribing to Plan Events + +```python +from cleveragents.application.container import Container +from cleveragents.infrastructure.events import EventType, DomainEvent + +container = Container() +bus = container.event_bus() + +def on_plan_created(event: DomainEvent) -> None: + print(f"New plan: {event.plan_id} | details: {event.details}") + +bus.subscribe(EventType.PLAN_CREATED, on_plan_created) + +# All subsequent plan.create operations will call on_plan_created +svc = container.plan_lifecycle_service() +``` diff --git a/features/event_bus.feature b/features/event_bus.feature new file mode 100644 index 00000000..2875b4f5 --- /dev/null +++ b/features/event_bus.feature @@ -0,0 +1,165 @@ +Feature: EventBus protocol and domain event emission + As a domain service + I want to emit typed domain events through an EventBus + So that audit logging, observability, and reactive subscribers work correctly + + # --------------------------------------------------------------------------- + # EventType enum + # --------------------------------------------------------------------------- + + Scenario: EventType enum covers required lifecycle values + Then EventType should include "plan.created" + And EventType should include "decision.created" + And EventType should include "tool.invoked" + And EventType should include "actor.completed" + And EventType should include "validation.passed" + And EventType should include "budget.exceeded" + + Scenario: EventType string values compare equal to their members + Then EventType member PLAN_CREATED should equal string "plan.created" + And EventType member DECISION_CREATED should equal string "decision.created" + + # --------------------------------------------------------------------------- + # DomainEvent model + # --------------------------------------------------------------------------- + + Scenario: DomainEvent is created with required event_type + When I create a minimal DomainEvent with event_type "plan.created" + Then the DomainEvent event_type should be "plan.created" + And the DomainEvent should have a non-empty correlation_id + And the DomainEvent should have a timestamp + + Scenario: DomainEvent defaults are applied for optional fields + When I create a minimal DomainEvent with event_type "decision.created" + Then the DomainEvent plan_id should be None + And the DomainEvent actor_name should be None + And the DomainEvent details should be empty + + Scenario: DomainEvent accepts plan_id and details + When I create a DomainEvent with event_type "plan.created" plan_id "01ARZ3NDEKTSV4RRFFQ69G5FAV" and details key "phase" value "strategize" + Then the DomainEvent plan_id should be "01ARZ3NDEKTSV4RRFFQ69G5FAV" + And the DomainEvent details should contain key "phase" + + Scenario: DomainEvent is immutable (frozen) + When I create a minimal DomainEvent with event_type "plan.created" + Then modifying the DomainEvent should raise an error + + # --------------------------------------------------------------------------- + # EventBus Protocol + # --------------------------------------------------------------------------- + + Scenario: ReactiveEventBus satisfies the EventBus protocol + Then ReactiveEventBus should satisfy the EventBus protocol + + Scenario: LoggingEventBus satisfies the EventBus protocol + Then LoggingEventBus should satisfy the EventBus protocol + + # --------------------------------------------------------------------------- + # ReactiveEventBus + # --------------------------------------------------------------------------- + + Scenario: ReactiveEventBus emits events to subscribed handlers + Given a ReactiveEventBus + When I subscribe to "plan.created" events + And I emit a "plan.created" DomainEvent + Then the handler should have received 1 event + And the received event type should be "plan.created" + + Scenario: ReactiveEventBus dispatches only to matching event type + Given a ReactiveEventBus + When I subscribe to "plan.created" events + And I emit a "decision.created" DomainEvent + Then the handler should have received 0 event + + Scenario: ReactiveEventBus supports multiple handlers per event type + Given a ReactiveEventBus + When I subscribe two handlers to "plan.created" events + And I emit a "plan.created" DomainEvent + Then each handler should have received 1 event + + Scenario: ReactiveEventBus exposes an observable stream + Given a ReactiveEventBus + Then the bus should expose an observable stream + + Scenario: ReactiveEventBus rejects non-DomainEvent inputs + Given a ReactiveEventBus + Then emitting a non-DomainEvent should raise TypeError + + Scenario: ReactiveEventBus rejects invalid event_type in subscribe + Given a ReactiveEventBus + Then subscribing with a non-EventType should raise TypeError + + # --------------------------------------------------------------------------- + # LoggingEventBus + # --------------------------------------------------------------------------- + + Scenario: LoggingEventBus emits events to subscribed handlers + Given a LoggingEventBus + When I subscribe to "decision.created" events + And I emit a "decision.created" DomainEvent + Then the handler should have received 1 event + + Scenario: LoggingEventBus rejects non-DomainEvent inputs + Given a LoggingEventBus + Then emitting a non-DomainEvent should raise TypeError + + Scenario: LoggingEventBus rejects invalid event_type in subscribe + Given a LoggingEventBus + Then subscribing with a non-EventType should raise TypeError + + Scenario: LoggingEventBus rejects non-callable handler in subscribe + Given a LoggingEventBus + Then subscribing with a non-callable handler should raise TypeError + + Scenario: ReactiveEventBus rejects non-callable handler in subscribe + Given a ReactiveEventBus + Then subscribing with a non-callable handler should raise TypeError + + Scenario: EventBus Protocol emit and subscribe stubs are executable + Then the EventBus Protocol stubs should be callable + + # --------------------------------------------------------------------------- + # DecisionService event emission + # --------------------------------------------------------------------------- + + Scenario: DecisionService emits DECISION_CREATED when recording a decision + Given a DecisionService with an injected event bus + When I call record_decision + Then the event bus should have received a "decision.created" event + And the emitted event plan_id should match the decision plan_id + + Scenario: DecisionService works without an event bus (no-op) + Given a DecisionService without an event bus + When I call record_decision + Then no event bus exception should be raised + + # --------------------------------------------------------------------------- + # PlanLifecycleService event emission + # --------------------------------------------------------------------------- + + Scenario: PlanLifecycleService emits PLAN_CREATED on use_action + Given a PlanLifecycleService with an injected event bus + When I call use_action to create a plan + Then the event bus should have received a "plan.created" event + + Scenario: PlanLifecycleService emits PLAN_PHASE_CHANGED on execute_plan + Given a PlanLifecycleService with an injected event bus and a strategized plan + When I call execute_plan + Then the event bus should have received a "plan.phase_changed" event + + Scenario: PlanLifecycleService works without an event bus (no-op) + Given a PlanLifecycleService without an event bus + When I call use_action to create a plan + Then no event bus exception should be raised + + # --------------------------------------------------------------------------- + # DI container + # --------------------------------------------------------------------------- + + Scenario: Container provides ReactiveEventBus as a Singleton + When I resolve event_bus from the DI container twice + Then both resolutions should return the same instance + + Scenario: Container wires event_bus into DecisionService + When I resolve decision_service from the DI container + Then the decision_service should have an event_bus attribute diff --git a/features/steps/event_bus_steps.py b/features/steps/event_bus_steps.py new file mode 100644 index 00000000..aa98cd25 --- /dev/null +++ b/features/steps/event_bus_steps.py @@ -0,0 +1,476 @@ +"""Step definitions for event_bus.feature. + +Tests EventType, DomainEvent, EventBus protocol, ReactiveEventBus, +LoggingEventBus, service event emission, and DI container registration. +""" + +from __future__ import annotations + +from typing import Any +from unittest.mock import MagicMock + +from behave import given, then, when # type: ignore[import-untyped] +from behave.runner import Context # type: ignore[import-untyped] +from ulid import ULID + +from cleveragents.application.container import Container +from cleveragents.application.services.decision_service import DecisionService +from cleveragents.application.services.plan_lifecycle_service import ( + PlanLifecycleService, +) +from cleveragents.domain.models.core.decision import Decision, DecisionType +from cleveragents.domain.models.core.plan import PlanPhase, ProcessingState +from cleveragents.infrastructure.events import ( + DomainEvent, + EventBus, + EventType, + LoggingEventBus, + ReactiveEventBus, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + + +def _make_event(event_type_str: str, **kwargs: Any) -> DomainEvent: + return DomainEvent(event_type=EventType(event_type_str), **kwargs) + + +class _EventCollector: + """Simple in-test stub that collects emitted events.""" + + def __init__(self) -> None: + self.received: list[DomainEvent] = [] + + def __call__(self, event: DomainEvent) -> None: + self.received.append(event) + + +# --------------------------------------------------------------------------- +# EventType enum steps +# --------------------------------------------------------------------------- + + +@then('EventType should include "{value}"') +def step_event_type_includes(ctx: Context, value: str) -> None: + values = {e.value for e in EventType} + assert value in values, f"{value!r} not found in EventType values: {values}" + + +@then('EventType member {member} should equal string "{value}"') +def step_event_type_member_equals(ctx: Context, member: str, value: str) -> None: + et = EventType[member] + assert et == value, f"EventType.{member} = {et!r}, expected {value!r}" + + +# --------------------------------------------------------------------------- +# DomainEvent steps +# --------------------------------------------------------------------------- + + +@when('I create a minimal DomainEvent with event_type "{et}"') +def step_create_domain_event(ctx: Context, et: str) -> None: + ctx.domain_event = _make_event(et) + + +@when( + 'I create a DomainEvent with event_type "{et}" plan_id "{pid}"' + ' and details key "{key}" value "{val}"' +) +def step_create_domain_event_with_details( + ctx: Context, et: str, pid: str, key: str, val: str +) -> None: + ctx.domain_event = _make_event(et, plan_id=pid, details={key: val}) + + +@then('the DomainEvent event_type should be "{et}"') +def step_domain_event_type(ctx: Context, et: str) -> None: + assert str(ctx.domain_event.event_type) == et + + +@then("the DomainEvent should have a non-empty correlation_id") +def step_domain_event_correlation_id(ctx: Context) -> None: + cid = ctx.domain_event.correlation_id + assert cid and len(cid) == 26, f"Expected 26-char ULID, got {cid!r}" + + +@then("the DomainEvent should have a timestamp") +def step_domain_event_timestamp(ctx: Context) -> None: + assert ctx.domain_event.timestamp is not None + + +@then("the DomainEvent plan_id should be None") +def step_domain_event_plan_id_none(ctx: Context) -> None: + assert ctx.domain_event.plan_id is None + + +@then("the DomainEvent actor_name should be None") +def step_domain_event_actor_name_none(ctx: Context) -> None: + assert ctx.domain_event.actor_name is None + + +@then("the DomainEvent details should be empty") +def step_domain_event_details_empty(ctx: Context) -> None: + assert ctx.domain_event.details == {} + + +@then('the DomainEvent plan_id should be "{pid}"') +def step_domain_event_plan_id(ctx: Context, pid: str) -> None: + assert ctx.domain_event.plan_id == pid + + +@then('the DomainEvent details should contain key "{key}"') +def step_domain_event_details_key(ctx: Context, key: str) -> None: + assert key in ctx.domain_event.details + + +@then("modifying the DomainEvent should raise an error") +def step_domain_event_immutable(ctx: Context) -> None: + raised = False + try: + ctx.domain_event.event_type = EventType.PLAN_CANCELLED # type: ignore[misc] + except Exception: + raised = True + assert raised, "Expected an exception when mutating a frozen DomainEvent" + + +# --------------------------------------------------------------------------- +# Protocol conformance +# --------------------------------------------------------------------------- + + +@then("ReactiveEventBus should satisfy the EventBus protocol") +def step_reactive_satisfies_protocol(ctx: Context) -> None: + bus = ReactiveEventBus() + assert isinstance(bus, EventBus), "ReactiveEventBus does not satisfy EventBus" + + +@then("LoggingEventBus should satisfy the EventBus protocol") +def step_logging_satisfies_protocol(ctx: Context) -> None: + bus = LoggingEventBus() + assert isinstance(bus, EventBus), "LoggingEventBus does not satisfy EventBus" + + +# --------------------------------------------------------------------------- +# ReactiveEventBus given +# --------------------------------------------------------------------------- + + +@given("a ReactiveEventBus") +def step_given_reactive_bus(ctx: Context) -> None: + ctx.bus = ReactiveEventBus() + ctx.collectors: list[_EventCollector] = [] + + +@given("a LoggingEventBus") +def step_given_logging_bus(ctx: Context) -> None: + ctx.bus = LoggingEventBus() + ctx.collectors = [] + + +# --------------------------------------------------------------------------- +# ReactiveEventBus / LoggingEventBus when +# --------------------------------------------------------------------------- + + +@when('I subscribe to "{et}" events') +def step_subscribe_single(ctx: Context, et: str) -> None: + collector = _EventCollector() + ctx.collectors.append(collector) + ctx.bus.subscribe(EventType(et), collector) + + +@when('I subscribe two handlers to "{et}" events') +def step_subscribe_two(ctx: Context, et: str) -> None: + for _ in range(2): + collector = _EventCollector() + ctx.collectors.append(collector) + ctx.bus.subscribe(EventType(et), collector) + + +@when('I emit a "{et}" DomainEvent') +def step_emit_event(ctx: Context, et: str) -> None: + ctx.bus.emit(_make_event(et)) + + +# --------------------------------------------------------------------------- +# ReactiveEventBus then +# --------------------------------------------------------------------------- + + +@then("the handler should have received {n:d} event") +def step_handler_received_n(ctx: Context, n: int) -> None: + assert ctx.collectors, "No collectors registered" + count = len(ctx.collectors[0].received) + assert count == n, f"Expected {n} events, got {count}" + + +@then('the received event type should be "{et}"') +def step_received_event_type(ctx: Context, et: str) -> None: + assert ctx.collectors[0].received, "No events received" + assert str(ctx.collectors[0].received[0].event_type) == et + + +@then("each handler should have received {n:d} event") +def step_each_handler_received(ctx: Context, n: int) -> None: + for i, col in enumerate(ctx.collectors): + assert len(col.received) == n, ( + f"Handler {i} received {len(col.received)}, expected {n}" + ) + + +@then("the bus should expose an observable stream") +def step_bus_has_stream(ctx: Context) -> None: + stream = ctx.bus.stream + assert stream is not None, "Expected a non-None observable stream" + + +@then("emitting a non-DomainEvent should raise TypeError") +def step_emit_non_domain_event(ctx: Context) -> None: + raised = False + try: + ctx.bus.emit("not-an-event") # type: ignore[arg-type] + except TypeError: + raised = True + assert raised, "Expected TypeError for non-DomainEvent" + + +@then("subscribing with a non-EventType should raise TypeError") +def step_subscribe_bad_type(ctx: Context) -> None: + raised = False + try: + ctx.bus.subscribe("plan.created", lambda e: None) # type: ignore[arg-type] + except TypeError: + raised = True + assert raised, "Expected TypeError for non-EventType event_type" + + +@then("subscribing with a non-callable handler should raise TypeError") +def step_subscribe_non_callable(ctx: Context) -> None: + raised = False + try: + ctx.bus.subscribe(EventType.PLAN_CREATED, "not-callable") # type: ignore[arg-type] + except TypeError: + raised = True + assert raised, "Expected TypeError for non-callable handler" + + +@then("the EventBus Protocol stubs should be callable") +def step_protocol_stubs_callable(ctx: Context) -> None: + """Exercise Protocol method stubs directly to ensure 100% coverage.""" + + class _BareImpl(EventBus): # type: ignore[misc] + pass + + obj = _BareImpl() + event = _make_event("plan.created") + obj.emit(event) + obj.subscribe(EventType.PLAN_CREATED, lambda e: None) + + +# --------------------------------------------------------------------------- +# DecisionService event emission steps +# --------------------------------------------------------------------------- + + +@given("a DecisionService with an injected event bus") +def step_given_decision_service_with_bus(ctx: Context) -> None: + ctx.all_events: list[DomainEvent] = [] + ctx.event_bus = ReactiveEventBus() + ctx.event_bus.subscribe(EventType.DECISION_CREATED, ctx.all_events.append) + + mock_uow = MagicMock() + mock_uow.transaction.return_value.__enter__ = MagicMock( + return_value=MagicMock(decisions=MagicMock()) + ) + mock_uow.transaction.return_value.__exit__ = MagicMock(return_value=False) + + ctx.decision_service = DecisionService( + settings=MagicMock(), + unit_of_work=mock_uow, + event_bus=ctx.event_bus, + ) + ctx.test_plan_id = str(ULID()) + ctx.test_decision = Decision( + plan_id=ctx.test_plan_id, + sequence_number=0, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Which approach?", + chosen_option="TDD", + rationale="Test-first", + ) + + +@given("a DecisionService without an event bus") +def step_given_decision_service_no_bus(ctx: Context) -> None: + mock_uow = MagicMock() + mock_uow.transaction.return_value.__enter__ = MagicMock( + return_value=MagicMock(decisions=MagicMock()) + ) + mock_uow.transaction.return_value.__exit__ = MagicMock(return_value=False) + + ctx.decision_service = DecisionService( + settings=MagicMock(), + unit_of_work=mock_uow, + ) + ctx.test_decision = Decision( + plan_id=str(ULID()), + sequence_number=0, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Which approach?", + chosen_option="TDD", + rationale="Test-first", + ) + + +@when("I call record_decision") +def step_call_record_decision(ctx: Context) -> None: + ctx.exception = None + try: + ctx.decision_service.record_decision(ctx.test_decision) + except Exception as exc: + ctx.exception = exc + + +@then('the event bus should have received a "{et}" event') +def step_event_bus_received(ctx: Context, et: str) -> None: + all_events: list[DomainEvent] = getattr(ctx, "all_events", []) + received_types = [str(e.event_type) for e in all_events] + assert et in received_types, ( + f"Expected {et!r} in received events, got {received_types}" + ) + + +@then("the emitted event plan_id should match the decision plan_id") +def step_emitted_plan_id_matches(ctx: Context) -> None: + all_events: list[DomainEvent] = getattr(ctx, "all_events", []) + assert all_events, "No events received" + emitted_plan_id = all_events[0].plan_id + assert emitted_plan_id == ctx.test_plan_id, ( + f"Expected plan_id {ctx.test_plan_id!r}, got {emitted_plan_id!r}" + ) + + +@then("no event bus exception should be raised") +def step_no_event_bus_exception(ctx: Context) -> None: + assert ctx.exception is None, f"Unexpected exception: {ctx.exception}" + + +# --------------------------------------------------------------------------- +# PlanLifecycleService event emission steps +# --------------------------------------------------------------------------- + + +def _make_settings_stub() -> MagicMock: + s = MagicMock() + s.database_url = "sqlite:///:memory:" + return s + + +@given("a PlanLifecycleService with an injected event bus") +def step_given_lifecycle_service_with_bus(ctx: Context) -> None: + ctx.all_events: list[DomainEvent] = [] + ctx.event_bus = ReactiveEventBus() + ctx.event_bus.subscribe(EventType.PLAN_CREATED, ctx.all_events.append) + + ctx.lifecycle_service = PlanLifecycleService( + settings=_make_settings_stub(), + event_bus=ctx.event_bus, + ) + ctx.exception = None + + +@given("a PlanLifecycleService with an injected event bus and a strategized plan") +def step_given_lifecycle_with_strategized_plan(ctx: Context) -> None: + ctx.all_events: list[DomainEvent] = [] + ctx.event_bus = ReactiveEventBus() + ctx.event_bus.subscribe(EventType.PLAN_PHASE_CHANGED, ctx.all_events.append) + + ctx.lifecycle_service = PlanLifecycleService( + settings=_make_settings_stub(), + event_bus=ctx.event_bus, + ) + ctx.exception = None + + ctx.lifecycle_service.create_action( + name="test/action", + description="benchmark action", + definition_of_done="Done", + strategy_actor="local/strategist", + execution_actor="local/executor", + ) + plan = ctx.lifecycle_service.use_action(action_name="test/action") + plan_id = plan.identity.plan_id + ctx.lifecycle_service._plans[plan_id] = plan.model_copy( + update={ + "phase": PlanPhase.STRATEGIZE, + "processing_state": ProcessingState.COMPLETE, + } + ) + ctx.test_plan_id = plan_id + + +@given("a PlanLifecycleService without an event bus") +def step_given_lifecycle_service_no_bus(ctx: Context) -> None: + ctx.lifecycle_service = PlanLifecycleService( + settings=_make_settings_stub(), + ) + ctx.exception = None + + +@when("I call use_action to create a plan") +def step_call_use_action(ctx: Context) -> None: + try: + ctx.lifecycle_service.create_action( + name="test/action", + description="test action", + definition_of_done="Done when implemented", + strategy_actor="local/strategist", + execution_actor="local/executor", + ) + ctx.lifecycle_service.use_action(action_name="test/action") + except Exception as exc: + ctx.exception = exc + + +@when("I call execute_plan") +def step_call_execute_plan(ctx: Context) -> None: + try: + ctx.lifecycle_service.execute_plan(ctx.test_plan_id) + except Exception as exc: + ctx.exception = exc + + +# --------------------------------------------------------------------------- +# DI container steps +# --------------------------------------------------------------------------- + + +@when("I resolve event_bus from the DI container twice") +def step_resolve_event_bus_twice(ctx: Context) -> None: + container = Container() + ctx.event_bus_a = container.event_bus() + ctx.event_bus_b = container.event_bus() + + +@then("both resolutions should return the same instance") +def step_same_instance(ctx: Context) -> None: + assert ctx.event_bus_a is ctx.event_bus_b, ( + "event_bus Singleton returned different instances" + ) + + +@when("I resolve decision_service from the DI container") +def step_resolve_decision_service(ctx: Context) -> None: + container = Container() + ctx.resolved_decision_service = container.decision_service() + + +@then("the decision_service should have an event_bus attribute") +def step_decision_service_has_event_bus(ctx: Context) -> None: + svc = ctx.resolved_decision_service + assert hasattr(svc, "event_bus"), ( + "DecisionService resolved from DI has no event_bus attribute" + ) + assert svc.event_bus is not None, "DecisionService.event_bus should not be None" diff --git a/robot/event_bus.robot b/robot/event_bus.robot new file mode 100644 index 00000000..96ae6aaf --- /dev/null +++ b/robot/event_bus.robot @@ -0,0 +1,91 @@ +*** Settings *** +Documentation Smoke tests for EventBus infrastructure (EventType, DomainEvent, +... ReactiveEventBus, LoggingEventBus, service emission, DI container) +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} ${CURDIR}/helper_event_bus.py + +*** Test Cases *** +EventType Values Are Present + [Documentation] Verify all required EventType string values exist + ${result}= Run Process ${PYTHON} ${HELPER} event_type_values + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} event-type-values-ok + +DomainEvent Creation With Defaults + [Documentation] Verify DomainEvent constructs and applies default fields + ${result}= Run Process ${PYTHON} ${HELPER} domain_event_creation + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} domain-event-creation-ok + +Protocol Conformance + [Documentation] Verify both bus implementations satisfy EventBus protocol + ${result}= Run Process ${PYTHON} ${HELPER} protocol_conformance + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} protocol-conformance-ok + +ReactiveEventBus Emit And Subscribe + [Documentation] Verify ReactiveEventBus dispatches events to handlers + ${result}= Run Process ${PYTHON} ${HELPER} reactive_emit_subscribe + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} reactive-emit-subscribe-ok + +LoggingEventBus Emit And Subscribe + [Documentation] Verify LoggingEventBus dispatches events to handlers + ${result}= Run Process ${PYTHON} ${HELPER} logging_emit_subscribe + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} logging-emit-subscribe-ok + +ReactiveEventBus Type Filtering + [Documentation] Verify ReactiveEventBus only dispatches to matching event type + ${result}= Run Process ${PYTHON} ${HELPER} reactive_type_filtering + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} reactive-type-filtering-ok + +DecisionService Emits DECISION_CREATED + [Documentation] Verify DecisionService emits event via injected bus + ${result}= Run Process ${PYTHON} ${HELPER} decision_service_emits_event + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} decision-service-emits-event-ok + +PlanLifecycleService Emits PLAN_CREATED + [Documentation] Verify PlanLifecycleService emits event via injected bus + ${result}= Run Process ${PYTHON} ${HELPER} plan_lifecycle_emits_event + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} plan-lifecycle-emits-event-ok + +Container EventBus Is Singleton + [Documentation] Verify DI container provides same ReactiveEventBus instance + ${result}= Run Process ${PYTHON} ${HELPER} container_event_bus_singleton + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} container-event-bus-singleton-ok diff --git a/robot/helper_event_bus.py b/robot/helper_event_bus.py new file mode 100644 index 00000000..ad3ebee2 --- /dev/null +++ b/robot/helper_event_bus.py @@ -0,0 +1,203 @@ +"""Helper script for event_bus.robot smoke tests. + +Each subcommand is a self-contained check that prints a sentinel on success. +""" + +from __future__ import annotations + +import sys +from pathlib import Path + +# Ensure local source tree is importable +_SRC = str(Path(__file__).resolve().parents[1] / "src") +if _SRC not in sys.path: + sys.path.insert(0, _SRC) + +from cleveragents.infrastructure.events import ( # noqa: E402 + DomainEvent, + EventBus, + EventType, + LoggingEventBus, + ReactiveEventBus, +) + +# --------------------------------------------------------------------------- +# Subcommands +# --------------------------------------------------------------------------- + + +def event_type_values() -> None: + """Verify key EventType values are present.""" + required = { + "plan.created", + "decision.created", + "tool.invoked", + "actor.completed", + "budget.exceeded", + } + actual = {e.value for e in EventType} + missing = required - actual + if missing: + print(f"FAIL: missing EventType values: {missing}", file=sys.stderr) + sys.exit(1) + print("event-type-values-ok") + + +def domain_event_creation() -> None: + """Verify DomainEvent constructs with defaults.""" + event = DomainEvent(event_type=EventType.PLAN_CREATED) + if event.plan_id is not None: + print("FAIL: expected plan_id=None", file=sys.stderr) + sys.exit(1) + if not event.correlation_id: + print("FAIL: expected non-empty correlation_id", file=sys.stderr) + sys.exit(1) + print("domain-event-creation-ok") + + +def protocol_conformance() -> None: + """Verify both bus implementations satisfy EventBus protocol.""" + for cls in (ReactiveEventBus, LoggingEventBus): + bus = cls() + if not isinstance(bus, EventBus): + print(f"FAIL: {cls.__name__} does not satisfy EventBus", file=sys.stderr) + sys.exit(1) + print("protocol-conformance-ok") + + +def reactive_emit_subscribe() -> None: + """Verify ReactiveEventBus dispatches events to handlers.""" + received: list[DomainEvent] = [] + bus = ReactiveEventBus() + bus.subscribe(EventType.PLAN_CREATED, received.append) + bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED)) + if len(received) != 1: + print(f"FAIL: expected 1 event, got {len(received)}", file=sys.stderr) + sys.exit(1) + if str(received[0].event_type) != "plan.created": + print(f"FAIL: wrong event type: {received[0].event_type}", file=sys.stderr) + sys.exit(1) + print("reactive-emit-subscribe-ok") + + +def logging_emit_subscribe() -> None: + """Verify LoggingEventBus dispatches events to handlers.""" + received: list[DomainEvent] = [] + bus = LoggingEventBus() + bus.subscribe(EventType.DECISION_CREATED, received.append) + bus.emit(DomainEvent(event_type=EventType.DECISION_CREATED)) + if len(received) != 1: + print(f"FAIL: expected 1 event, got {len(received)}", file=sys.stderr) + sys.exit(1) + print("logging-emit-subscribe-ok") + + +def reactive_type_filtering() -> None: + """Verify ReactiveEventBus only dispatches to matching event type.""" + received: list[DomainEvent] = [] + bus = ReactiveEventBus() + bus.subscribe(EventType.PLAN_CREATED, received.append) + bus.emit(DomainEvent(event_type=EventType.DECISION_CREATED)) + if received: + print("FAIL: handler should not have been called", file=sys.stderr) + sys.exit(1) + print("reactive-type-filtering-ok") + + +def decision_service_emits_event() -> None: + """Verify DecisionService emits DECISION_CREATED via injected bus.""" + from unittest.mock import MagicMock + + from ulid import ULID + + from cleveragents.application.services.decision_service import DecisionService + from cleveragents.domain.models.core.decision import Decision, DecisionType + + received: list[DomainEvent] = [] + bus = ReactiveEventBus() + bus.subscribe(EventType.DECISION_CREATED, received.append) + + mock_uow = MagicMock() + mock_uow.transaction.return_value.__enter__ = MagicMock( + return_value=MagicMock(decisions=MagicMock()) + ) + mock_uow.transaction.return_value.__exit__ = MagicMock(return_value=False) + + svc = DecisionService(settings=MagicMock(), unit_of_work=mock_uow, event_bus=bus) + decision = Decision( + plan_id=str(ULID()), + sequence_number=0, + decision_type=DecisionType.STRATEGY_CHOICE, + question="Q?", + chosen_option="A", + rationale="because", + ) + svc.record_decision(decision) + if not received: + print("FAIL: no DECISION_CREATED event emitted", file=sys.stderr) + sys.exit(1) + print("decision-service-emits-event-ok") + + +def plan_lifecycle_emits_event() -> None: + """Verify PlanLifecycleService emits PLAN_CREATED via injected bus.""" + from unittest.mock import MagicMock + + from cleveragents.application.services.plan_lifecycle_service import ( + PlanLifecycleService, + ) + + received: list[DomainEvent] = [] + bus = ReactiveEventBus() + bus.subscribe(EventType.PLAN_CREATED, received.append) + + svc = PlanLifecycleService(settings=MagicMock(), event_bus=bus) + svc.create_action( + name="test/action", + description="test action", + definition_of_done="Done when implemented", + strategy_actor="local/strategist", + execution_actor="local/executor", + ) + svc.use_action(action_name="test/action") + if not received: + print("FAIL: no PLAN_CREATED event emitted", file=sys.stderr) + sys.exit(1) + print("plan-lifecycle-emits-event-ok") + + +def container_event_bus_singleton() -> None: + """Verify DI container provides ReactiveEventBus as Singleton.""" + from cleveragents.application.container import Container + + container = Container() + bus_a = container.event_bus() + bus_b = container.event_bus() + if bus_a is not bus_b: + print("FAIL: event_bus not a singleton", file=sys.stderr) + sys.exit(1) + print("container-event-bus-singleton-ok") + + +# --------------------------------------------------------------------------- +# Dispatch +# --------------------------------------------------------------------------- + +_COMMANDS = { + "event_type_values": event_type_values, + "domain_event_creation": domain_event_creation, + "protocol_conformance": protocol_conformance, + "reactive_emit_subscribe": reactive_emit_subscribe, + "logging_emit_subscribe": logging_emit_subscribe, + "reactive_type_filtering": reactive_type_filtering, + "decision_service_emits_event": decision_service_emits_event, + "plan_lifecycle_emits_event": plan_lifecycle_emits_event, + "container_event_bus_singleton": container_event_bus_singleton, +} + +if __name__ == "__main__": + if len(sys.argv) < 2 or sys.argv[1] not in _COMMANDS: + print(f"Usage: {sys.argv[0]} ", file=sys.stderr) + print(f"Commands: {list(_COMMANDS)}", file=sys.stderr) + sys.exit(1) + _COMMANDS[sys.argv[1]]() diff --git a/src/cleveragents/application/container.py b/src/cleveragents/application/container.py index 08df9ad4..b8f17d8b 100644 --- a/src/cleveragents/application/container.py +++ b/src/cleveragents/application/container.py @@ -32,6 +32,7 @@ from cleveragents.infrastructure.database.repositories import ( ProjectResourceLinkRepository, ) from cleveragents.infrastructure.database.unit_of_work import UnitOfWork +from cleveragents.infrastructure.events.reactive import ReactiveEventBus from cleveragents.langgraph.bridge import RxPyLangGraphBridge from cleveragents.providers.registry import ProviderRegistry, get_provider_registry from cleveragents.reactive.route_bridge import RouteBridge @@ -178,6 +179,9 @@ class Container(containers.DeclarativeContainer): provider_registry=provider_registry, ) + # Event Bus - Singleton (one shared instance for the process) + event_bus = providers.Singleton(ReactiveEventBus) + # Services - Factory (new instance per request with injected dependencies) project_service = providers.Factory( ProjectService, @@ -225,6 +229,7 @@ class Container(containers.DeclarativeContainer): DecisionService, settings=settings, unit_of_work=unit_of_work, + event_bus=event_bus, ) # Plan Lifecycle Service - Factory (v3 four-phase lifecycle) @@ -233,6 +238,7 @@ class Container(containers.DeclarativeContainer): settings=settings, unit_of_work=unit_of_work, decision_service=decision_service, + event_bus=event_bus, ) # Resource Registry Service - uses database session factory from UoW diff --git a/src/cleveragents/application/services/decision_service.py b/src/cleveragents/application/services/decision_service.py index 59fd951f..525eddd0 100644 --- a/src/cleveragents/application/services/decision_service.py +++ b/src/cleveragents/application/services/decision_service.py @@ -18,9 +18,12 @@ from typing import TYPE_CHECKING import structlog from cleveragents.domain.models.core.decision import Decision +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.types import EventType if TYPE_CHECKING: from cleveragents.infrastructure.database.unit_of_work import UnitOfWork + from cleveragents.infrastructure.events.protocol import EventBus logger = structlog.get_logger(__name__) @@ -37,15 +40,21 @@ class DecisionService: consistency with the service constructor convention). unit_of_work: A :class:`UnitOfWork` instance used to open transactional scopes. + event_bus: Optional + :class:`~cleveragents.infrastructure.events.protocol.EventBus` + for emitting ``DECISION_CREATED`` events. When ``None``, + event emission is silently skipped. """ def __init__( self, settings: object, unit_of_work: UnitOfWork, + event_bus: EventBus | None = None, ) -> None: self.settings = settings self.unit_of_work = unit_of_work + self.event_bus = event_bus self._logger = logger.bind(service="decision") # ------------------------------------------------------------------ @@ -81,6 +90,18 @@ class DecisionService: "decision_recorded", decision_id=decision.decision_id, ) + if self.event_bus is not None: + self.event_bus.emit( + DomainEvent( + event_type=EventType.DECISION_CREATED, + plan_id=decision.plan_id, + details={ + "decision_id": decision.decision_id, + "decision_type": str(decision.decision_type), + "sequence_number": decision.sequence_number, + }, + ) + ) return decision # ------------------------------------------------------------------ diff --git a/src/cleveragents/application/services/plan_lifecycle_service.py b/src/cleveragents/application/services/plan_lifecycle_service.py index 194bc089..6d879061 100644 --- a/src/cleveragents/application/services/plan_lifecycle_service.py +++ b/src/cleveragents/application/services/plan_lifecycle_service.py @@ -80,11 +80,14 @@ from cleveragents.domain.models.core.plan import ( ProjectLink, can_transition, ) +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.types import EventType if TYPE_CHECKING: from cleveragents.application.services.decision_service import DecisionService from cleveragents.config.settings import Settings from cleveragents.infrastructure.database.unit_of_work import UnitOfWork + from cleveragents.infrastructure.events.protocol import EventBus logger = structlog.get_logger(__name__) @@ -149,6 +152,7 @@ class PlanLifecycleService: settings: Settings, unit_of_work: UnitOfWork | None = None, decision_service: DecisionService | None = None, + event_bus: EventBus | None = None, ): """Initialize the plan lifecycle service. @@ -163,10 +167,15 @@ class PlanLifecycleService: automatically recording decisions during phase transitions. When ``None``, decision recording is silently skipped. + event_bus: Optional + :class:`~cleveragents.infrastructure.events.protocol.EventBus` + for emitting plan lifecycle events. When ``None``, + event emission is silently skipped. """ self.settings = settings self.unit_of_work = unit_of_work self.decision_service = decision_service + self.event_bus = event_bus self._logger = logger.bind(service="plan_lifecycle") # In-memory fallback storage (used only when no UoW is provided) @@ -635,6 +644,17 @@ class PlanLifecycleService: action_name=action_full_name, phase=plan.phase.value, ) + if self.event_bus is not None: + self.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_CREATED, + plan_id=plan_id, + details={ + "action_name": action_full_name, + "phase": plan.phase.value, + }, + ) + ) return plan @@ -844,6 +864,19 @@ class PlanLifecycleService: plan_id=plan_id, phase=plan.phase.value, ) + if self.event_bus is not None: + self.event_bus.emit( + DomainEvent( + event_type=EventType.PLAN_PHASE_CHANGED, + plan_id=plan_id, + details={ + "phase": plan.phase.value, + "processing_state": plan.processing_state.value + if plan.processing_state + else None, + }, + ) + ) return plan diff --git a/src/cleveragents/infrastructure/events/__init__.py b/src/cleveragents/infrastructure/events/__init__.py new file mode 100644 index 00000000..f05d55c8 --- /dev/null +++ b/src/cleveragents/infrastructure/events/__init__.py @@ -0,0 +1,37 @@ +"""Domain event bus infrastructure. + +This package provides: + +- :class:`EventType` — enumeration of all domain event types. +- :class:`DomainEvent` — base Pydantic model for all domain events. +- :class:`EventBus` — structural Protocol that implementations satisfy. +- :class:`ReactiveEventBus` — RxPY-backed in-process bus. +- :class:`LoggingEventBus` — structured-log-only bus. + +Typical usage:: + + from cleveragents.infrastructure.events import ( + DomainEvent, + EventBus, + EventType, + ReactiveEventBus, + ) + + bus = ReactiveEventBus() + bus.subscribe(EventType.DECISION_CREATED, lambda e: print(e)) + bus.emit(DomainEvent(event_type=EventType.DECISION_CREATED)) +""" + +from cleveragents.infrastructure.events.logging_bus import LoggingEventBus +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.protocol import EventBus +from cleveragents.infrastructure.events.reactive import ReactiveEventBus +from cleveragents.infrastructure.events.types import EventType + +__all__ = [ + "DomainEvent", + "EventBus", + "EventType", + "LoggingEventBus", + "ReactiveEventBus", +] diff --git a/src/cleveragents/infrastructure/events/logging_bus.py b/src/cleveragents/infrastructure/events/logging_bus.py new file mode 100644 index 00000000..6962eacc --- /dev/null +++ b/src/cleveragents/infrastructure/events/logging_bus.py @@ -0,0 +1,96 @@ +"""LoggingEventBus — structured-logging event bus implementation. + +Emits every :class:`~cleveragents.infrastructure.events.models.DomainEvent` +to structured logs via :mod:`structlog`. Unlike +:class:`~cleveragents.infrastructure.events.reactive.ReactiveEventBus` this +implementation requires no RxPY dependency and is useful for audit trails, +testing, and environments where reactive streaming is not needed. + +Based on: + - docs/specification.md §Event-Driven Architecture + - Forgejo issue #473 +""" + +from __future__ import annotations + +from collections.abc import Callable + +import structlog + +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.types import EventType + +_logger = structlog.get_logger(__name__) + + +class LoggingEventBus: + """Event bus that writes every event to structured logs. + + Events are logged at ``INFO`` level with all :class:`DomainEvent` + fields included as structured key-value pairs. Type-specific + handlers registered via :meth:`subscribe` are also invoked + synchronously after logging. + + This implementation satisfies the + :class:`~cleveragents.infrastructure.events.protocol.EventBus` + protocol and can be used as a drop-in replacement for + :class:`~cleveragents.infrastructure.events.reactive.ReactiveEventBus` + in environments where RxPY is undesirable. + """ + + def __init__(self) -> None: + self._subscriptions: dict[EventType, list[Callable[[DomainEvent], None]]] = {} + + # ------------------------------------------------------------------ + # Public API (satisfies EventBus protocol) + # ------------------------------------------------------------------ + + def emit(self, event: DomainEvent) -> None: + """Log *event* and dispatch to type-specific handlers. + + Args: + event: The :class:`DomainEvent` to emit. + + Raises: + TypeError: If *event* is not a :class:`DomainEvent`. + """ + if not isinstance(event, DomainEvent): + raise TypeError( + f"event must be a DomainEvent, got {type(event).__name__!r}" + ) + _logger.info( + "domain_event", + event_type=str(event.event_type), + correlation_id=event.correlation_id, + plan_id=event.plan_id, + actor_name=event.actor_name, + details=event.details, + ) + for handler in self._subscriptions.get(event.event_type, []): + handler(event) + + def subscribe( + self, + event_type: EventType, + handler: Callable[[DomainEvent], None], + ) -> None: + """Register *handler* for events of *event_type*. + + Args: + event_type: The :class:`EventType` to listen for. + handler: Callable invoked for each matching :class:`DomainEvent`. + + Raises: + TypeError: If *event_type* is not an :class:`EventType`. + TypeError: If *handler* is not callable. + """ + if not isinstance(event_type, EventType): + raise TypeError( + f"event_type must be an EventType, got {type(event_type).__name__!r}" + ) + if not callable(handler): + raise TypeError("handler must be callable") + self._subscriptions.setdefault(event_type, []).append(handler) + + +__all__ = ["LoggingEventBus"] diff --git a/src/cleveragents/infrastructure/events/models.py b/src/cleveragents/infrastructure/events/models.py new file mode 100644 index 00000000..1b3b89c6 --- /dev/null +++ b/src/cleveragents/infrastructure/events/models.py @@ -0,0 +1,68 @@ +"""DomainEvent base model. + +All domain events emitted through an +:class:`~cleveragents.infrastructure.events.protocol.EventBus` must be +instances of :class:`DomainEvent` (or a subclass). + +Based on: + - docs/specification.md §Event-Driven Architecture + - Forgejo issue #473 +""" + +from __future__ import annotations + +from datetime import UTC, datetime +from typing import Any + +from pydantic import BaseModel, ConfigDict, Field +from ulid import ULID + +from cleveragents.infrastructure.events.types import EventType + + +class DomainEvent(BaseModel): + """Base event model emitted by all domain operations. + + Attributes: + event_type: The type of domain event. + timestamp: UTC datetime when the event was created. + correlation_id: Unique ULID that links related events in a + single request/operation chain. + plan_id: ULID of the plan associated with this event, if any. + root_plan_id: ULID of the top-level plan in a hierarchy, + if any. + session_id: ULID of the session, if any. + actor_name: Name of the actor that emitted the event, if any. + project_name: Name of the project context, if any. + details: Event-type-specific payload (arbitrary key/value pairs). + """ + + model_config = ConfigDict(frozen=True) + + event_type: EventType + timestamp: datetime = Field( + default_factory=lambda: datetime.now(tz=UTC), + description="UTC datetime of event creation.", + ) + correlation_id: str = Field( + default_factory=lambda: str(ULID()), + description="Unique ULID linking related events.", + ) + plan_id: str | None = Field(default=None, description="Associated plan ULID.") + root_plan_id: str | None = Field( + default=None, description="Root plan ULID in a hierarchy." + ) + session_id: str | None = Field(default=None, description="Associated session ULID.") + actor_name: str | None = Field( + default=None, description="Actor that emitted this event." + ) + project_name: str | None = Field( + default=None, description="Project context of this event." + ) + details: dict[str, Any] = Field( + default_factory=dict, + description="Event-type-specific payload.", + ) + + +__all__ = ["DomainEvent"] diff --git a/src/cleveragents/infrastructure/events/protocol.py b/src/cleveragents/infrastructure/events/protocol.py new file mode 100644 index 00000000..2c3cae85 --- /dev/null +++ b/src/cleveragents/infrastructure/events/protocol.py @@ -0,0 +1,53 @@ +"""EventBus Protocol definition. + +Defines the :class:`EventBus` structural interface that all event-bus +implementations must satisfy. Domain services depend only on this +protocol, never on concrete implementations, keeping the domain layer +decoupled from infrastructure choices. + +Based on: + - docs/specification.md §Event-Driven Architecture + - Forgejo issue #473 +""" + +from __future__ import annotations + +from collections.abc import Callable +from typing import Protocol, runtime_checkable + +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.types import EventType + + +@runtime_checkable +class EventBus(Protocol): + """Structural interface for emitting and subscribing to domain events. + + Implementations may be reactive (RxPY), logging-only, or in-memory + stubs. Callers (domain services) depend on this protocol and receive + the concrete implementation through dependency injection. + """ + + def emit(self, event: DomainEvent) -> None: + """Publish *event* to all registered subscribers. + + Args: + event: The :class:`DomainEvent` to publish. + """ + ... + + def subscribe( + self, + event_type: EventType, + handler: Callable[[DomainEvent], None], + ) -> None: + """Register *handler* to receive events of *event_type*. + + Args: + event_type: The :class:`EventType` to listen for. + handler: Callable invoked synchronously for each matching event. + """ + ... + + +__all__ = ["EventBus"] diff --git a/src/cleveragents/infrastructure/events/reactive.py b/src/cleveragents/infrastructure/events/reactive.py new file mode 100644 index 00000000..55b28b9e --- /dev/null +++ b/src/cleveragents/infrastructure/events/reactive.py @@ -0,0 +1,100 @@ +"""ReactiveEventBus — in-process event bus backed by RxPY. + +Uses an RxPY :class:`rx.subject.Subject` as the hot observable backbone. +Every :meth:`emit` call: + +1. Pushes the event into the reactive stream (for advanced RxPY operators). +2. Dispatches synchronously to any type-specific handlers registered via + :meth:`subscribe`. + +This dual-dispatch model means plain callback-based subscribers work without +any RxPY knowledge, while subscribers that want filtering, buffering, or +debouncing can operate directly on :attr:`stream`. + +Based on: + - docs/specification.md §Event-Driven Architecture + - Forgejo issue #473 +""" + +from __future__ import annotations + +from collections.abc import Callable + +from rx.core.observable.observable import Observable +from rx.subject.subject import Subject + +from cleveragents.infrastructure.events.models import DomainEvent +from cleveragents.infrastructure.events.types import EventType + + +class ReactiveEventBus: + """In-process event bus using RxPY for real-time distribution. + + Attributes: + _subject: Hot RxPY Subject — the raw observable stream. + _subscriptions: Per-event-type handler lists. + """ + + def __init__(self) -> None: + self._subject: Subject = Subject() + self._subscriptions: dict[EventType, list[Callable[[DomainEvent], None]]] = {} + + # ------------------------------------------------------------------ + # Public API (satisfies EventBus protocol) + # ------------------------------------------------------------------ + + def emit(self, event: DomainEvent) -> None: + """Publish *event* to the reactive stream and all type handlers. + + Args: + event: The :class:`DomainEvent` to publish. + + Raises: + TypeError: If *event* is not a :class:`DomainEvent`. + """ + if not isinstance(event, DomainEvent): + raise TypeError( + f"event must be a DomainEvent, got {type(event).__name__!r}" + ) + self._subject.on_next(event) + for handler in self._subscriptions.get(event.event_type, []): + handler(event) + + def subscribe( + self, + event_type: EventType, + handler: Callable[[DomainEvent], None], + ) -> None: + """Register *handler* for events of *event_type*. + + Handlers are called synchronously inside :meth:`emit` in the order + they were registered. + + Args: + event_type: The :class:`EventType` to listen for. + handler: Callable invoked for each matching :class:`DomainEvent`. + + Raises: + TypeError: If *event_type* is not an :class:`EventType`. + TypeError: If *handler* is not callable. + """ + if not isinstance(event_type, EventType): + raise TypeError( + f"event_type must be an EventType, got {type(event_type).__name__!r}" + ) + if not callable(handler): + raise TypeError("handler must be callable") + self._subscriptions.setdefault(event_type, []).append(handler) + + @property + def stream(self) -> Observable: + """Raw observable stream for advanced RxPY operators. + + Returns: + The underlying :class:`rx.subject.Subject` cast as an + :class:`~rx.core.observable.observable.Observable`. + """ + return self._subject + + +__all__ = ["ReactiveEventBus"] diff --git a/src/cleveragents/infrastructure/events/types.py b/src/cleveragents/infrastructure/events/types.py new file mode 100644 index 00000000..f3d391bb --- /dev/null +++ b/src/cleveragents/infrastructure/events/types.py @@ -0,0 +1,86 @@ +"""Domain event type enumeration. + +Defines the full set of :class:`EventType` values that domain services +may emit through an :class:`~cleveragents.infrastructure.events.protocol.EventBus`. + +Based on: + - docs/specification.md §Event-Driven Architecture + - Forgejo issue #473 +""" + +from __future__ import annotations + +from enum import StrEnum + + +class EventType(StrEnum): + """Typed identifiers for every domain event the system can emit. + + Values are dot-separated strings following the convention + ``.`` (e.g. ``plan.created``). Using :class:`StrEnum` + means an :class:`EventType` member compares equal to its string + representation, which simplifies JSON serialisation and log filtering. + """ + + # --- Plan lifecycle --- + PLAN_CREATED = "plan.created" + PLAN_PHASE_CHANGED = "plan.phase_changed" + PLAN_STATE_CHANGED = "plan.state_changed" + PLAN_APPLIED = "plan.applied" + PLAN_CANCELLED = "plan.cancelled" + PLAN_ERRORED = "plan.errored" + + # --- Decision --- + DECISION_CREATED = "decision.created" + DECISION_APPROVED = "decision.approved" + DECISION_CORRECTED = "decision.corrected" + DECISION_SUPERSEDED = "decision.superseded" + + # --- Invariant --- + INVARIANT_RECONCILED = "invariant.reconciled" + INVARIANT_VIOLATED = "invariant.violated" + INVARIANT_ENFORCED = "invariant.enforced" + + # --- Actor --- + ACTOR_INVOKED = "actor.invoked" + ACTOR_COMPLETED = "actor.completed" + ACTOR_ERRORED = "actor.errored" + ACTOR_ESCALATED = "actor.escalated" + + # --- Tool --- + TOOL_INVOKED = "tool.invoked" + TOOL_COMPLETED = "tool.completed" + TOOL_ERRORED = "tool.errored" + TOOL_RETRIED = "tool.retried" + + # --- Resource --- + RESOURCE_ACCESSED = "resource.accessed" + RESOURCE_MODIFIED = "resource.modified" + RESOURCE_INDEXED = "resource.indexed" + + # --- Sandbox --- + SANDBOX_CREATED = "sandbox.created" + SANDBOX_COMMITTED = "sandbox.committed" + SANDBOX_ROLLED_BACK = "sandbox.rolled_back" + CHECKPOINT_CREATED = "checkpoint.created" + CHECKPOINT_RESTORED = "checkpoint.restored" + + # --- Context --- + CONTEXT_BUILT = "context.built" + CONTEXT_QUERY_EXECUTED = "context.query_executed" + + # --- Validation --- + VALIDATION_STARTED = "validation.started" + VALIDATION_PASSED = "validation.passed" + VALIDATION_FAILED = "validation.failed" + + # --- Session --- + SESSION_CREATED = "session.created" + SESSION_MESSAGE_SENT = "session.message_sent" + + # --- Budget / cost --- + BUDGET_WARNING = "budget.warning" + BUDGET_EXCEEDED = "budget.exceeded" + + +__all__ = ["EventType"]