fix replay leak

[paul-nechifor]

• The leak was only affecting replay.
• schedule_relative returns a disposable, but the disposable is for cancelling scheduled action, not for cleaning up the resources. The code was adding the disposable to disp which was cleaned up at the end. The issue was that, by storing the cancelation Disposable, it actually holds on to the data.

[chatgpt-codex-connector]

Codex usage limits have been reached for code reviews. Please check with the admins of this repo to increase the limits by adding credits.
Credits must be used to enable repository wide code reviews.

[greptile-apps]

Greptile Summary

Fixed memory leak in TimedSensorReplay.stream() by removing unnecessary disposable tracking. The schedule_relative method returns a disposable for cancelling scheduled actions, but storing it in the CompositeDisposable was keeping references to all scheduled data, preventing garbage collection. Since the code already has proper cleanup via the is_disposed flag and custom dispose function, the returned disposable doesn't need to be tracked.

Confidence Score: 5/5

• This PR is safe to merge with minimal risk
• The fix correctly addresses the memory leak by removing unnecessary disposable tracking. The existing cleanup mechanism (is_disposed flag + custom dispose function) properly handles resource cleanup. The change is minimal, well-explained, and doesn't affect functionality.
• No files require special attention

Important Files Changed

Filename	Overview
dimos/utils/testing/replay.py	Removed unnecessary disposable tracking from schedule_relative to fix memory leak

Sequence Diagram

sequenceDiagram
    participant Client
    participant Observable
    participant Scheduler
    participant Iterator
    
    Client->>Observable: subscribe()
    Observable->>Iterator: iterate_ts()
    Iterator-->>Observable: first_ts, first_data
    Observable->>Client: on_next(first_data)
    
    Iterator-->>Observable: next_message
    
    loop For each message
        Observable->>Scheduler: schedule_relative(delay, emit)
        Note over Scheduler: Schedule emission at target time
        Scheduler-->>Observable: Disposable (for cancellation)
        Note over Observable: OLD: disp.add(disposable) - caused leak<br/>NEW: Return value ignored
        
        Scheduler->>Observable: emit() callback
        Observable->>Client: on_next(data)
        Observable->>Iterator: next()
        alt More messages
            Iterator-->>Observable: next_message
        else No more messages
            Iterator-->>Observable: None
            Observable->>Client: on_completed()
            Observable->>Scheduler: dispose()
        end
    end
    
    Client->>Observable: dispose()
    Note over Observable: is_disposed = True
    Observable->>Scheduler: dispose()

Loading