cleveragents-core/robot/helper_event_system_taxonomy.py

"""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 <domain>.<action>."""
    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]} <command>", file=sys.stderr)
        print(f"Commands: {list(_COMMANDS)}", file=sys.stderr)
        sys.exit(1)
    _COMMANDS[sys.argv[1]]()