feat(funnel): add ObservationFunnel read-side listener mechanism - #131
Conversation
Commands already fail over between transports; reads do not, so a field bound to a single source goes unavailable whenever that source is down. StreamRouter arbitrates per-field push observations across sources and reports transport loss as availability rather than as a data value. The router is entirely synchronous and structurally unable to originate a request: no async, polling loop, request callable, or scheduling task. Polling stays with an external consumer, which may gate its own schedule on listen_demand and feed results back through VehicleDataResultPublisher. First vertical slice: Locked, ChargePortDoorOpen, DoorState.TrunkFront, over the existing VehicleBluetooth broadcast and connection-status seams.
![chatgpt-codex-connector[bot]](https://avatars.githubusercontent.com/in/1144995?s=60&v=4){kind=small}
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: 613cd7eb2e
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| previous = state.observations.get(observation.path) | ||
| if previous is not None and previous.observed_at > observation.observed_at: | ||
| return | ||
| state.observations[observation.path] = observation |
There was a problem hiding this comment.
When a source has already reported set_health(source_id, False), this still caches any later observation from it. A final queued BLE broadcast or publisher callback that races after disconnect can therefore be stored while the source is unhealthy; because broadcast capabilities are connection-bound with max_age=None, the next set_health(True) can select and emit that old value even though no healthy source observed it in the recovered session. Please ignore or drop observations from sources while state.healthy is false.
AGENTS.md reference: AGENTS.md:L72-L72
Useful? React with 👍 / 👎.
A frame already in flight can land after its transport loss is recorded. set_health(False) cleared cached observations but publish() kept accepting new ones, so the pre-disconnect reading was re-cached and, because broadcast capabilities are connection-bound and never expire, selected and emitted as a current reading of the recovered session.
Bre77
changed the title
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: e758ab4b9a
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| # displace the one already selected. | ||
| chosen = current | ||
| else: | ||
| better = [c for c in candidates if _rank(c) == best_rank] |
There was a problem hiding this comment.
In a three-source setup, when a low-priority current source is selected and a recovered top-priority source publishes before its failback delay has elapsed, better is built only from that delayed top rank. If a healthy intermediate-priority source also has a fresh observation, filtering better empty falls back to the low-priority current value instead of selecting the next eligible source, so the field can stay on a worse backend until the top source's delay expires and another event happens. Please exclude delayed candidates before choosing the best rank.
AGENTS.md reference: AGENTS.md:L72-L72
Useful? React with 👍 / 👎.
| listeners = self._availability_listeners.setdefault(path, []) | ||
| listeners.append(callback) | ||
| available = self.is_available(path) | ||
| self._announced_availability.setdefault(path, available) |
There was a problem hiding this comment.
When this is called after a field silently aged out of grace, _announced_availability may still be True even though the recomputed available is now False. Because this line only setdefaults, the new listener gets its initial False but the cache remains True, so the next fresh observation is suppressed as “already true” and that listener never receives the recovery. Assign the recomputed value here before dispatching the initial callback.
AGENTS.md reference: AGENTS.md:L72-L72
Useful? React with 👍 / 👎.
…ility A source held by its failback delay was ranked before being filtered out, so an eligible middle-priority source could not displace a worse one while the top source waited. Availability announced to a late listener was recorded with setdefault, leaving a stale True that suppressed the next real recovery for that listener.
There was a problem hiding this comment.
💡 Codex Review
Here are some automated review suggestions for this pull request.
Reviewed commit: d96fc8e3fa
ℹ️ About Codex in GitHub
Your team has set up Codex to review pull requests in this repo. Reviews are triggered when you
- Open a pull request for review
- Mark a draft as ready
- Comment "@codex review".
If Codex has suggestions, it will comment; otherwise it will react with 👍.
Codex can also answer questions or update the PR. Try commenting "@codex address that feedback".
| return True | ||
| if path not in self._values: | ||
| return False | ||
| return self._clock() - self._observed_at[path] <= self._grace |
There was a problem hiding this comment.
For connection-bound/on-change sources such as BLE broadcasts, an unchanged value can be hours old while the source is still healthy; when that only source then reports set_health(False), _reselect() clears the candidate and this line measures the grace window from the original observation timestamp instead of from the source-loss time. In that common case is_available() flips false immediately on disconnect, so the field disappears even though the router is meant to keep the last-known value available for the grace period after the source drops. Track the dropout/last-candidate-loss time separately from observed_at for the grace check.
AGENTS.md reference: AGENTS.md:L72-L72
Useful? React with 👍 / 👎.
An on-change source only speaks when the value changes, so a healthy BLE broadcast source is legitimately hours behind. Measuring the grace window from the chosen observation's own timestamp meant that when such a source dropped, the window was already spent and the field went unavailable instantly - defeating grace for exactly the source type it exists to protect. Track when the last candidate for a path was lost and measure grace from there, including the case where an age-bounded source expires with no event to fire on. Two further findings from a sweep of the same surface: - A value callback that publishes re-entrantly left every listener the outer dispatch had not yet reached holding the superseded value, permanently disagreeing with the router's own value(). - Releasing one availability registration twice dropped a second registration of the same callback.
…patch reentrancy and demand double-unsubscribe"}
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
…Funnel Every publisher feeds the same per-field listeners; nothing selects between sources. Source health, availability, grace windows, failback delay, priority, stickiness and all per-field arbitration are deleted: unavailability is a value a source reports, never something inferred from a link dropping. The surviving logic is hard-coded and not configurable: ignore an observation older than the last one for that field, and do not re-dispatch an unchanged value.
|
You have reached your Codex usage limits for code reviews. You can see your limits in the Codex usage dashboard. |
Bre77
changed the title
1 participant
Intent
On a live Home Assistant install, the lock, charge port and front trunk entities go unavailable: firmware 6.1.1 rerouted those three fields to Bluetooth broadcasts, and when Bluetooth cannot connect they have no other source. Commands already fail over between backends; reads do not.
This adds the read-side counterpart as
ObservationFunnel(tesla_fleet_api/funnel.py). It is a funnel, not a router: every attached publisher — a stream, BLE broadcasts, a suppliedvehicle_dataresult — feeds the same per-field listeners. Nothing selects between sources, so a field keeps its value while any source can still report it.Two rules follow from that, and they are the ones worth reviewing:
The funnel can never originate a request — no polling loop, request callable, HTTP/BLE read, scheduling task, or
asyncanywhere in the module.tests/test_funnel.py::TestFunnelCannotOriginateWorkenforces that against the module's own AST. Polling belongs entirely to an external consumer, which can gate its own schedule onlisten_demand()and feed results back throughVehicleDataResultPublisher.publish_result(dict)— a publisher that holds no client, session, or callable able to obtain one.Scope is deliberately three fields (
Locked,ChargePortDoorOpen,DoorState.TrunkFront) with positive-allowlist translations: an unmapped VCSEC enum or an absent JSON leaf emits no observation rather than a guess. The stream publisher and the Home Assistant wiring are follow-up work in their own repositories.On the earlier revision of this PR
An earlier revision of this branch built a selecting router: source health, availability listeners, grace windows, failback delay, priority ranking, stickiness, and per-field arbitration. That was the wrong shape, and all of it is deleted here along with its tests. Worth stating plainly: all six bugs found in review of this PR were in that machinery, and none of it survives.
Capability/Delivery/Fidelitycollapsed into a plainpaths: frozenset[FieldPath]on the publisher contract, since every field onCapabilityexisted only to feed the selection that is now gone.Testing
uv run pytest tests(707 pass),uv run ruff check,uv run pyright tesla_fleet_api— all green. Existing command-Routertests are untouched and pass unchanged.The load-bearing cases are proven rather than asserted: both sources reaching one listener across a source going away and coming back, with no transient
Noneanywhere in the emitted timeline; BLE broadcasts driven through the realVehicleBluetooth._on_messagerouting path rather than a stand-in; and the supplied-result publisher exercised with literal dictionaries as its only possible data source, including a tripwire mapping that fails if anything callable on it is touched.Pipeline
Updates from git push no-mistakes
✅ **intent** - passed
✅ No issues found.
✅ **Rebase** - passed
✅ No issues found.
✅ **Review** - passed
✅ No issues found.
✅ **Test** - passed
✅ No issues found.
uv run pytest tests/test_funnel.py tests/test_funnel_bluetooth.py tests/test_funnel_vehicle_data.py tests/test_router.py -v (89 passed)grep sweep for deleted-machinery leftovers (StreamRouter, PRIORITY_*, listen_availability, is_available, set_health, failback, PaidPollingPolicy, CostClass, FREE_ONLY) — none foundmanual end-to-end script (/tmp/no-mistakes-evidence/01M0MK0GYQBKKVNZGGJEFWAQTZ/funnel_demo.py) reproducing the maintainer's reported HA bug: real VehicleBluetooth broadcast -> BleBroadcastPublisher -> funnel listener, then BLE detach (no observation emitted), then VehicleDataResultPublisher.publish_result feeding the same listener, then an explicit null reading treated as a real unavailable value✅ **Document** - passed
✅ No issues found.
✅ **Lint** - passed
✅ No issues found.
✅ **Push** - passed
✅ No issues found.