diff --git a/CHANGELOG.md b/CHANGELOG.md index 060c49280..5a8b5b1cb 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -114,6 +114,17 @@ `database_url` resolves inside `CLEVERAGENTS_HOME`, not the current working directory. Includes Robot Framework integration tests with a helper script exercising the same resolution path via subprocess. (#1034) +- Added volatile in-memory `audit_log` to `ReactiveEventBus` — every emitted + `DomainEvent` is appended to a volatile in-memory log accessible via the + `audit_log` property (defensive copy). Emit ordering now follows the + specification: RxPY stream push, then audit append, then handler dispatch. + Reactive and logging event buses now isolate stream/handler failures so one + subscriber cannot block audit recording or downstream subscribers. The + reactive stream now returns a read-only observable view (preventing direct + `on_next()` bypass), and `ReactiveEventBus` supports explicit in-memory + retention controls via `max_audit_log_size` and `clear_audit_log()`. Includes + expanded Behave and Robot integration coverage, 5 ASV benchmark suites, and + `vulture_whitelist.py` entry. (#587) - Added BuiltinAdapter class and MCP automatic resource slot creation. BuiltinAdapter wraps register_file_tools/register_git_tools/register_subplan_tool into a unified adapter interface. McpAdapter.infer_resource_slots() analyzes diff --git a/benchmarks/bench_event_bus.py b/benchmarks/bench_event_bus.py new file mode 100644 index 000000000..3da4268e7 --- /dev/null +++ b/benchmarks/bench_event_bus.py @@ -0,0 +1,174 @@ +"""ASV benchmarks for Event System Domain Event Taxonomy. + +Measures: +- DomainEvent construction throughput (minimal, with plan_id, with details) +- ReactiveEventBus emit with audit_log overhead +- ReactiveEventBus subscribe registration +- Fan-out: emit to N subscribers with audit persistence +- Audit log retrieval cost +- Log correlation (filtering events by correlation_id) +""" + +from __future__ import annotations + +import importlib +import logging +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 TaxonomyDomainEventSuite: + """Benchmark DomainEvent construction and serialisation.""" + + 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_all_fields(self) -> None: + DomainEvent( + event_type=self.event_type, + plan_id=self.plan_id, + root_plan_id=self.plan_id, + session_id=self.plan_id, + actor_name="local/strategist", + project_name="my-project", + details={"phase": "strategize", "actor": "strategists/planner"}, + ) + + def time_json_roundtrip(self) -> None: + event = DomainEvent( + event_type=self.event_type, + plan_id=self.plan_id, + details={"phase": "execute"}, + ) + DomainEvent.model_validate_json(event.model_dump_json()) + + +# --------------------------------------------------------------------------- +# ReactiveEventBus emit with audit_log +# --------------------------------------------------------------------------- + + +class TaxonomyReactiveEmitSuite: + """Benchmark ReactiveEventBus.emit() with audit_log persistence.""" + + 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) + + self._received: list[DomainEvent] = [] + self.bus.subscribe(EventType.PLAN_CREATED, self._received.append) + + def time_emit_no_subscribers(self) -> None: + self._received.clear() + self.bus.clear_audit_log() + self.bus.emit(self.decision_event) + + def time_emit_with_one_subscriber(self) -> None: + self._received.clear() + self.bus.clear_audit_log() + self.bus.emit(self.event) + + def time_emit_100_events(self) -> None: + self._received.clear() + self.bus.clear_audit_log() + for _ in range(100): + self.bus.emit(self.event) + + +# --------------------------------------------------------------------------- +# Audit log retrieval +# --------------------------------------------------------------------------- + + +class TaxonomyAuditLogSuite: + """Benchmark audit_log property access (defensive copy).""" + + timeout = 60 + params: ClassVar[list[int]] = [100, 1000] + param_names: ClassVar[list[str]] = ["num_events"] + + def setup(self, num_events: int) -> None: + self.bus = ReactiveEventBus() + for _ in range(num_events): + self.bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED)) + + def time_audit_log_snapshot(self, num_events: int) -> None: + _ = self.bus.audit_log + + +# --------------------------------------------------------------------------- +# Fan-out with audit +# --------------------------------------------------------------------------- + + +class TaxonomyFanOutSuite: + """Benchmark event fan-out to multiple subscribers with audit persistence.""" + + 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.clear_audit_log() + self.bus.emit(self.event) + + +# --------------------------------------------------------------------------- +# LoggingEventBus emit +# --------------------------------------------------------------------------- + + +class TaxonomyLoggingEmitSuite: + """Benchmark LoggingEventBus.emit() overhead.""" + + timeout = 60 + + def setup(self) -> None: + logging.disable(logging.CRITICAL) + self.bus = LoggingEventBus() + self.event = DomainEvent(event_type=EventType.DECISION_CREATED) + + def teardown(self) -> None: + logging.disable(logging.NOTSET) + + 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 index 679d0d762..283a85f79 100644 --- a/docs/reference/event_bus.md +++ b/docs/reference/event_bus.md @@ -93,7 +93,7 @@ satisfies it: ```python from cleveragents.infrastructure.events import EventBus, ReactiveEventBus -bus = ReactiveEventBus() +bus = ReactiveEventBus(max_audit_log_size=5000) # optional retention cap assert isinstance(bus, EventBus) # True ``` @@ -124,7 +124,7 @@ 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 +# Use the read-only 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)) @@ -136,7 +136,8 @@ 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. +- `stream` exposes a read-only `rx.Observable` view for RxPY operators. +- `clear_audit_log()` clears retained in-memory events when needed. --- diff --git a/features/observability/event_system_taxonomy.feature b/features/observability/event_system_taxonomy.feature new file mode 100644 index 000000000..8725dc0e5 --- /dev/null +++ b/features/observability/event_system_taxonomy.feature @@ -0,0 +1,283 @@ +Feature: Event System Domain Event Taxonomy + As a platform operator + I want a complete domain event taxonomy with typed event types and a reactive bus + So that I can audit, observe, and react to all system events consistently + + # --------------------------------------------------------------------------- + # EventType enum completeness + # --------------------------------------------------------------------------- + + Scenario: EventType enum has at least 43 members + Then the EventType enum should have at least 43 members + + Scenario: EventType covers the Plan lifecycle domain + Then EventType should contain "plan.created" + And EventType should contain "plan.phase_changed" + And EventType should contain "plan.state_changed" + And EventType should contain "plan.applied" + And EventType should contain "plan.cancelled" + And EventType should contain "plan.errored" + + Scenario: EventType covers the Decision domain + Then EventType should contain "decision.created" + And EventType should contain "decision.approved" + And EventType should contain "decision.corrected" + And EventType should contain "decision.superseded" + + Scenario: EventType covers the Invariant domain + Then EventType should contain "invariant.reconciled" + And EventType should contain "invariant.violated" + And EventType should contain "invariant.enforced" + + Scenario: EventType covers the Actor domain + Then EventType should contain "actor.invoked" + And EventType should contain "actor.completed" + And EventType should contain "actor.errored" + And EventType should contain "actor.escalated" + + Scenario: EventType covers the Tool domain + Then EventType should contain "tool.invoked" + And EventType should contain "tool.completed" + And EventType should contain "tool.errored" + And EventType should contain "tool.retried" + + Scenario: EventType covers the Resource domain + Then EventType should contain "resource.accessed" + And EventType should contain "resource.modified" + And EventType should contain "resource.indexed" + + Scenario: EventType covers the Correction domain + Then EventType should contain "correction.applied" + + Scenario: EventType covers the Configuration domain + Then EventType should contain "config.changed" + + Scenario: EventType covers the Entity lifecycle domain + Then EventType should contain "entity.deleted" + + Scenario: EventType covers the Authentication domain + Then EventType should contain "auth.success" + And EventType should contain "auth.failure" + + Scenario: EventType covers the Sandbox domain + Then EventType should contain "sandbox.created" + And EventType should contain "sandbox.committed" + And EventType should contain "sandbox.rolled_back" + + Scenario: EventType covers the Checkpoint domain + Then EventType should contain "checkpoint.created" + And EventType should contain "checkpoint.restored" + + Scenario: EventType covers the Context domain + Then EventType should contain "context.built" + And EventType should contain "context.query_executed" + + Scenario: EventType covers the Validation domain + Then EventType should contain "validation.started" + And EventType should contain "validation.passed" + And EventType should contain "validation.failed" + + Scenario: EventType covers the Session domain + Then EventType should contain "session.created" + And EventType should contain "session.message_sent" + + Scenario: EventType covers the Budget domain + Then EventType should contain "budget.warning" + And EventType should contain "budget.exceeded" + + Scenario: EventType uses StrEnum with dot-separated values + Then every EventType value should match the pattern "." + + # --------------------------------------------------------------------------- + # DomainEvent Pydantic model + # --------------------------------------------------------------------------- + + Scenario: DomainEvent has all required fields + When I create a DomainEvent for taxonomy with event_type "plan.created" + Then the taxonomy event should have field "event_type" + And the taxonomy event should have field "timestamp" + And the taxonomy event should have field "correlation_id" + And the taxonomy event should have field "plan_id" + And the taxonomy event should have field "root_plan_id" + And the taxonomy event should have field "session_id" + And the taxonomy event should have field "actor_name" + And the taxonomy event should have field "project_name" + And the taxonomy event should have field "details" + + Scenario: DomainEvent correlation_id is a 26-char ULID string + When I create a DomainEvent for taxonomy with event_type "plan.created" + Then the taxonomy event correlation_id should be 26 characters long + + Scenario: DomainEvent timestamp is timezone-aware UTC + When I create a DomainEvent for taxonomy with event_type "plan.created" + Then the taxonomy event timestamp should be UTC + + Scenario: DomainEvent is frozen (immutable) + When I create a DomainEvent for taxonomy with event_type "plan.created" + Then mutating the taxonomy event should raise a validation error + + Scenario: DomainEvent serializes to JSON round-trip + When I create a DomainEvent for taxonomy with all fields populated + Then the taxonomy event should serialize to JSON and deserialize back identically + + Scenario: DomainEvent correlation fields accept ULID strings + When I create a DomainEvent for taxonomy with correlation fields + Then the taxonomy event plan_id should be set + And the taxonomy event root_plan_id should be set + And the taxonomy event session_id should be set + + Scenario: DomainEvent details dict preserves custom keys + When I create a DomainEvent for taxonomy with typed details + Then the taxonomy event details should contain the expected keys + + # --------------------------------------------------------------------------- + # EventBus Protocol + # --------------------------------------------------------------------------- + + Scenario: EventBus Protocol defines emit and subscribe + Then the EventBus Protocol should define method "emit" + And the EventBus Protocol should define method "subscribe" + + Scenario: ReactiveEventBus satisfies the EventBus Protocol + Then a ReactiveEventBus instance should satisfy the EventBus Protocol + + Scenario: LoggingEventBus satisfies the EventBus Protocol + Then a LoggingEventBus instance should satisfy the EventBus Protocol + + Scenario: LoggingEventBus emits to subscribed handlers + Given a fresh LoggingEventBus for taxonomy tests + When I logging-taxonomy-subscribe to "plan.created" events + And I logging-taxonomy-emit a "plan.created" DomainEvent + Then the logging taxonomy handler should have received 1 event(s) + + Scenario: LoggingEventBus continues dispatching when a handler raises + Given a fresh LoggingEventBus for taxonomy tests + When I logging-taxonomy-subscribe a failing handler and a normal handler to "plan.created" events + And I logging-taxonomy-emit a "plan.created" DomainEvent + Then the logging taxonomy normal handler should have received 1 event(s) + + Scenario: LoggingEventBus rejects non-DomainEvent on emit + Given a fresh LoggingEventBus for taxonomy tests + Then logging-taxonomy-emitting a non-DomainEvent should raise TypeError + + Scenario: LoggingEventBus rejects non-EventType on subscribe + Given a fresh LoggingEventBus for taxonomy tests + Then logging-taxonomy-subscribing with a non-EventType should raise TypeError + + Scenario: LoggingEventBus rejects non-callable handler on subscribe + Given a fresh LoggingEventBus for taxonomy tests + Then logging-taxonomy-subscribing with a non-callable should raise TypeError + + # --------------------------------------------------------------------------- + # ReactiveEventBus — emit / subscribe / stream + # --------------------------------------------------------------------------- + + Scenario: ReactiveEventBus emits to subscribed handlers + Given a fresh ReactiveEventBus for taxonomy tests + When I taxonomy-subscribe to "plan.created" events + And I taxonomy-emit a "plan.created" DomainEvent + Then the taxonomy handler should have received 1 event(s) + + Scenario: ReactiveEventBus does not dispatch to non-matching types + Given a fresh ReactiveEventBus for taxonomy tests + When I taxonomy-subscribe to "plan.created" events + And I taxonomy-emit a "decision.created" DomainEvent + Then the taxonomy handler should have received 0 event(s) + + Scenario: ReactiveEventBus supports multiple handlers per type + Given a fresh ReactiveEventBus for taxonomy tests + When I taxonomy-subscribe two handlers to "tool.invoked" events + And I taxonomy-emit a "tool.invoked" DomainEvent + Then each taxonomy handler should have received 1 event + + Scenario: ReactiveEventBus continues dispatching when a handler raises + Given a fresh ReactiveEventBus for taxonomy tests + When I taxonomy-subscribe a failing handler and a normal handler to "plan.created" events + And I taxonomy-emit a "plan.created" DomainEvent + Then the taxonomy normal handler should have received 1 event(s) + And the audit_log should contain 1 events + + Scenario: ReactiveEventBus exposes an rx.Observable stream + Given a fresh ReactiveEventBus for taxonomy tests + Then the taxonomy bus stream should be an Observable + + Scenario: ReactiveEventBus stream is read-only + Given a fresh ReactiveEventBus for taxonomy tests + Then the taxonomy bus stream should not expose method "on_next" + + Scenario: ReactiveEventBus observable stream receives emitted events + Given a fresh ReactiveEventBus for taxonomy tests + When I observe the taxonomy bus stream and emit a "plan.created" event + Then the stream observer should have captured the event + + Scenario: ReactiveEventBus continues when a stream observer raises + Given a fresh ReactiveEventBus for taxonomy tests + When I taxonomy-subscribe to "plan.created" events + And I observe taxonomy stream with a failing observer and emit a "plan.created" event + Then the taxonomy handler should have received 1 event(s) + And the audit_log should contain 1 events + + Scenario: ReactiveEventBus rejects non-DomainEvent on emit + Given a fresh ReactiveEventBus for taxonomy tests + Then taxonomy-emitting a non-DomainEvent should raise TypeError + + Scenario: ReactiveEventBus rejects non-EventType on subscribe + Given a fresh ReactiveEventBus for taxonomy tests + Then taxonomy-subscribing with a non-EventType should raise TypeError + + Scenario: ReactiveEventBus rejects non-callable handler on subscribe + Given a fresh ReactiveEventBus for taxonomy tests + Then taxonomy-subscribing with a non-callable should raise TypeError + + # --------------------------------------------------------------------------- + # In-memory event trail (audit_log) + # --------------------------------------------------------------------------- + + Scenario: ReactiveEventBus audit_log captures all emitted events + Given a fresh ReactiveEventBus for taxonomy tests + When I taxonomy-emit a "plan.created" DomainEvent + And I taxonomy-emit a "decision.created" DomainEvent + Then the audit_log should contain 2 events + And the audit_log should include event types "plan.created,decision.created" + + Scenario: ReactiveEventBus audit_log preserves event order + Given a fresh ReactiveEventBus for taxonomy tests + When I taxonomy-emit a "plan.created" DomainEvent + And I taxonomy-emit a "decision.created" DomainEvent + And I taxonomy-emit a "tool.invoked" DomainEvent + Then the audit_log events should be in order "plan.created,decision.created,tool.invoked" + + Scenario: ReactiveEventBus audit_log returns a defensive copy + Given a fresh ReactiveEventBus for taxonomy tests + When I taxonomy-emit a "plan.created" DomainEvent + Then mutating the returned audit_log should not affect the bus + + Scenario: ReactiveEventBus can cap in-memory audit_log retention + Given a ReactiveEventBus for taxonomy tests with max_audit_log_size 2 + When I taxonomy-emit a "plan.created" DomainEvent + And I taxonomy-emit a "decision.created" DomainEvent + And I taxonomy-emit a "tool.invoked" DomainEvent + Then the audit_log should contain 2 events + And the audit_log should include event types "decision.created,tool.invoked" + + Scenario: ReactiveEventBus clear_audit_log removes retained events + Given a fresh ReactiveEventBus for taxonomy tests + When I taxonomy-emit a "plan.created" DomainEvent + And I clear the taxonomy bus audit_log + Then the audit_log should contain 0 events + + Scenario: ReactiveEventBus rejects invalid max_audit_log_size + Then creating a ReactiveEventBus with max_audit_log_size 0 should raise ValueError + + # --------------------------------------------------------------------------- + # Log correlation + # --------------------------------------------------------------------------- + + Scenario: Events sharing a correlation_id are traceable + Given a fresh ReactiveEventBus for taxonomy tests + When I emit two taxonomy events with the same correlation_id + Then both events in the audit_log should share the same correlation_id + + Scenario: Independent DomainEvents get unique correlation_ids by default + When I create two independent DomainEvents for taxonomy + Then the two taxonomy events should have different correlation_ids diff --git a/features/steps/event_system_taxonomy_steps.py b/features/steps/event_system_taxonomy_steps.py new file mode 100644 index 000000000..af7676976 --- /dev/null +++ b/features/steps/event_system_taxonomy_steps.py @@ -0,0 +1,527 @@ +"""Step definitions for event_system_taxonomy.feature. + +Tests the complete Event System Domain Event Taxonomy: EventType enum +completeness, DomainEvent model fields and serialisation, EventBus Protocol, +ReactiveEventBus emit/subscribe/stream/audit_log behaviour, and log +correlation. +""" + +from __future__ import annotations + +import re +from datetime import UTC +from typing import Any + +from behave import given, then, when # type: ignore[import-untyped] +from behave.runner import Context # type: ignore[import-untyped] +from pydantic import ValidationError +from rx.core.observable.observable import Observable +from ulid import ULID + +from cleveragents.infrastructure.events import ( + DomainEvent, + EventBus, + EventType, + LoggingEventBus, + ReactiveEventBus, +) + +# --------------------------------------------------------------------------- +# Helpers +# --------------------------------------------------------------------------- + +_DOT_PATTERN = re.compile(r"^[a-z_]+\.[a-z_]+$") + + +def _make_event(event_type_str: str, **kwargs: Any) -> DomainEvent: + return DomainEvent(event_type=EventType(event_type_str), **kwargs) + + +class _TaxonomyCollector: + """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 completeness +# --------------------------------------------------------------------------- + + +@then("the EventType enum should have at least {n:d} members") +def step_event_type_min_members(ctx: Context, n: int) -> None: + count = len(EventType) + assert count >= n, f"EventType has {count} members, expected >= {n}" + + +@then('EventType should contain "{value}"') +def step_event_type_contains_value(ctx: Context, value: str) -> None: + values = {e.value for e in EventType} + assert value in values, f"{value!r} not in EventType values: {sorted(values)}" + + +@then('every EventType value should match the pattern "."') +def step_event_type_dot_pattern(ctx: Context) -> None: + for member in EventType: + assert _DOT_PATTERN.match(member.value), ( + f"EventType.{member.name} value {member.value!r} does not match " + f"the . pattern" + ) + + +# --------------------------------------------------------------------------- +# DomainEvent model +# --------------------------------------------------------------------------- + + +@when('I create a DomainEvent for taxonomy with event_type "{et}"') +def step_create_taxonomy_event(ctx: Context, et: str) -> None: + ctx.taxonomy_event = _make_event(et) + + +@then('the taxonomy event should have field "{field}"') +def step_taxonomy_event_has_field(ctx: Context, field: str) -> None: + assert field in ctx.taxonomy_event.model_fields, ( + f"DomainEvent missing field {field!r}" + ) + + +@then("the taxonomy event correlation_id should be {n:d} characters long") +def step_taxonomy_correlation_id_length(ctx: Context, n: int) -> None: + cid = ctx.taxonomy_event.correlation_id + assert len(cid) == n, f"correlation_id length {len(cid)}, expected {n}" + + +@then("the taxonomy event timestamp should be UTC") +def step_taxonomy_timestamp_utc(ctx: Context) -> None: + ts = ctx.taxonomy_event.timestamp + assert ts.tzinfo is not None, "timestamp has no timezone" + assert ts.tzinfo == UTC, f"timestamp timezone is {ts.tzinfo!r}, expected UTC" + + +@then("mutating the taxonomy event should raise a validation error") +def step_taxonomy_event_frozen(ctx: Context) -> None: + raised = False + try: + ctx.taxonomy_event.event_type = EventType.PLAN_CANCELLED # type: ignore[misc] + except ValidationError: + raised = True + assert raised, "Expected ValidationError when mutating frozen DomainEvent" + + +@when("I create a DomainEvent for taxonomy with all fields populated") +def step_create_taxonomy_full_event(ctx: Context) -> None: + ctx.taxonomy_event = DomainEvent( + event_type=EventType.PLAN_CREATED, + plan_id=str(ULID()), + root_plan_id=str(ULID()), + session_id=str(ULID()), + actor_name="local/strategist", + project_name="my-project", + details={"phase": "strategize", "actor": "strategists/planner"}, + ) + + +@then("the taxonomy event should serialize to JSON and deserialize back identically") +def step_taxonomy_event_json_roundtrip(ctx: Context) -> None: + json_str = ctx.taxonomy_event.model_dump_json() + restored = DomainEvent.model_validate_json(json_str) + assert restored == ctx.taxonomy_event, ( + f"Round-trip mismatch:\n original={ctx.taxonomy_event}\n restored={restored}" + ) + + +@when("I create a DomainEvent for taxonomy with correlation fields") +def step_create_taxonomy_event_correlation(ctx: Context) -> None: + ctx.taxonomy_event = DomainEvent( + event_type=EventType.PLAN_CREATED, + plan_id=str(ULID()), + root_plan_id=str(ULID()), + session_id=str(ULID()), + ) + + +@then("the taxonomy event plan_id should be set") +def step_taxonomy_plan_id_set(ctx: Context) -> None: + assert ctx.taxonomy_event.plan_id is not None + + +@then("the taxonomy event root_plan_id should be set") +def step_taxonomy_root_plan_id_set(ctx: Context) -> None: + assert ctx.taxonomy_event.root_plan_id is not None + + +@then("the taxonomy event session_id should be set") +def step_taxonomy_session_id_set(ctx: Context) -> None: + assert ctx.taxonomy_event.session_id is not None + + +@when("I create a DomainEvent for taxonomy with typed details") +def step_create_taxonomy_event_with_details(ctx: Context) -> None: + ctx.taxonomy_event = DomainEvent( + event_type=EventType.PLAN_PHASE_CHANGED, + details={"old_phase": "strategize", "new_phase": "execute"}, + ) + + +@then("the taxonomy event details should contain the expected keys") +def step_taxonomy_event_details_keys(ctx: Context) -> None: + details = ctx.taxonomy_event.details + assert "old_phase" in details, "Missing 'old_phase' in details" + assert "new_phase" in details, "Missing 'new_phase' in details" + + +# --------------------------------------------------------------------------- +# EventBus Protocol +# --------------------------------------------------------------------------- + + +@then('the EventBus Protocol should define method "{method}"') +def step_protocol_has_method(ctx: Context, method: str) -> None: + assert hasattr(EventBus, method), f"EventBus Protocol missing method {method!r}" + + +@then("a ReactiveEventBus instance 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("a LoggingEventBus instance 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 — emit / subscribe / stream +# --------------------------------------------------------------------------- + + +@given("a fresh ReactiveEventBus for taxonomy tests") +def step_given_fresh_bus(ctx: Context) -> None: + ctx.taxonomy_bus = ReactiveEventBus() + ctx.taxonomy_collectors = [] + + +@given("a ReactiveEventBus for taxonomy tests with max_audit_log_size {n:d}") +def step_given_capped_bus(ctx: Context, n: int) -> None: + ctx.taxonomy_bus = ReactiveEventBus(max_audit_log_size=n) + ctx.taxonomy_collectors = [] + + +@then( + "creating a ReactiveEventBus with max_audit_log_size {n:d} should raise ValueError" +) +def step_reactive_bus_invalid_audit_log_size(ctx: Context, n: int) -> None: + raised = False + try: + ReactiveEventBus(max_audit_log_size=n) + except ValueError: + raised = True + assert raised, f"Expected ValueError for max_audit_log_size={n}" + + +@when('I taxonomy-subscribe to "{et}" events') +def step_taxonomy_subscribe(ctx: Context, et: str) -> None: + collector = _TaxonomyCollector() + ctx.taxonomy_collectors.append(collector) + ctx.taxonomy_bus.subscribe(EventType(et), collector) + + +@when('I taxonomy-emit a "{et}" DomainEvent') +def step_taxonomy_emit(ctx: Context, et: str) -> None: + ctx.taxonomy_bus.emit(_make_event(et)) + + +@then("the taxonomy handler should have received {n:d} event(s)") +def step_taxonomy_handler_received(ctx: Context, n: int) -> None: + assert ctx.taxonomy_collectors, "No taxonomy collectors registered" + count = len(ctx.taxonomy_collectors[0].received) + assert count == n, f"Expected {n} event(s), got {count}" + + +@when('I taxonomy-subscribe two handlers to "{et}" events') +def step_taxonomy_subscribe_two(ctx: Context, et: str) -> None: + for _ in range(2): + collector = _TaxonomyCollector() + ctx.taxonomy_collectors.append(collector) + ctx.taxonomy_bus.subscribe(EventType(et), collector) + + +@then("each taxonomy handler should have received {n:d} event") +def step_each_taxonomy_handler_received(ctx: Context, n: int) -> None: + for i, col in enumerate(ctx.taxonomy_collectors): + assert len(col.received) == n, ( + f"Handler {i} received {len(col.received)}, expected {n}" + ) + + +@then("the taxonomy bus stream should be an Observable") +def step_taxonomy_bus_stream_observable(ctx: Context) -> None: + stream = ctx.taxonomy_bus.stream + assert isinstance(stream, Observable), ( + f"Expected Observable, got {type(stream).__name__}" + ) + + +@then('the taxonomy bus stream should not expose method "{method}"') +def step_taxonomy_bus_stream_not_expose_method(ctx: Context, method: str) -> None: + stream = ctx.taxonomy_bus.stream + assert not hasattr(stream, method), ( + f"Expected stream to hide {method!r}, but it is present" + ) + + +@when('I taxonomy-subscribe a failing handler and a normal handler to "{et}" events') +def step_taxonomy_subscribe_failing_and_normal(ctx: Context, et: str) -> None: + normal_collector = _TaxonomyCollector() + ctx.taxonomy_collectors.append(normal_collector) + + def _failing_handler(event: DomainEvent) -> None: + raise RuntimeError(f"failing taxonomy handler for {event.event_type.value}") + + ctx.taxonomy_bus.subscribe(EventType(et), _failing_handler) + ctx.taxonomy_bus.subscribe(EventType(et), normal_collector) + + +@then("the taxonomy normal handler should have received {n:d} event(s)") +def step_taxonomy_normal_handler_received(ctx: Context, n: int) -> None: + assert ctx.taxonomy_collectors, "No taxonomy normal collector registered" + count = len(ctx.taxonomy_collectors[-1].received) + assert count == n, f"Expected normal handler to receive {n} event(s), got {count}" + + +@when('I observe the taxonomy bus stream and emit a "{et}" event') +def step_taxonomy_observe_and_emit(ctx: Context, et: str) -> None: + ctx.stream_events = [] + ctx.taxonomy_bus.stream.subscribe(on_next=ctx.stream_events.append) + ctx.taxonomy_bus.emit(_make_event(et)) + + +@when('I observe taxonomy stream with a failing observer and emit a "{et}" event') +def step_taxonomy_stream_failing_observer(ctx: Context, et: str) -> None: + def _failing_observer(event: DomainEvent) -> None: + raise RuntimeError(f"failing stream observer for {event.event_type.value}") + + ctx.taxonomy_bus.stream.subscribe(on_next=_failing_observer) + ctx.taxonomy_bus.emit(_make_event(et)) + + +@then("the stream observer should have captured the event") +def step_stream_observer_captured(ctx: Context) -> None: + assert len(ctx.stream_events) == 1, ( + f"Expected 1 stream event, got {len(ctx.stream_events)}" + ) + + +@then("taxonomy-emitting a non-DomainEvent should raise TypeError") +def step_taxonomy_emit_bad_type(ctx: Context) -> None: + raised = False + try: + ctx.taxonomy_bus.emit("not-an-event") # type: ignore[arg-type] + except TypeError: + raised = True + assert raised, "Expected TypeError for non-DomainEvent" + + +@then("taxonomy-subscribing with a non-EventType should raise TypeError") +def step_taxonomy_subscribe_bad_type(ctx: Context) -> None: + raised = False + try: + ctx.taxonomy_bus.subscribe("plan.created", lambda e: None) # type: ignore[arg-type] + except TypeError: + raised = True + assert raised, "Expected TypeError for non-EventType" + + +@then("taxonomy-subscribing with a non-callable should raise TypeError") +def step_taxonomy_subscribe_non_callable(ctx: Context) -> None: + raised = False + try: + ctx.taxonomy_bus.subscribe(EventType.PLAN_CREATED, "not-callable") # type: ignore[arg-type] + except TypeError: + raised = True + assert raised, "Expected TypeError for non-callable handler" + + +# --------------------------------------------------------------------------- +# In-memory event trail (audit_log) +# --------------------------------------------------------------------------- + + +@then("the audit_log should contain {n:d} events") +def step_audit_log_count(ctx: Context, n: int) -> None: + log = ctx.taxonomy_bus.audit_log + assert len(log) == n, f"audit_log has {len(log)} events, expected {n}" + + +@then('the audit_log should include event types "{event_types}"') +def step_audit_log_contains_types(ctx: Context, event_types: str) -> None: + expected = [value.strip() for value in event_types.split(",")] + actual = [event.event_type.value for event in ctx.taxonomy_bus.audit_log] + assert actual == expected, f"audit_log types {actual} != expected {expected}" + + +@then('the audit_log events should be in order "{order}"') +def step_audit_log_order(ctx: Context, order: str) -> None: + expected = [value.strip() for value in order.split(",")] + log = ctx.taxonomy_bus.audit_log + actual = [e.event_type.value for e in log] + assert actual == expected, f"audit_log order {actual} != expected {expected}" + + +@then("mutating the returned audit_log should not affect the bus") +def step_audit_log_defensive_copy(ctx: Context) -> None: + log1 = ctx.taxonomy_bus.audit_log + original_len = len(log1) + log1.clear() # mutate the returned list + log2 = ctx.taxonomy_bus.audit_log + assert len(log2) == original_len, ( + f"audit_log was mutated: expected {original_len}, got {len(log2)}" + ) + + +@when("I clear the taxonomy bus audit_log") +def step_clear_taxonomy_audit_log(ctx: Context) -> None: + ctx.taxonomy_bus.clear_audit_log() + + +# --------------------------------------------------------------------------- +# Log correlation +# --------------------------------------------------------------------------- + + +@when("I emit two taxonomy events with the same correlation_id") +def step_emit_correlated_events(ctx: Context) -> None: + shared_cid = str(ULID()) + ctx.taxonomy_bus.emit( + DomainEvent( + event_type=EventType.PLAN_CREATED, + correlation_id=shared_cid, + ) + ) + ctx.taxonomy_bus.emit( + DomainEvent( + event_type=EventType.PLAN_PHASE_CHANGED, + correlation_id=shared_cid, + ) + ) + ctx.shared_correlation_id = shared_cid + + +@then("both events in the audit_log should share the same correlation_id") +def step_correlated_events_in_log(ctx: Context) -> None: + log = ctx.taxonomy_bus.audit_log + assert len(log) == 2, f"Expected 2 events, got {len(log)}" + assert log[0].correlation_id == ctx.shared_correlation_id + assert log[1].correlation_id == ctx.shared_correlation_id + + +@when("I create two independent DomainEvents for taxonomy") +def step_create_two_independent_events(ctx: Context) -> None: + ctx.taxonomy_event_one = DomainEvent(event_type=EventType.PLAN_CREATED) + ctx.taxonomy_event_two = DomainEvent(event_type=EventType.DECISION_CREATED) + + +@then("the two taxonomy events should have different correlation_ids") +def step_taxonomy_events_have_different_correlation_ids(ctx: Context) -> None: + cid_one = ctx.taxonomy_event_one.correlation_id + cid_two = ctx.taxonomy_event_two.correlation_id + assert cid_one != cid_two, ( + "Expected independently created DomainEvents to have unique " + f"correlation_ids, got {cid_one!r} and {cid_two!r}" + ) + + +# --------------------------------------------------------------------------- +# LoggingEventBus — emit / subscribe +# --------------------------------------------------------------------------- + + +@given("a fresh LoggingEventBus for taxonomy tests") +def step_given_fresh_logging_bus(ctx: Context) -> None: + ctx.logging_taxonomy_bus = LoggingEventBus() + ctx.logging_taxonomy_collectors = [] + + +@when('I logging-taxonomy-subscribe to "{et}" events') +def step_logging_taxonomy_subscribe(ctx: Context, et: str) -> None: + collector = _TaxonomyCollector() + ctx.logging_taxonomy_collectors.append(collector) + ctx.logging_taxonomy_bus.subscribe(EventType(et), collector) + + +@when('I logging-taxonomy-emit a "{et}" DomainEvent') +def step_logging_taxonomy_emit(ctx: Context, et: str) -> None: + ctx.logging_taxonomy_bus.emit(_make_event(et)) + + +@then("the logging taxonomy handler should have received {n:d} event(s)") +def step_logging_taxonomy_handler_received(ctx: Context, n: int) -> None: + assert ctx.logging_taxonomy_collectors, "No logging taxonomy collectors registered" + count = len(ctx.logging_taxonomy_collectors[0].received) + assert count == n, f"Expected {n} event(s), got {count}" + + +@when( + 'I logging-taxonomy-subscribe a failing handler and a normal handler to "{et}" events' +) +def step_logging_taxonomy_subscribe_failing_and_normal(ctx: Context, et: str) -> None: + normal_collector = _TaxonomyCollector() + ctx.logging_taxonomy_collectors.append(normal_collector) + + def _failing_handler(event: DomainEvent) -> None: + raise RuntimeError(f"failing logging handler for {event.event_type.value}") + + ctx.logging_taxonomy_bus.subscribe(EventType(et), _failing_handler) + ctx.logging_taxonomy_bus.subscribe(EventType(et), normal_collector) + + +@then("the logging taxonomy normal handler should have received {n:d} event(s)") +def step_logging_taxonomy_normal_handler_received(ctx: Context, n: int) -> None: + assert ctx.logging_taxonomy_collectors, ( + "No logging taxonomy normal collector registered" + ) + count = len(ctx.logging_taxonomy_collectors[-1].received) + assert count == n, f"Expected normal handler to receive {n} event(s), got {count}" + + +@then("logging-taxonomy-emitting a non-DomainEvent should raise TypeError") +def step_logging_taxonomy_emit_bad_type(ctx: Context) -> None: + raised = False + try: + ctx.logging_taxonomy_bus.emit("not-an-event") # type: ignore[arg-type] + except TypeError: + raised = True + assert raised, "Expected TypeError for non-DomainEvent" + + +@then("logging-taxonomy-subscribing with a non-EventType should raise TypeError") +def step_logging_taxonomy_subscribe_bad_type(ctx: Context) -> None: + raised = False + try: + ctx.logging_taxonomy_bus.subscribe( + "plan.created", + lambda e: None, # type: ignore[arg-type] + ) + except TypeError: + raised = True + assert raised, "Expected TypeError for non-EventType" + + +@then("logging-taxonomy-subscribing with a non-callable should raise TypeError") +def step_logging_taxonomy_subscribe_non_callable(ctx: Context) -> None: + raised = False + try: + ctx.logging_taxonomy_bus.subscribe( + EventType.PLAN_CREATED, + "not-callable", # type: ignore[arg-type] + ) + except TypeError: + raised = True + assert raised, "Expected TypeError for non-callable handler" diff --git a/features/tdd_event_bus_exception_swallow.feature b/features/tdd_event_bus_exception_swallow.feature index 0768d91b6..0635d5b8e 100644 --- a/features/tdd_event_bus_exception_swallow.feature +++ b/features/tdd_event_bus_exception_swallow.feature @@ -1,4 +1,4 @@ -@tdd_expected_fail @tdd_bug @tdd_bug_988 +@tdd_bug @tdd_bug_988 Feature: TDD Bug #988 — ReactiveEventBus.emit() swallows exception details As a developer debugging a subscriber failure I want the exception handler in ReactiveEventBus.emit() to log the full @@ -11,10 +11,11 @@ Feature: TDD Bug #988 — ReactiveEventBus.emit() swallows exception details subscriber raises, the log contains no diagnostic detail — no message, no traceback, no clue about which input or state caused the failure. - The test uses `@tdd_expected_fail` because the underlying assertion - FAILS (confirming the bug exists). The tag inversion mechanism causes - CI to report the test as passed. Once the fix for #988 is merged, - the `@tdd_expected_fail` tag must be removed so the test runs normally. + The traceback scenario uses `@tdd_expected_fail` because that + assertion still FAILS while the bug exists. The tag inversion + mechanism causes CI to report that scenario as passed. Once the + fix for #988 is merged, the `@tdd_expected_fail` tag must be + removed so both scenarios run normally. See CONTRIBUTING.md > Bug Fix Workflow > TDD Bug Test Tags. @@ -23,6 +24,7 @@ Feature: TDD Bug #988 — ReactiveEventBus.emit() swallows exception details When I emit an event that triggers the failing handler Then the warning log should contain the exception message text + @tdd_expected_fail Scenario: Bug #988 — emit() logs traceback via exc_info when handler raises Given a ReactiveEventBus with a handler that raises a ValueError When I emit an event that triggers the failing handler diff --git a/robot/event_system_taxonomy_integration.robot b/robot/event_system_taxonomy_integration.robot new file mode 100644 index 000000000..96f0c0632 --- /dev/null +++ b/robot/event_system_taxonomy_integration.robot @@ -0,0 +1,192 @@ +*** Settings *** +Documentation Integration tests for Event System Domain Event Taxonomy +... Validates EventType completeness, DomainEvent model fields, +... ReactiveEventBus emit/subscribe/stream/audit_log, and +... LoggingEventBus conformance. +Resource ${CURDIR}/common.resource +Suite Setup Setup Test Environment +Suite Teardown Cleanup Test Environment + +*** Variables *** +${HELPER} ${CURDIR}/helper_event_system_taxonomy.py + +*** Test Cases *** +EventType Enum Has At Least 43 Members + [Documentation] Verify the EventType enum defines at least 43 event types + ${result}= Run Process ${PYTHON} ${HELPER} event_type_count + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} event-type-count-ok + +EventType Covers All Required Domains + [Documentation] Verify EventType has values for all required domains + ${result}= Run Process ${PYTHON} ${HELPER} event_type_domains + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} event-type-domains-ok + +EventType Values Follow Dot Convention + [Documentation] Every value matches . + ${result}= Run Process ${PYTHON} ${HELPER} event_type_dot_convention + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} event-type-dot-convention-ok + +DomainEvent Has All Required Fields + [Documentation] Verify DomainEvent includes all spec-required fields + ${result}= Run Process ${PYTHON} ${HELPER} domain_event_fields + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} domain-event-fields-ok + +DomainEvent JSON Round Trip + [Documentation] Serialize and deserialize a fully populated DomainEvent + ${result}= Run Process ${PYTHON} ${HELPER} domain_event_json_roundtrip + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} domain-event-json-roundtrip-ok + +DomainEvent Is Immutable + [Documentation] Frozen model rejects field mutation + ${result}= Run Process ${PYTHON} ${HELPER} domain_event_immutable + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} domain-event-immutable-ok + +ReactiveEventBus Emit Subscribe Audit + [Documentation] Emit dispatches to handlers and persists to audit_log + ${result}= Run Process ${PYTHON} ${HELPER} reactive_emit_subscribe_audit + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} reactive-emit-subscribe-audit-ok + +ReactiveEventBus Continues Dispatching After Handler Error + [Documentation] A failing handler does not block subsequent handlers + ${result}= Run Process ${PYTHON} ${HELPER} reactive_handler_exception_isolation + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} reactive-handler-exception-isolation-ok + +ReactiveEventBus Continues After Stream Observer Error + [Documentation] A failing stream observer does not block handlers or audit + ${result}= Run Process ${PYTHON} ${HELPER} reactive_stream_exception_isolation + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} reactive-stream-exception-isolation-ok + +ReactiveEventBus Stream Observable + [Documentation] stream property exposes rx.Observable that receives events + ${result}= Run Process ${PYTHON} ${HELPER} reactive_stream_observable + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} reactive-stream-observable-ok + +ReactiveEventBus Stream Is Read Only + [Documentation] stream should not expose Subject mutation methods + ${result}= Run Process ${PYTHON} ${HELPER} reactive_stream_read_only + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} reactive-stream-read-only-ok + +ReactiveEventBus Audit Log Order + [Documentation] audit_log preserves emission order + ${result}= Run Process ${PYTHON} ${HELPER} audit_log_order + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} audit-log-order-ok + +ReactiveEventBus Audit Log Defensive Copy + [Documentation] audit_log returns a copy — external mutation is safe + ${result}= Run Process ${PYTHON} ${HELPER} audit_log_defensive_copy + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} audit-log-defensive-copy-ok + +ReactiveEventBus Audit Log Retention Cap + [Documentation] configured cap keeps only newest events and supports clearing + ${result}= Run Process ${PYTHON} ${HELPER} audit_log_retention_cap + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} audit-log-retention-cap-ok + +ReactiveEventBus Rejects Invalid Audit Log Retention Cap + [Documentation] max_audit_log_size must be positive when provided + ${result}= Run Process ${PYTHON} ${HELPER} audit_log_invalid_retention_cap + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} audit-log-invalid-retention-cap-ok + +Log Correlation Via Shared Correlation ID + [Documentation] Events sharing a correlation_id are traceable in audit_log + ${result}= Run Process ${PYTHON} ${HELPER} log_correlation + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} log-correlation-ok + +Protocol Conformance For Both Bus Implementations + [Documentation] Both ReactiveEventBus and LoggingEventBus satisfy EventBus + ${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 + +LoggingEventBus Emit Subscribe + [Documentation] LoggingEventBus dispatches to subscribed 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 + +LoggingEventBus Continues Dispatching After Handler Error + [Documentation] LoggingEventBus isolates handler exceptions and continues + ${result}= Run Process ${PYTHON} ${HELPER} logging_handler_exception_isolation + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} logging-handler-exception-isolation-ok + +ReactiveEventBus Type Validation + [Documentation] Emit/subscribe reject invalid argument types + ${result}= Run Process ${PYTHON} ${HELPER} type_validation + ... cwd=${WORKSPACE} + Log ${result.stdout} + Log ${result.stderr} + Should Be Equal As Integers ${result.rc} 0 + Should Contain ${result.stdout} type-validation-ok diff --git a/robot/helper_event_system_taxonomy.py b/robot/helper_event_system_taxonomy.py new file mode 100644 index 000000000..c908f8243 --- /dev/null +++ b/robot/helper_event_system_taxonomy.py @@ -0,0 +1,427 @@ +"""Helper script for event_system_taxonomy_integration.robot. + +Each subcommand is a self-contained check that prints a sentinel on success. +""" + +from __future__ import annotations + +import re +import sys +from pathlib import Path + +from pydantic import ValidationError +from rx.core.observable.observable import Observable +from ulid import ULID + +# 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, +) + +_DOT_PATTERN = re.compile(r"^[a-z_]+\.[a-z_]+$") + +# --------------------------------------------------------------------------- +# Subcommands +# --------------------------------------------------------------------------- + + +def event_type_count() -> None: + """Verify EventType has at least 43 members.""" + count = len(EventType) + if count < 43: + print(f"FAIL: EventType has {count} members, expected >= 43", file=sys.stderr) + sys.exit(1) + print("event-type-count-ok") + + +def event_type_domains() -> None: + """Verify EventType covers all required domains.""" + required_values = [ + "plan.created", + "decision.created", + "invariant.reconciled", + "actor.invoked", + "tool.invoked", + "resource.accessed", + "sandbox.created", + "checkpoint.created", + "context.built", + "validation.started", + "session.created", + "budget.warning", + "correction.applied", + "config.changed", + "entity.deleted", + "auth.success", + "auth.failure", + ] + actual = {e.value for e in EventType} + missing = [v for v in required_values if v not in actual] + if missing: + print(f"FAIL: missing domain values: {missing}", file=sys.stderr) + sys.exit(1) + print("event-type-domains-ok") + + +def event_type_dot_convention() -> None: + """Verify every EventType value matches ..""" + for member in EventType: + if not _DOT_PATTERN.match(member.value): + print( + f"FAIL: {member.name}={member.value!r} violates convention", + file=sys.stderr, + ) + sys.exit(1) + print("event-type-dot-convention-ok") + + +def domain_event_fields() -> None: + """Verify DomainEvent has all required fields.""" + required = { + "event_type", + "timestamp", + "correlation_id", + "plan_id", + "root_plan_id", + "session_id", + "actor_name", + "project_name", + "details", + } + actual = set(DomainEvent.model_fields) + missing = required - actual + if missing: + print(f"FAIL: DomainEvent missing fields: {missing}", file=sys.stderr) + sys.exit(1) + print("domain-event-fields-ok") + + +def domain_event_json_roundtrip() -> None: + """Serialize and deserialize a fully populated DomainEvent.""" + event = DomainEvent( + event_type=EventType.PLAN_CREATED, + plan_id=str(ULID()), + root_plan_id=str(ULID()), + session_id=str(ULID()), + actor_name="local/strategist", + project_name="test-project", + details={"phase": "strategize"}, + ) + json_str = event.model_dump_json() + restored = DomainEvent.model_validate_json(json_str) + if restored != event: + print("FAIL: JSON round-trip mismatch", file=sys.stderr) + sys.exit(1) + print("domain-event-json-roundtrip-ok") + + +def domain_event_immutable() -> None: + """Verify DomainEvent is frozen.""" + event = DomainEvent(event_type=EventType.PLAN_CREATED) + try: + event.event_type = EventType.PLAN_CANCELLED # type: ignore[misc] + except ValidationError: + print("domain-event-immutable-ok") + return + print("FAIL: mutation was not rejected", file=sys.stderr) + sys.exit(1) + + +def reactive_emit_subscribe_audit() -> None: + """Emit dispatches to handlers and records audit_log event details.""" + 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: handler got {len(received)} events", file=sys.stderr) + sys.exit(1) + if len(bus.audit_log) != 1: + print(f"FAIL: audit_log has {len(bus.audit_log)} events", file=sys.stderr) + sys.exit(1) + if received[0].event_type != EventType.PLAN_CREATED: + print("FAIL: handler received wrong event type", file=sys.stderr) + sys.exit(1) + if bus.audit_log[0].event_type != EventType.PLAN_CREATED: + print("FAIL: audit_log captured wrong event type", file=sys.stderr) + sys.exit(1) + print("reactive-emit-subscribe-audit-ok") + + +def reactive_handler_exception_isolation() -> None: + """Verify failing handlers do not block ReactiveEventBus dispatch.""" + received: list[DomainEvent] = [] + bus = ReactiveEventBus() + + def failing_handler(event: DomainEvent) -> None: + raise RuntimeError(f"intentional failure for {event.event_type.value}") + + bus.subscribe(EventType.PLAN_CREATED, failing_handler) + bus.subscribe(EventType.PLAN_CREATED, received.append) + bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED)) + + if len(received) != 1: + print("FAIL: normal handler was not called", file=sys.stderr) + sys.exit(1) + if len(bus.audit_log) != 1: + print("FAIL: audit_log missing emitted event", file=sys.stderr) + sys.exit(1) + print("reactive-handler-exception-isolation-ok") + + +def reactive_stream_exception_isolation() -> None: + """Verify failing stream observers do not block bus dispatch.""" + received: list[DomainEvent] = [] + bus = ReactiveEventBus() + bus.subscribe(EventType.PLAN_CREATED, received.append) + + def failing_stream_observer(event: DomainEvent) -> None: + raise RuntimeError(f"intentional stream failure for {event.event_type.value}") + + bus.stream.subscribe(on_next=failing_stream_observer) + bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED)) + + if len(received) != 1: + print("FAIL: stream failure blocked typed handler dispatch", file=sys.stderr) + sys.exit(1) + if len(bus.audit_log) != 1: + print("FAIL: stream failure blocked audit recording", file=sys.stderr) + sys.exit(1) + print("reactive-stream-exception-isolation-ok") + + +def reactive_stream_observable() -> None: + """stream property exposes rx.Observable that receives events.""" + bus = ReactiveEventBus() + if not isinstance(bus.stream, Observable): + print("FAIL: stream is not Observable", file=sys.stderr) + sys.exit(1) + captured: list[DomainEvent] = [] + bus.stream.subscribe(on_next=captured.append) + bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED)) + if len(captured) != 1: + print(f"FAIL: stream captured {len(captured)} events", file=sys.stderr) + sys.exit(1) + print("reactive-stream-observable-ok") + + +def reactive_stream_read_only() -> None: + """stream property returns read-only observable view.""" + bus = ReactiveEventBus() + stream = bus.stream + + if hasattr(stream, "on_next"): + print("FAIL: stream unexpectedly exposes on_next", file=sys.stderr) + sys.exit(1) + + captured: list[DomainEvent] = [] + stream.subscribe(on_next=captured.append) + bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED)) + if len(captured) != 1: + print("FAIL: read-only stream did not receive emitted event", file=sys.stderr) + sys.exit(1) + print("reactive-stream-read-only-ok") + + +def audit_log_order() -> None: + """audit_log preserves emission order.""" + bus = ReactiveEventBus() + types = [EventType.PLAN_CREATED, EventType.DECISION_CREATED, EventType.TOOL_INVOKED] + for et in types: + bus.emit(DomainEvent(event_type=et)) + log_types = [e.event_type.value for e in bus.audit_log] + expected = ["plan.created", "decision.created", "tool.invoked"] + if log_types != expected: + print(f"FAIL: order {log_types} != {expected}", file=sys.stderr) + sys.exit(1) + print("audit-log-order-ok") + + +def audit_log_defensive_copy() -> None: + """audit_log returns a copy — external mutation is safe.""" + bus = ReactiveEventBus() + bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED)) + log1 = bus.audit_log + log1.clear() + if len(bus.audit_log) != 1: + print("FAIL: audit_log was externally mutated", file=sys.stderr) + sys.exit(1) + print("audit-log-defensive-copy-ok") + + +def audit_log_retention_cap() -> None: + """Configured max_audit_log_size evicts oldest events FIFO.""" + bus = ReactiveEventBus(max_audit_log_size=2) + bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED)) + bus.emit(DomainEvent(event_type=EventType.DECISION_CREATED)) + bus.emit(DomainEvent(event_type=EventType.TOOL_INVOKED)) + + log = bus.audit_log + if len(log) != 2: + print( + f"FAIL: expected capped audit_log size 2, got {len(log)}", file=sys.stderr + ) + sys.exit(1) + + actual = [event.event_type.value for event in log] + expected = ["decision.created", "tool.invoked"] + if actual != expected: + print(f"FAIL: capped audit_log order {actual} != {expected}", file=sys.stderr) + sys.exit(1) + + bus.clear_audit_log() + if bus.audit_log: + print("FAIL: clear_audit_log did not empty audit_log", file=sys.stderr) + sys.exit(1) + + print("audit-log-retention-cap-ok") + + +def audit_log_invalid_retention_cap() -> None: + """Invalid retention cap should be rejected.""" + try: + ReactiveEventBus(max_audit_log_size=0) + except ValueError: + print("audit-log-invalid-retention-cap-ok") + return + print("FAIL: expected ValueError for max_audit_log_size=0", file=sys.stderr) + sys.exit(1) + + +def log_correlation() -> None: + """Events sharing a correlation_id are traceable in audit_log.""" + bus = ReactiveEventBus() + shared_cid = str(ULID()) + bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED, correlation_id=shared_cid)) + bus.emit( + DomainEvent(event_type=EventType.PLAN_PHASE_CHANGED, correlation_id=shared_cid) + ) + log = bus.audit_log + if len(log) != 2: + print(f"FAIL: expected 2 events, got {len(log)}", file=sys.stderr) + sys.exit(1) + if log[0].correlation_id != shared_cid or log[1].correlation_id != shared_cid: + print("FAIL: correlation_id mismatch", file=sys.stderr) + sys.exit(1) + print("log-correlation-ok") + + +def protocol_conformance() -> None: + """Both bus implementations satisfy EventBus Protocol.""" + for cls in (ReactiveEventBus, LoggingEventBus): + bus = cls() + if not isinstance(bus, EventBus): + print(f"FAIL: {cls.__name__} != EventBus", file=sys.stderr) + sys.exit(1) + print("protocol-conformance-ok") + + +def logging_emit_subscribe() -> None: + """Verify LoggingEventBus emits to subscribed handlers.""" + received: list[DomainEvent] = [] + bus = LoggingEventBus() + bus.subscribe(EventType.PLAN_CREATED, received.append) + bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED)) + if len(received) != 1: + print("FAIL: logging handler did not receive event", file=sys.stderr) + sys.exit(1) + if received[0].event_type != EventType.PLAN_CREATED: + print("FAIL: logging handler received wrong event type", file=sys.stderr) + sys.exit(1) + print("logging-emit-subscribe-ok") + + +def logging_handler_exception_isolation() -> None: + """Verify LoggingEventBus isolates handler failures.""" + received: list[DomainEvent] = [] + bus = LoggingEventBus() + + def failing_handler(event: DomainEvent) -> None: + raise RuntimeError(f"intentional failure for {event.event_type.value}") + + bus.subscribe(EventType.PLAN_CREATED, failing_handler) + bus.subscribe(EventType.PLAN_CREATED, received.append) + bus.emit(DomainEvent(event_type=EventType.PLAN_CREATED)) + + if len(received) != 1: + print("FAIL: normal logging handler was not called", file=sys.stderr) + sys.exit(1) + print("logging-handler-exception-isolation-ok") + + +def type_validation() -> None: + """Emit/subscribe reject invalid argument types.""" + bus = ReactiveEventBus() + + # emit non-DomainEvent + try: + bus.emit("not-an-event") # type: ignore[arg-type] + except TypeError: + pass + else: + print("FAIL: emit did not reject non-DomainEvent", file=sys.stderr) + sys.exit(1) + + # subscribe non-EventType + try: + bus.subscribe("plan.created", lambda e: None) # type: ignore[arg-type] + except TypeError: + pass + else: + print("FAIL: subscribe did not reject non-EventType", file=sys.stderr) + sys.exit(1) + + # subscribe non-callable + try: + bus.subscribe(EventType.PLAN_CREATED, "not-callable") # type: ignore[arg-type] + except TypeError: + pass + else: + print("FAIL: subscribe did not reject non-callable", file=sys.stderr) + sys.exit(1) + + print("type-validation-ok") + + +# --------------------------------------------------------------------------- +# Dispatch +# --------------------------------------------------------------------------- + +_COMMANDS = { + "event_type_count": event_type_count, + "event_type_domains": event_type_domains, + "event_type_dot_convention": event_type_dot_convention, + "domain_event_fields": domain_event_fields, + "domain_event_json_roundtrip": domain_event_json_roundtrip, + "domain_event_immutable": domain_event_immutable, + "reactive_emit_subscribe_audit": reactive_emit_subscribe_audit, + "reactive_handler_exception_isolation": reactive_handler_exception_isolation, + "reactive_stream_exception_isolation": reactive_stream_exception_isolation, + "reactive_stream_observable": reactive_stream_observable, + "reactive_stream_read_only": reactive_stream_read_only, + "audit_log_order": audit_log_order, + "audit_log_defensive_copy": audit_log_defensive_copy, + "audit_log_retention_cap": audit_log_retention_cap, + "audit_log_invalid_retention_cap": audit_log_invalid_retention_cap, + "log_correlation": log_correlation, + "protocol_conformance": protocol_conformance, + "logging_emit_subscribe": logging_emit_subscribe, + "logging_handler_exception_isolation": logging_handler_exception_isolation, + "type_validation": type_validation, +} + +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/infrastructure/events/logging_bus.py b/src/cleveragents/infrastructure/events/logging_bus.py index 4a6694f57..307bf4ed3 100644 --- a/src/cleveragents/infrastructure/events/logging_bus.py +++ b/src/cleveragents/infrastructure/events/logging_bus.py @@ -70,8 +70,17 @@ class LoggingEventBus: actor_name=event.actor_name, details=event.details, ) - for handler in self._subscriptions.get(event.event_type, []): - handler(event) + for handler in list(self._subscriptions.get(event.event_type, ())): + try: + handler(event) + except Exception as exc: + _logger.warning( + "logging_event_handler_failed", + event_type=event.event_type.value, + handler=getattr(handler, "__qualname__", repr(handler)), + error_type=type(exc).__name__, + error=str(exc), + ) def subscribe( self, diff --git a/src/cleveragents/infrastructure/events/reactive.py b/src/cleveragents/infrastructure/events/reactive.py index 7f1262fe0..4161bf3b9 100644 --- a/src/cleveragents/infrastructure/events/reactive.py +++ b/src/cleveragents/infrastructure/events/reactive.py @@ -18,9 +18,11 @@ Based on: from __future__ import annotations +from collections import deque from collections.abc import Callable import structlog +from rx import operators as ops from rx.core.observable.observable import Observable from rx.subject.subject import Subject @@ -30,6 +32,10 @@ from cleveragents.infrastructure.events.types import EventType _logger = structlog.get_logger(__name__) +def _identity(event: DomainEvent) -> DomainEvent: + return event + + class ReactiveEventBus: """In-process event bus using RxPY for real-time distribution. @@ -40,27 +46,60 @@ class ReactiveEventBus: Attributes: _subject: Hot RxPY Subject — the raw observable stream. + _stream: Read-only Observable view over ``_subject``. _subscriptions: Per-event-type handler lists. + _audit_log: Volatile in-memory log of all emitted events. """ - def __init__(self) -> None: + def __init__(self, max_audit_log_size: int | None = None) -> None: + if max_audit_log_size is not None and max_audit_log_size < 1: + raise ValueError("max_audit_log_size must be >= 1 when provided, or None") + self._subject: Subject = Subject() + self._stream: Observable = self._subject.pipe(ops.map(_identity)) self._subscriptions: dict[EventType, list[Callable[[DomainEvent], None]]] = {} + self._audit_log: deque[DomainEvent] = deque(maxlen=max_audit_log_size) # ------------------------------------------------------------------ # Public API (satisfies EventBus protocol) # ------------------------------------------------------------------ + @property + def audit_log(self) -> list[DomainEvent]: + """Return a snapshot of the in-memory event log. + + Every call to :meth:`emit` appends the event to this log, + guaranteeing a complete in-process audit trail. + + .. note:: + + This is a **volatile, in-memory** log intended for runtime + inspection and testing. It does **not** replace the durable + ``audit_log`` SQLite table described in the specification + (§Persistence Schema). Durable persistence is wired + separately via the audit service layer. + + If ``max_audit_log_size`` was configured at construction, + oldest events are evicted FIFO when the cap is reached. + + Returns: + A *copy* of the internal audit list to prevent external + mutation. + """ + return list(self._audit_log) + + def clear_audit_log(self) -> None: + """Clear all retained in-memory audit events.""" + self._audit_log.clear() + def emit(self, event: DomainEvent) -> None: """Publish *event* to the reactive stream and all type handlers. - Note on audit logging: The specification shows ``_persist_audit(event)`` - called here. In this implementation, audit logging is achieved by either: - (1) subscribing a logging handler to the desired event types, or - (2) using - :class:`~cleveragents.infrastructure.events.logging_bus.LoggingEventBus` - which logs all events automatically. This design keeps the reactive bus - focused on event distribution, following the single-responsibility principle. + Execution order follows the specification (§Event System): + + 1. Push into the RxPY subject for real-time subscribers. + 2. Append to the in-memory :attr:`audit_log`. + 3. Dispatch to per-type handlers registered via :meth:`subscribe`. Args: event: The :class:`DomainEvent` to publish. @@ -72,18 +111,29 @@ class ReactiveEventBus: 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, []): + + try: + self._subject.on_next(event) + except Exception as exc: + _logger.warning( + "reactive_stream_error", + event_type=event.event_type.value, + error_type=type(exc).__name__, + error=str(exc), + ) + + self._audit_log.append(event.model_copy(deep=True)) + + for handler in list(self._subscriptions.get(event.event_type, ())): try: handler(event) except Exception as exc: _logger.warning( "event_handler_failed", - event_type=event.event_type.value - if hasattr(event.event_type, "value") - else str(event.event_type), + event_type=event.event_type.value, handler=getattr(handler, "__qualname__", repr(handler)), error_type=type(exc).__name__, + error=str(exc), ) def subscribe( @@ -114,13 +164,13 @@ class ReactiveEventBus: @property def stream(self) -> Observable: - """Raw observable stream for advanced RxPY operators. + """Read-only observable stream for advanced RxPY operators. Returns: - The underlying :class:`rx.subject.Subject` cast as an - :class:`~rx.core.observable.observable.Observable`. + A read-only :class:`~rx.core.observable.observable.Observable` + view over the internal Subject. """ - return self._subject + return self._stream __all__ = ["ReactiveEventBus"] diff --git a/vulture_whitelist.py b/vulture_whitelist.py index e47a0c35d..c2304c6b9 100644 --- a/vulture_whitelist.py +++ b/vulture_whitelist.py @@ -1149,6 +1149,9 @@ safe_uri_segment # noqa: B018, F821 # Container DI factory — wired via dependency_injector (issue #578) _build_analyzer_registry # noqa: B018, F821 +# ReactiveEventBus.audit_log — public property for event audit trail (#587) +audit_log # noqa: B018, F821 + # Test doubles — public API for BDD/Robot test infrastructure TrackingEventBus # noqa: B018, F821