Enable TF service to wait for future transforms

[arkluc]

Problem

Closes #2212

Solution

Add a forward_tolerance param to TF service get functions. Default value is 0, keeps old behaviour.

When the lookup cannot find a transform and forward_tolerance > 0, it re-checks under a threading.Condition lock and waits until the deadline.

How to Test

uv run pytest dimos/protocol/tf/test_tf.py

Contributor License Agreement

• I have read and approved the CLA.

[codecov]

Codecov Report

❌ Patch coverage is 95.52239% with 6 lines in your changes missing coverage. Please review.
✅ All tests successful. No failed tests found.

Files with missing lines	Patch %	Lines
dimos/protocol/tf/tf.py	93.10%	6 Missing ⚠️

📢 Thoughts on this report? Let us know!

[leshy]

dimos --dtop --replay --replay-db hk_village4 run unitree-go2-markers

verified apriltag transforms visible in replay now

[greptile-apps]

Greptile Summary

This PR adds a forward_tolerance parameter to MultiTBuffer.get (and the PubSubTF overrides) that lets callers block until a matching transform arrives or a deadline elapses, implemented via a threading.Condition. It also fixes all previously-identified unprotected iterations of self.buffers (get_frames, get_connections, get_transform_search, publish_all, __str__, graph) by consistently holding _cv.

• receive_transform now mutates self.buffers under _cv and calls notify_all(), while the new _wait_get method loops on _cv.wait() until a result is available or the deadline is reached.
• The re-entrant nesting (_wait_get → _get → get_transform / get_connections, all doing with self._cv:) is safe because threading.Condition() defaults to an RLock as its underlying lock.
• Four new tests cover the happy path, timeout path, already-available fast path, and multi-hop chain completion.

Confidence Score: 4/5

Safe to merge; the wait/notify logic is correct and all previously-flagged buffer-iteration races are resolved.

The core threading design is sound — re-entrant Condition/RLock nesting is handled correctly, spurious wakeups are covered by the while loop, and the race window between the initial _get and _wait_get acquiring the lock is closed by the first iteration of the loop rechecking before waiting. The one remaining issue is that logger.warning fires unconditionally on a forward_tolerance timeout, which could produce log noise for callers using forward_tolerance as a finite retry window.

dimos/protocol/tf/tf.py — the warning emission on timeout is the only outstanding concern.

Important Files Changed

Filename	Overview
dimos/protocol/tf/tf.py	Adds threading.Condition to MultiTBuffer for safe concurrent access; introduces _get / _wait_get split for forward-tolerance waiting. All previously-flagged unprotected buffer iterations are now correctly guarded. The re-entrant lock nesting is safe because Condition() defaults to an RLock. Minor: warning log fires on forward_tolerance timeout.
dimos/protocol/tf/test_tf.py	Adds four new tests covering: returns when buffer fills during wait, timeout path, fast path when data already present, and chain-completion wake. Good coverage of the new forward_tolerance logic.

Sequence Diagram

sequenceDiagram
    participant Caller
    participant MultiTBuffer
    participant _cv as threading.Condition (_cv)
    participant Publisher

    Caller->>MultiTBuffer: "get(parent, child, forward_tolerance=T)"
    MultiTBuffer->>_cv: acquire (via _get)
    _cv-->>MultiTBuffer: locked
    MultiTBuffer->>MultiTBuffer: get_transform / get_transform_search → None
    _cv-->>MultiTBuffer: release
    Note over MultiTBuffer: result is None, forward_tolerance > 0
    MultiTBuffer->>_cv: acquire (via _wait_get)
    _cv-->>MultiTBuffer: locked

    loop until result or deadline
        MultiTBuffer->>MultiTBuffer: _get() [re-entrant RLock]
        alt transform not yet available
            MultiTBuffer->>_cv: "wait(timeout=remaining)"
            _cv-->>MultiTBuffer: releases lock
            Publisher->>_cv: acquire (via receive_transform)
            Publisher->>MultiTBuffer: buffers[key].add(transform)
            Publisher->>_cv: notify_all()
            _cv-->>Publisher: release
            _cv-->>MultiTBuffer: re-acquire (woken)
        else transform available
            MultiTBuffer-->>Caller: Transform result
        end
    end

    alt deadline expired
        MultiTBuffer-->>Caller: None + logger.warning
    end

Loading

_{Reviews (7): Last reviewed commit: "[autofix.ci] apply automated fixes" | Re-trigger Greptile}