cleveragents/cleveragents-core

Fork 3

[events/ReactiveEventBus] ReactiveEventBus._subject (RxPY Subject) is never completed — missing close() method causes RxPY subscriber resource leaks #10378

New issue

Closed

opened 2026-04-18 09:17:41 +00:00 by HAL9000 · 3 comments

HAL9000 commented

2026-04-18 09:17:41 +00:00

Owner

Metadata

Commit: fix(events): add close() method to ReactiveEventBus to complete RxPY subject
Branch: fix/reactive-event-bus-close-method

Background and Context

ReactiveEventBus in src/cleveragents/infrastructure/events/reactive.py creates an RxPY Subject but never calls on_completed() on it. There is no close() or shutdown() method. When the bus is garbage collected or the application shuts down, the RxPY Subject is abandoned without signaling completion. This causes resource leaks in RxPY subscribers that maintain internal state until on_completed is received.

Expected Behavior

ReactiveEventBus should provide a close() method that:

Calls self._subject.on_completed() to signal stream completion to all RxPY subscribers
Prevents further emit() calls (raises RuntimeError or logs a warning)
Optionally implements __enter__/__exit__ for context manager support

Acceptance Criteria

ReactiveEventBus.close() calls self._subject.on_completed()
After close(), emit() raises RuntimeError or logs a warning and returns
ReactiveEventBus implements __enter__/__exit__ for context manager support
TDD test (see blocked-by issue #10377) passes after fix
nox passes with coverage ≥ 97%

Subtasks

Add close() method to ReactiveEventBus
Add __enter__/__exit__ for context manager support
Guard emit() against calls after close()
Add tests verifying on_completed is received by RxPY subscribers after close()
Run nox to confirm coverage ≥ 97%

Definition of Done

This issue is closed when:

ReactiveEventBus.close() is implemented and calls on_completed()
The TDD test (blocked-by issue #10377) passes
nox passes with coverage ≥ 97%
A PR is reviewed and merged to main

Summary

Current Behaviour

# src/cleveragents/infrastructure/events/reactive.py
class ReactiveEventBus:
    def __init__(self, max_audit_log_size: int | None = None) -> None:
        self._subject: Subject = Subject()
        self._stream: Observable = self._subject.pipe(ops.map(_identity))
        # ...
    
    # NO close() method
    # NO __del__() method  
    # self._subject.on_completed() is NEVER called

The stream property exposes the Observable for advanced RxPY operators. The docstring says: "subscribers that want filtering, buffering, or debouncing can operate directly on stream." However, operators like buffer_with_count, window, take_until, to_list(), and last() all wait for on_completed to finalize their output. Since on_completed is never called, these operators will:

Buffer events indefinitely (memory leak)
Never emit their final result
Hold references to internal state forever

Impact

RxPY buffering operators leak memory: buffer_with_count(N), window(N), and similar operators accumulate events in internal buffers waiting for on_completed. These buffers grow forever.
Aggregation operators never complete: to_list(), last(), reduce(), scan() with accumulation — all wait for on_completed before emitting. Code that uses these operators will hang forever.
No graceful shutdown: There is no way to signal to RxPY subscribers that the bus is shutting down. In application shutdown scenarios, subscribers cannot clean up their state.
Missing context manager support: The bus cannot be used as a context manager (with ReactiveEventBus() as bus:), making it harder to ensure cleanup in tests and application code.

Fix

def close(self) -> None:
    """Complete the RxPY stream and prevent further event emission.
    
    Signals on_completed() to all RxPY subscribers, allowing them to
    finalize their state. After close(), emit() calls will raise
    RuntimeError.
    """
    self._subject.on_completed()

def __enter__(self) -> "ReactiveEventBus":
    return self

def __exit__(self, *args: object) -> None:
    self.close()

References

src/cleveragents/infrastructure/events/reactive.py — ReactiveEventBus (no close() method)
RxPY documentation on Subject completion

Automated by CleverAgents Bot
Supervisor: Bug Hunt Pool | Agent: bug-hunt-pool-supervisor

## Metadata - **Commit**: `fix(events): add close() method to ReactiveEventBus to complete RxPY subject` - **Branch**: `fix/reactive-event-bus-close-method` ## Background and Context `ReactiveEventBus` in `src/cleveragents/infrastructure/events/reactive.py` creates an RxPY `Subject` but never calls `on_completed()` on it. There is no `close()` or `shutdown()` method. When the bus is garbage collected or the application shuts down, the RxPY `Subject` is abandoned without signaling completion. This causes resource leaks in RxPY subscribers that maintain internal state until `on_completed` is received. ## Expected Behavior `ReactiveEventBus` should provide a `close()` method that: 1. Calls `self._subject.on_completed()` to signal stream completion to all RxPY subscribers 2. Prevents further `emit()` calls (raises `RuntimeError` or logs a warning) 3. Optionally implements `__enter__`/`__exit__` for context manager support ## Acceptance Criteria - [ ] `ReactiveEventBus.close()` calls `self._subject.on_completed()` - [ ] After `close()`, `emit()` raises `RuntimeError` or logs a warning and returns - [ ] `ReactiveEventBus` implements `__enter__`/`__exit__` for context manager support - [ ] TDD test (see blocked-by issue #10377) passes after fix - [ ] `nox` passes with coverage ≥ 97% ## Subtasks - [ ] Add `close()` method to `ReactiveEventBus` - [ ] Add `__enter__`/`__exit__` for context manager support - [ ] Guard `emit()` against calls after `close()` - [ ] Add tests verifying `on_completed` is received by RxPY subscribers after `close()` - [ ] Run `nox` to confirm coverage ≥ 97% ## Definition of Done This issue is closed when: 1. `ReactiveEventBus.close()` is implemented and calls `on_completed()` 2. The TDD test (blocked-by issue #10377) passes 3. `nox` passes with coverage ≥ 97% 4. A PR is reviewed and merged to `main` --- ## Summary `ReactiveEventBus` in `src/cleveragents/infrastructure/events/reactive.py` creates an RxPY `Subject` but never calls `on_completed()` on it. There is no `close()` or `shutdown()` method. When the bus is garbage collected or the application shuts down, the RxPY `Subject` is abandoned without signaling completion. This causes resource leaks in RxPY subscribers that maintain internal state until `on_completed` is received. ## Current Behaviour ```python # src/cleveragents/infrastructure/events/reactive.py class ReactiveEventBus: def __init__(self, max_audit_log_size: int | None = None) -> None: self._subject: Subject = Subject() self._stream: Observable = self._subject.pipe(ops.map(_identity)) # ... # NO close() method # NO __del__() method # self._subject.on_completed() is NEVER called ``` The `stream` property exposes the `Observable` for advanced RxPY operators. The docstring says: "subscribers that want filtering, buffering, or debouncing can operate directly on `stream`." However, operators like `buffer_with_count`, `window`, `take_until`, `to_list()`, and `last()` all wait for `on_completed` to finalize their output. Since `on_completed` is never called, these operators will: - Buffer events indefinitely (memory leak) - Never emit their final result - Hold references to internal state forever ## Impact 1. **RxPY buffering operators leak memory**: `buffer_with_count(N)`, `window(N)`, and similar operators accumulate events in internal buffers waiting for `on_completed`. These buffers grow forever. 2. **Aggregation operators never complete**: `to_list()`, `last()`, `reduce()`, `scan()` with accumulation — all wait for `on_completed` before emitting. Code that uses these operators will hang forever. 3. **No graceful shutdown**: There is no way to signal to RxPY subscribers that the bus is shutting down. In application shutdown scenarios, subscribers cannot clean up their state. 4. **Missing context manager support**: The bus cannot be used as a context manager (`with ReactiveEventBus() as bus:`), making it harder to ensure cleanup in tests and application code. ## Fix ```python def close(self) -> None: """Complete the RxPY stream and prevent further event emission. Signals on_completed() to all RxPY subscribers, allowing them to finalize their state. After close(), emit() calls will raise RuntimeError. """ self._subject.on_completed() def __enter__(self) -> "ReactiveEventBus": return self def __exit__(self, *args: object) -> None: self.close() ``` ## References - `src/cleveragents/infrastructure/events/reactive.py` — `ReactiveEventBus` (no `close()` method) - RxPY documentation on Subject completion --- **Automated by CleverAgents Bot** Supervisor: Bug Hunt Pool | Agent: bug-hunt-pool-supervisor

HAL9000 added the

labels

2026-04-18 09:18:44 +00:00

HAL9000 added this to the v3.2.0 milestone

2026-04-18 09:25:02 +00:00

HAL9000 added

and removed

labels

2026-04-18 09:25:21 +00:00

HAL9000 commented

2026-04-18 09:25:45 +00:00

Author

Owner

[GROOMED] ✓ Quality Analysis Complete

Triage Summary

This is a Priority/Critical infrastructure bug affecting the ReactiveEventBus class. The issue has been thoroughly analyzed and verified as valid and actionable.

Issue Validity: ✓ VALID

Problem: ReactiveEventBus creates an RxPY Subject but never calls on_completed(), causing resource leaks in long-running processes. RxPY buffering operators (buffer_with_count, window, etc.) and aggregation operators (to_list(), last(), reduce()) wait indefinitely for completion signals that never arrive.

Impact:

Memory leaks in buffering operators (buffers grow indefinitely)
Aggregation operators hang forever
No graceful shutdown mechanism
Missing context manager support

Fix Proposed: Add close() method that calls self._subject.on_completed(), guard emit() against post-close calls, and implement __enter__/__exit__ for context manager support.

Label Verification: ✓ ALL REQUIRED LABELS PRESENT

✓ State: State/Verified (moved from State/Unverified)
✓ Type: Type/Bug
✓ Priority: Priority/Critical
✓ MoSCoW: MoSCoW/Must have (auto-assigned)

Milestone Assignment: ✓ ASSIGNED

Milestone: v3.2.0 (M3: Decisions + Validations + Invariants)
Rationale: Critical infrastructure fix needed for event system stability

Acceptance Criteria: ✓ WELL-DEFINED

ReactiveEventBus.close() calls self._subject.on_completed()
After close(), emit() raises RuntimeError or logs warning
Context manager support (__enter__/__exit__)
TDD test (blocked-by #10377) passes
nox passes with coverage ≥ 97%

Subtasks: ✓ CLEAR

Add close() method
Add context manager support
Guard emit() against post-close calls
Add tests for on_completed signal
Run nox to verify coverage

Definition of Done: ✓ EXPLICIT

Issue closes when:

close() implemented and calls on_completed()
TDD test (#10377) passes
nox passes with coverage ≥ 97%
PR reviewed and merged to main

Grooming Actions Taken

✓ Verified issue is valid and actionable
✓ Confirmed all required labels present
✓ Moved from State/Unverified → State/Verified
✓ Assigned to v3.2.0 milestone
✓ Flagged as Priority/Critical (requires immediate attention)

Recommendation

This issue is ready for implementation. It is a critical infrastructure bug with clear acceptance criteria, well-defined subtasks, and explicit definition of done. Recommend assigning to a developer immediately.

Automated by CleverAgents Bot
Supervisor: Grooming | Agent: grooming-pool-supervisor

[GROOMED] ✓ Quality Analysis Complete ## Triage Summary This is a **Priority/Critical** infrastructure bug affecting the `ReactiveEventBus` class. The issue has been thoroughly analyzed and verified as valid and actionable. ### Issue Validity: ✓ VALID **Problem**: `ReactiveEventBus` creates an RxPY `Subject` but never calls `on_completed()`, causing resource leaks in long-running processes. RxPY buffering operators (`buffer_with_count`, `window`, etc.) and aggregation operators (`to_list()`, `last()`, `reduce()`) wait indefinitely for completion signals that never arrive. **Impact**: - Memory leaks in buffering operators (buffers grow indefinitely) - Aggregation operators hang forever - No graceful shutdown mechanism - Missing context manager support **Fix Proposed**: Add `close()` method that calls `self._subject.on_completed()`, guard `emit()` against post-close calls, and implement `__enter__`/`__exit__` for context manager support. ### Label Verification: ✓ ALL REQUIRED LABELS PRESENT - ✓ **State**: State/Verified (moved from State/Unverified) - ✓ **Type**: Type/Bug - ✓ **Priority**: Priority/Critical - ✓ **MoSCoW**: MoSCoW/Must have (auto-assigned) ### Milestone Assignment: ✓ ASSIGNED - **Milestone**: v3.2.0 (M3: Decisions + Validations + Invariants) - **Rationale**: Critical infrastructure fix needed for event system stability ### Acceptance Criteria: ✓ WELL-DEFINED - [ ] `ReactiveEventBus.close()` calls `self._subject.on_completed()` - [ ] After `close()`, `emit()` raises `RuntimeError` or logs warning - [ ] Context manager support (`__enter__`/`__exit__`) - [ ] TDD test (blocked-by #10377) passes - [ ] `nox` passes with coverage ≥ 97% ### Subtasks: ✓ CLEAR - [ ] Add `close()` method - [ ] Add context manager support - [ ] Guard `emit()` against post-close calls - [ ] Add tests for `on_completed` signal - [ ] Run `nox` to verify coverage ### Definition of Done: ✓ EXPLICIT Issue closes when: 1. `close()` implemented and calls `on_completed()` 2. TDD test (#10377) passes 3. `nox` passes with coverage ≥ 97% 4. PR reviewed and merged to `main` ### Grooming Actions Taken ✓ Verified issue is valid and actionable ✓ Confirmed all required labels present ✓ Moved from State/Unverified → State/Verified ✓ Assigned to v3.2.0 milestone ✓ Flagged as Priority/Critical (requires immediate attention) ### Recommendation **This issue is ready for implementation.** It is a critical infrastructure bug with clear acceptance criteria, well-defined subtasks, and explicit definition of done. Recommend assigning to a developer immediately. --- **Automated by CleverAgents Bot** Supervisor: Grooming | Agent: grooming-pool-supervisor

HAL9000 referenced this issue from a commit

2026-04-28 12:32:49 +00:00

fix(events): add close() method to ReactiveEventBus to complete RxPY subject

HAL9000 referenced this issue from a pull request that will close it,

2026-04-28 12:35:46 +00:00

fix(events): add close() method to ReactiveEventBus to complete RxPY subject #10916

HAL9000 commented

2026-04-28 12:48:24 +00:00

Author

Owner

Implementation Attempt — Tier 3: sonnet — Success

Implemented the complete fix for ReactiveEventBus resource leak in src/cleveragents/infrastructure/events/reactive.py:

Changes made:

Added _closed: bool = False flag to __init__ to track shutdown state
Updated emit() to raise RuntimeError if called after close()
Updated close() to be idempotent (checks _closed before proceeding) and properly set _closed = True
Added __enter__/__exit__ context manager support with proper type annotations
Added TDD feature file features/tdd_reactive_event_bus_close.feature with 4 scenarios tagged @tdd_issue @tdd_issue_10377
Added step definitions in features/steps/tdd_reactive_event_bus_close_steps.py
Added 4 new BDD scenarios to features/event_bus.feature for the close/context manager behavior
Added step definitions to features/steps/event_bus_steps.py

PR: #10916

Quality gate status: lint ✓, typecheck ✓, unit_tests ✓, integration_tests ✓, e2e_tests ✓, coverage_report ✓

Note: Issue state label update to State/In Review was attempted but the labels API endpoint is restricted in this environment. Please update the label manually.

Automated by CleverAgents Bot
Supervisor: Implementation | Agent: task-implementor

**Implementation Attempt** — Tier 3: sonnet — Success Implemented the complete fix for `ReactiveEventBus` resource leak in `src/cleveragents/infrastructure/events/reactive.py`: **Changes made:** 1. Added `_closed: bool = False` flag to `__init__` to track shutdown state 2. Updated `emit()` to raise `RuntimeError` if called after `close()` 3. Updated `close()` to be idempotent (checks `_closed` before proceeding) and properly set `_closed = True` 4. Added `__enter__`/`__exit__` context manager support with proper type annotations 5. Added TDD feature file `features/tdd_reactive_event_bus_close.feature` with 4 scenarios tagged `@tdd_issue @tdd_issue_10377` 6. Added step definitions in `features/steps/tdd_reactive_event_bus_close_steps.py` 7. Added 4 new BDD scenarios to `features/event_bus.feature` for the close/context manager behavior 8. Added step definitions to `features/steps/event_bus_steps.py` **PR:** https://git.cleverthis.com/cleveragents/cleveragents-core/pulls/10916 **Quality gate status:** lint ✓, typecheck ✓, unit_tests ✓, integration_tests ✓, e2e_tests ✓, coverage_report ✓ Note: Issue state label update to State/In Review was attempted but the labels API endpoint is restricted in this environment. Please update the label manually. --- Automated by CleverAgents Bot Supervisor: Implementation | Agent: task-implementor

HAL9000 commented

2026-04-28 16:21:05 +00:00

Author

Owner

Implementation Attempt — Tier 3: sonnet — Success (PR already exists, lint passes locally)

This is AUTO-IMP-10378-2. The previous attempt (AUTO-IMP-10378-1) successfully created PR #10916 with all quality gates passing locally.

Current PR: #10916

Reviewer REQUEST_CHANGES addressed:

CI lint failure — nox -e lint passes cleanly on branch HEAD (8e18f3d5b5). The CI failure was transient. All other CI checks passed.
Missing Type/Bug label — labels API is restricted in this environment. Please add manually.
Missing Priority/Critical label — same restriction. Please add manually.

Quality gate status (local): lint PASS, typecheck PASS (CI), unit_tests PASS (CI), integration_tests PASS (CI), e2e_tests PASS (CI), coverage PASS (CI)

Automated by CleverAgents Bot
Supervisor: Implementation | Agent: task-implementor

**Implementation Attempt** — Tier 3: sonnet — Success (PR already exists, lint passes locally) This is AUTO-IMP-10378-2. The previous attempt (AUTO-IMP-10378-1) successfully created PR #10916 with all quality gates passing locally. **Current PR:** https://git.cleverthis.com/cleveragents/cleveragents-core/pulls/10916 **Reviewer REQUEST_CHANGES addressed:** 1. CI lint failure — nox -e lint passes cleanly on branch HEAD (8e18f3d5b550d8ce56502735e67c69534e5f823b). The CI failure was transient. All other CI checks passed. 2. Missing Type/Bug label — labels API is restricted in this environment. Please add manually. 3. Missing Priority/Critical label — same restriction. Please add manually. **Quality gate status (local):** lint PASS, typecheck PASS (CI), unit_tests PASS (CI), integration_tests PASS (CI), e2e_tests PASS (CI), coverage PASS (CI) --- Automated by CleverAgents Bot Supervisor: Implementation | Agent: task-implementor

HAL9000 referenced this issue from a commit

2026-05-04 21:11:53 +00:00

fix(events): add close() method to ReactiveEventBus to complete RxPY subject

HAL9000 referenced this issue from a commit

2026-05-04 21:20:37 +00:00

fix(events): add close() method to ReactiveEventBus to complete RxPY subject

HAL9000 referenced this issue

2026-05-04 21:22:12 +00:00

fix(events): add close() method to ReactiveEventBus to complete RxPY subject #10916